Coverage for astrocyte/_astrocyte.py: 87%
918 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-10-29 22:10 +0000
1"""Core Astrocyte class — the main entry point for the framework.
3Handles tier routing, policy enforcement, capability negotiation,
4multi-bank orchestration, and hook dispatch.
5"""
7from __future__ import annotations
9import inspect
10import logging
11from datetime import datetime, timedelta, timezone
12from pathlib import Path
13from typing import TYPE_CHECKING, Any, Literal
15from astrocyte._hooks import HookHandler, HookManager
16from astrocyte._log_safety import safe as _safe_log
17from astrocyte._multi_bank import MultiBankOrchestrator
18from astrocyte._output_scanner import OutputScanner
19from astrocyte._policy import NoisyBankVerdict, PolicyEnforcer
20from astrocyte._provider_dispatch import ProviderDispatcher
21from astrocyte._recall_params import RecallParams
22from astrocyte._validation import validate_bank_id
23from astrocyte.analytics import BankMetricsCollector, compute_bank_health, counters_to_quality_point
24from astrocyte.config import AstrocyteConfig, load_config
25from astrocyte.errors import (
26 AccessDenied,
27 CapabilityNotSupported,
28 ConfigError,
29 MipRoutingError,
30 PiiRejected,
31 ProviderUnavailable,
32 RateLimited,
33)
34from astrocyte.identity import context_principal_label
35from astrocyte.lifecycle import LifecycleManager
36from astrocyte.mip.router import MipRouter
37from astrocyte.policy.observability import MetricsCollector, StructuredLogger, span, timed
38from astrocyte.recall.authority import apply_recall_authority
39from astrocyte.types import (
40 AccessGrant,
41 AstrocyteContext,
42 AuditResult,
43 BankHealth,
44 CompileResult,
45 Entity,
46 ForgetRequest,
47 ForgetResult,
48 GraphHit,
49 HealthStatus,
50 HistoryResult,
51 LegalHold,
52 LifecycleRunResult,
53 MemoryChangePage,
54 MemoryHit,
55 MentalModel,
56 MultiBankStrategy,
57 QualityDataPoint,
58 RecallRequest,
59 RecallResult,
60 ReflectRequest,
61 ReflectResult,
62 RetainRequest,
63 RetainResult,
64 SyncPushRecord,
65 SyncPushResult,
66)
68if TYPE_CHECKING:
69 from astrocyte.pipeline.orchestrator import PipelineOrchestrator
70 from astrocyte.provider import EngineProvider
72logger = logging.getLogger("astrocyte")
75def _normalize_multi_bank_strategy(
76 strategy: Literal["cascade", "parallel", "first_match"] | MultiBankStrategy | None,
77) -> MultiBankStrategy:
78 if strategy is None:
79 return MultiBankStrategy(mode="parallel")
80 if isinstance(strategy, MultiBankStrategy):
81 return strategy
82 if strategy in ("cascade", "parallel", "first_match"):
83 return MultiBankStrategy(mode=strategy)
84 raise ConfigError(f"Unknown multi-bank strategy: {strategy!r}")
87class Astrocyte:
88 """The Astrocyte memory framework — unified API for AI agent memory.
90 Usage:
91 brain = Astrocyte.from_config("astrocyte.yaml")
92 await brain.retain("Calvin prefers dark mode", bank_id="user-123")
93 hits = await brain.recall("What are Calvin's preferences?", bank_id="user-123")
94 """
96 def __init__(self, config: AstrocyteConfig) -> None:
97 self._config = config
98 self._logger = StructuredLogger(level=config.observability.log_level)
100 # Extracted subsystems
101 self._policy = PolicyEnforcer(config)
102 self._hook_manager = HookManager(self._logger)
103 self._output_scanner = OutputScanner(config, self._logger)
104 self._dispatcher = ProviderDispatcher(config)
106 # Metrics
107 self._metrics = MetricsCollector(enabled=config.observability.prometheus_enabled)
109 # Lifecycle
110 self._lifecycle = LifecycleManager(config.lifecycle)
112 # MIP router (loaded from mip.yaml if configured)
113 self._mip_router: MipRouter | None = None
114 if config.mip_config_path:
115 from astrocyte.mip.loader import load_mip_config
117 mip_config = load_mip_config(config.mip_config_path)
118 self._mip_router = MipRouter(mip_config)
120 # Analytics
121 self._analytics = BankMetricsCollector()
123 # Provider state (managed by dispatcher, exposed for wiring)
124 self._engine_provider: EngineProvider | None = None
125 self._pipeline: PipelineOrchestrator | None = None
127 # Multi-bank orchestrator (wired lazily after provider is set)
128 self._multi_bank: MultiBankOrchestrator | None = None
130 # M8: wiki store (optional; enables brain.compile())
131 self._wiki_store: object | None = None
132 # M8 W4: async compile queue (optional; enables automatic threshold triggering)
133 self._compile_queue: object | None = None
134 # M9: mental-model store (optional; enables /v1/mental-models endpoints).
135 # First-class replacement for the prior wiki-piggyback (kind="concept" +
136 # metadata["_mental_model"] = True) — see provider.MentalModelStore docs.
137 self._mental_model_store: object | None = None
138 # M10: source-document / chunk store (optional; enables source-aware
139 # retain to preserve provenance back to the originating document and
140 # chunk). Deployments without a SourceStore work exactly as before —
141 # vectors stay anonymous flat rows.
142 self._source_store: object | None = None
144 @property
145 def config(self) -> AstrocyteConfig:
146 """Loaded :class:`~astrocyte.config.AstrocyteConfig` (read-only for callers)."""
147 return self._config
149 async def __aenter__(self) -> "Astrocyte":
150 await self.start_background_tasks()
151 return self
153 async def __aexit__(self, *exc: object) -> None:
154 await self.stop_background_tasks()
156 async def start_background_tasks(self) -> None:
157 """Start optional background services configured on this instance.
159 Today this covers the M8 compile queue. Keeping the lifecycle on
160 ``Astrocyte`` lets library users and the REST gateway start the same
161 services without reaching into private attributes.
162 """
163 queue = self._compile_queue
164 if queue is not None and hasattr(queue, "start"):
165 await queue.start() # type: ignore[union-attr]
167 async def stop_background_tasks(self) -> None:
168 """Stop optional background services configured on this instance."""
169 queue = self._compile_queue
170 if queue is not None and hasattr(queue, "stop"):
171 await queue.stop() # type: ignore[union-attr]
172 shutdown = getattr(self._pipeline, "shutdown", None)
173 if shutdown is not None:
174 result = shutdown()
175 if inspect.isawaitable(result):
176 _ = await result
178 @classmethod
179 def from_config(cls, path: str | Path) -> "Astrocyte":
180 """Create a ready-to-use Astrocyte from a YAML config file.
182 When the config names a ``vector_store``, the storage pipeline and
183 every configured store are wired, so ``retain``/``recall`` work
184 immediately. (Previously this returned an unwired instance whose first
185 ``retain`` raised "No provider or pipeline configured" — the documented
186 hello-world did not run.) Configs without a vector store are returned
187 unwired, for callers that attach an engine or pipeline themselves.
188 """
189 return cls._from_loaded_config(load_config(path))
191 @classmethod
192 def from_config_dict(cls, data: dict[str, str | int | float | bool | None | dict | list]) -> "Astrocyte":
193 """Create an Astrocyte from a config dictionary; wired like :meth:`from_config`."""
194 from astrocyte.config import _dict_to_config, validate_astrocyte_config
196 config = _dict_to_config(data)
197 validate_astrocyte_config(config)
198 return cls._from_loaded_config(config)
200 @classmethod
201 def _from_loaded_config(cls, config: AstrocyteConfig) -> "Astrocyte":
202 from astrocyte.wiring import wants_storage_pipeline, wire_astrocyte
204 brain = cls(config)
205 return wire_astrocyte(brain, config) if wants_storage_pipeline(config) else brain
207 def set_engine_provider(self, provider: EngineProvider) -> None:
208 """Set the engine provider (for programmatic setup)."""
209 from astrocyte.provider import check_spi_version
211 check_spi_version(provider, "EngineProvider")
212 self._engine_provider = provider
213 self._dispatcher.engine_provider = provider
214 if hasattr(provider, "capabilities"):
215 self._dispatcher.capabilities = provider.capabilities()
216 self._rebuild_tiered_retrieval()
217 self._rebuild_multi_bank()
219 def set_wiki_store(self, wiki_store: object) -> None:
220 """Set the WikiStore provider (M8 wiki compile). Optional.
222 When a WikiStore is configured, ``brain.compile()`` becomes available.
223 Compile is disabled by default — banks that don't opt in keep today's
224 retain/recall behaviour exactly.
226 Args:
227 wiki_store: Any object satisfying the
228 :class:`~astrocyte.provider.WikiStore` protocol.
229 """
230 from astrocyte.provider import check_spi_version
232 check_spi_version(wiki_store, "WikiStore")
233 self._wiki_store = wiki_store
235 def set_mental_model_store(self, store: object) -> None:
236 """Set the :class:`~astrocyte.provider.MentalModelStore` provider. Optional.
238 When configured, the gateway's ``/v1/mental-models`` endpoints
239 (and any in-process consumer) route reads/writes through this
240 store instead of erroring with HTTP 501. Cuts the prior
241 wiki-piggyback path (``kind="concept"`` + metadata discriminator)
242 — mental models now live in their own
243 ``astrocyte_mental_models`` table with proper revision history.
245 Also wires the agentic reflect loop so ``search_mental_models``
246 becomes available as a tool — the highest-quality tier in the
247 hierarchical retrieval order. Idempotent: if a pipeline is
248 already attached, the service is pushed onto it; otherwise
249 ``set_pipeline`` will pick up the store and build the service
250 when it runs.
251 """
252 from astrocyte.provider import check_spi_version
254 check_spi_version(store, "MentalModelStore")
255 self._mental_model_store = store
256 if self._pipeline is not None:
257 from astrocyte.pipeline.mental_model import MentalModelService
259 self._pipeline.mental_model_service = MentalModelService(store)
261 def set_source_store(self, store: object) -> None:
262 """Set the :class:`~astrocyte.provider.SourceStore` provider. Optional.
264 When configured, ingestion paths can persist
265 :class:`~astrocyte.types.SourceDocument` and
266 :class:`~astrocyte.types.SourceChunk` rows alongside vectors so memory
267 retains a backreference to its originating document. Deployments
268 without a SourceStore continue to work — vectors remain anonymous
269 flat rows with ``chunk_id = NULL``.
271 Args:
272 store: Any object satisfying the
273 :class:`~astrocyte.provider.SourceStore` protocol.
274 """
275 from astrocyte.provider import check_spi_version
277 check_spi_version(store, "SourceStore")
278 self._source_store = store
280 def set_compile_queue(self, queue: object) -> None:
281 """Set the async compile queue (M8 W4 threshold trigger). Optional.
283 When a :class:`~astrocyte.pipeline.compile_trigger.CompileQueue` is
284 configured, each successful ``brain.retain()`` call notifies the queue.
285 The queue fires a background compile job whenever the bank crosses the
286 configured size or staleness threshold.
288 The queue must be started (``await queue.start()``) before the first
289 retain call, and stopped (``await queue.stop()``) on shutdown.
291 Args:
292 queue: A :class:`~astrocyte.pipeline.compile_trigger.CompileQueue`
293 instance (or any object with a compatible ``notify_retain``
294 method).
295 """
296 self._compile_queue = queue
298 def use_pageindex_pipeline(
299 self,
300 store: Any,
301 embedding_provider: Any | None = None,
302 *,
303 document_resolver: Any | None = None,
304 ) -> None:
305 """M32 — install the PageIndex recall pipeline as the active
306 retrieval stack.
308 After this call, ``self.recall(query, ...)`` routes through
309 :class:`~astrocyte.pipeline.pageindex_pipeline.PageIndexPipeline`
310 (the same stack the bench harness validates) instead of the
311 legacy ``orchestrator.recall()`` / vector_store path.
313 This closes the bench-vs-API parity gap surfaced in the v0.15.0
314 ship audit (docs/_design/m32-stack-unification.md). All
315 M14-M31 cycle work (RRF fusion, fact↔chunk pairing,
316 per-Q-type prompts, M27 fields, M28-M29 coreference, M30
317 parallelization, M31 session_filter + event_date) now flows
318 through the public API.
320 Args:
321 store: A configured :class:`PageIndexStore` (Postgres or
322 in-memory).
323 embedding_provider: An :class:`LLMProvider` to embed query
324 strings. Defaults to the Astrocyte instance's configured
325 provider when ``None``.
326 document_resolver: Optional ``(bank_id) -> document_id``
327 callable for single-doc scoping. See
328 ``PageIndexPipeline.__init__`` for the contract.
329 """
330 from astrocyte.pipeline.pageindex_pipeline import ( # noqa: PLC0415
331 PageIndexPipeline,
332 )
334 if embedding_provider is None:
335 embedding_provider = getattr(self, "_provider", None)
336 if embedding_provider is None:
337 raise ValueError(
338 "use_pageindex_pipeline: no embedding_provider supplied and "
339 "no provider configured on Astrocyte instance",
340 )
341 self._dispatcher.pageindex_pipeline = PageIndexPipeline(
342 store=store,
343 embedding_provider=embedding_provider,
344 config=self._config,
345 document_resolver=document_resolver,
346 )
348 def set_pipeline(self, pipeline: PipelineOrchestrator) -> None:
349 """Set the Tier 1 pipeline orchestrator (for programmatic setup)."""
350 self._pipeline = pipeline
351 self._dispatcher.pipeline = pipeline
352 # All feature flags are derived once from config (env overrides,
353 # lazy service construction) and applied in one place — see
354 # PipelineConfig.from_config / PipelineOrchestrator.apply_config.
355 from astrocyte.pipeline.pipeline_config import PipelineConfig
357 pipeline.apply_config(
358 PipelineConfig.from_config(
359 self._config,
360 source_store=self._source_store,
361 mental_model_store=self._mental_model_store,
362 )
363 )
364 # Wire the LLM provider to the MIP router for intent-layer escalation.
365 # Intra-package collaborator wiring (both objects are astrocyte-internal).
366 if self._mip_router and hasattr(pipeline, "llm_provider"):
367 self._mip_router._llm_provider = pipeline.llm_provider # noqa: SLF001
368 # Expose router for per-bank MIP resolution at recall time (P3)
369 pipeline.mip_router = self._mip_router
370 self._rebuild_tiered_retrieval()
371 self._rebuild_multi_bank()
373 def _rebuild_tiered_retrieval(self) -> None:
374 """Construct :class:`~astrocyte.pipeline.tiered_retrieval.TieredRetriever` when enabled."""
375 from astrocyte.hybrid import HybridEngineProvider
376 from astrocyte.pipeline.recall_cache import RecallCache
377 from astrocyte.pipeline.recent_buffer import RecentMemoryBuffer
378 from astrocyte.pipeline.tiered_retrieval import TieredRetriever
380 self._dispatcher.tiered_retriever = None
381 if not self._pipeline or not self._config.tiered_retrieval.enabled:
382 return
383 trc = self._config.tiered_retrieval
384 if self._engine_provider is not None and trc.full_recall != "hybrid":
385 return
386 full_recall_fn = None
387 if trc.full_recall == "hybrid":
388 if not isinstance(self._engine_provider, HybridEngineProvider):
389 logger.warning(
390 "tiered_retrieval.full_recall=hybrid requires HybridEngineProvider; "
391 "tiered retrieval disabled until a hybrid provider is set",
392 )
393 return
394 full_recall_fn = self._engine_provider.recall
395 rcc = self._config.recall_cache
396 cache: RecallCache | None = None
397 if rcc.enabled:
398 cache = RecallCache(
399 similarity_threshold=rcc.similarity_threshold,
400 max_entries=rcc.max_entries,
401 ttl_seconds=rcc.ttl_seconds,
402 )
403 recent = RecentMemoryBuffer()
404 self._dispatcher.tiered_retriever = TieredRetriever(
405 self._pipeline,
406 recall_cache=cache,
407 recent_buffer=recent,
408 min_results=trc.min_results,
409 min_score=trc.min_score,
410 max_tier=trc.max_tier,
411 full_recall=full_recall_fn,
412 )
414 def _rebuild_multi_bank(self) -> None:
415 """Rebuild MultiBankOrchestrator when provider changes."""
416 self._multi_bank = MultiBankOrchestrator(
417 do_recall=self._dispatcher.recall,
418 make_request=self._make_recall_request,
419 circuit_breaker_record_failure=self._policy.record_failure,
420 metrics=self._metrics,
421 provider_name=self._dispatcher.provider_name,
422 )
424 def set_access_grants(self, grants: list[AccessGrant]) -> None:
425 """Configure access grants."""
426 self._policy.set_access_grants(grants)
428 def check_access(self, bank_id: str, permission: str, context: AstrocyteContext | None) -> None:
429 """Check whether ``context`` holds ``permission`` on ``bank_id``.
431 Raises :class:`~astrocyte.errors.AccessDenied` when access control is
432 enabled and the principal lacks the permission; returns silently
433 otherwise. Public counterpart of the internal policy check — external
434 callers (e.g. the HTTP gateway guarding endpoints that bypass
435 retain/recall, such as mental-model CRUD) should use this rather than
436 reaching into ``_policy``.
437 """
438 self._policy.check_access(bank_id, permission, context)
440 @property
441 def _rate_limiters(self) -> dict:
442 """Expose rate limiters for testing/introspection."""
443 return self._policy._rate_limiters # noqa: SLF001 — deliberate test/introspection window into the policy collaborator
445 def register_hook(self, event_type: str, handler: HookHandler) -> None:
446 """Register an event hook handler."""
447 self._hook_manager.register(event_type, handler)
449 # ---------------------------------------------------------------------------
450 # Public API
451 # ---------------------------------------------------------------------------
453 async def retain(
454 self,
455 content: str,
456 bank_id: str,
457 *,
458 metadata: dict[str, str | int | float | bool | None] | None = None,
459 tags: list[str] | None = None,
460 context: AstrocyteContext | None = None,
461 content_type: str = "text",
462 extraction_profile: str | None = None,
463 occurred_at: datetime | None = None,
464 source: str | None = None,
465 pii_detected: bool = False,
466 ) -> RetainResult:
467 """Store content into memory.
469 Args:
470 content_type: ``text``, ``conversation``, ``document``, etc.
471 extraction_profile: Name under YAML ``extraction_profiles:``.
472 occurred_at: When the content originally occurred.
473 source: Origin identifier for the content.
474 pii_detected: Whether PII was already detected upstream.
475 """
476 validate_bank_id(bank_id)
477 with span("astrocyte.retain", {"astrocyte.bank_id": bank_id}):
478 # Input size validation — reject oversized content before pipeline processing
479 input_error = self._policy.validate_retain_input(content, tags, bank_id=bank_id)
480 if input_error:
481 return RetainResult(stored=False, error=input_error)
483 # Access control
484 self._policy.check_access(bank_id, "write", context)
486 # Identity spec §3 Gap 2: resolved actor flows into MIP so rules can
487 # branch on principal_type / principal_id / etc. Resolved once here,
488 # unconditionally, because it is also persisted below — the actor is
489 # the only record of *who* wrote a memory, and it was previously
490 # computed for routing and then dropped.
491 from astrocyte.identity import format_principal, resolve_actor
493 actor_identity = resolve_actor(context) if context else None
494 if actor_identity is not None:
495 # Stamped into metadata rather than a column: metadata is JSONB,
496 # so provenance-of-authorship needs no migration. Overwrites a
497 # caller-supplied ``_actor``: with a context present, the
498 # context is who wrote this, and anything else is a forgery.
499 # Context-free callers (imports) keep their own ``_actor``.
500 metadata = dict(metadata or {})
501 metadata["_actor"] = format_principal(actor_identity)
503 # MIP routing (before policy layer)
504 mip_pipeline = None
505 mip_rule_name = None
506 if self._mip_router:
507 from astrocyte.mip.rule_engine import RuleEngineInput
509 mip_input = RuleEngineInput(
510 content=content,
511 content_type=content_type,
512 metadata=metadata,
513 tags=tags,
514 pii_detected=pii_detected,
515 source=source,
516 actor_identity=actor_identity,
517 )
518 routing = await self._mip_router.route(mip_input)
519 if routing.bank_id:
520 # Re-check access for new bank
521 if routing.bank_id != bank_id:
522 self._policy.check_access(routing.bank_id, "write", context)
523 bank_id = routing.bank_id
524 if routing.tags is not None:
525 tags = routing.tags
526 if routing.retain_policy == "reject":
527 return RetainResult(stored=False, error="Rejected by MIP routing rule")
528 mip_pipeline = routing.pipeline
529 mip_rule_name = routing.rule_name
531 # Rate limiting + quota (atomic to prevent TOCTOU)
532 self._policy.check_rate_and_quota(bank_id, "retain")
534 # Noisy-bank detection (signal_quality.noisy_bank)
535 noisy = self._policy.check_noisy_bank(bank_id)
536 if noisy is not None:
537 rejection = self._apply_noisy_bank_verdict(bank_id, noisy)
538 if rejection is not None:
539 return rejection
541 # Content validation
542 errors = self._policy.validate_content(content, content_type, bank_id=bank_id)
543 if errors:
544 return RetainResult(stored=False, error="; ".join(errors))
546 # PII scanning (async for LLM/rules_then_llm modes)
547 content, pii_matches = await self._policy.scan_pii(content, bank_id=bank_id)
548 if pii_matches:
549 await self._note_pii(
550 bank_id, [m.pii_type for m in pii_matches], self._policy.pii_action_for(bank_id), "retain"
551 )
553 # Metadata sanitization
554 metadata, meta_warnings = self._policy.sanitize_metadata(metadata, bank_id=bank_id)
556 # Build request
557 request = RetainRequest(
558 content=content,
559 bank_id=bank_id,
560 metadata=metadata,
561 tags=tags,
562 occurred_at=occurred_at,
563 source=source,
564 content_type=content_type,
565 extraction_profile=extraction_profile,
566 mip_pipeline=mip_pipeline,
567 mip_rule_name=mip_rule_name,
568 )
570 # Route to provider
571 try:
572 self._policy.check_circuit(self._provider_name)
573 with timed() as t:
574 result = await self._dispatcher.retain(request)
575 self._policy.record_success()
576 self._policy.record_quota(bank_id, "retain")
577 self._metrics.inc_counter(
578 "astrocyte_retain_total",
579 {"bank_id": bank_id, "provider": self._provider_name, "status": "ok"},
580 )
581 self._metrics.observe_histogram(
582 "astrocyte_retain_duration_seconds",
583 t["elapsed_ms"] / 1000,
584 {"bank_id": bank_id, "provider": self._provider_name},
585 )
586 self._analytics.record_retain(
587 bank_id,
588 len(content),
589 deduplicated=getattr(result, "deduplicated", False),
590 )
591 self._policy.record_retain_signal(bank_id, len(content), getattr(result, "deduplicated", False))
592 await self._hook_manager.fire(
593 "on_retain",
594 bank_id=bank_id,
595 data={
596 "memory_id": result.memory_id or "",
597 "content_length": len(content),
598 },
599 )
600 # M8 W4: notify the compile queue so it can trigger a background
601 # compile when the bank crosses the configured threshold.
602 if self._compile_queue is not None and result.stored:
603 self._compile_queue.notify_retain(bank_id) # type: ignore[union-attr]
604 return result
605 except ProviderUnavailable:
606 self._policy.handle_degraded_retain(self._provider_name)
607 return RetainResult(stored=False, error="Provider unavailable (degraded mode)")
608 except Exception:
609 self._policy.record_failure()
610 self._metrics.inc_counter(
611 "astrocyte_retain_total",
612 {"bank_id": bank_id, "provider": self._provider_name, "status": "error"},
613 )
614 raise
616 def _apply_noisy_bank_verdict(self, bank_id: str, verdict: NoisyBankVerdict) -> RetainResult | None:
617 """Act on a noisy-bank check: log on a change of state, count every
618 flagged retain, then ``warn`` (proceed), ``throttle`` (raise
619 :class:`RateLimited`, retryable) or ``reject`` (refuse the retain).
620 Returns the refusal for ``reject``, else ``None``."""
621 if verdict.changed:
622 self._logger.log(
623 "astrocyte.signal_quality.noisy_bank"
624 if verdict.reasons
625 else "astrocyte.signal_quality.noisy_bank_cleared",
626 bank_id=bank_id,
627 operation="retain",
628 data={"reasons": ",".join(verdict.reasons), "action": verdict.action},
629 level=logging.WARNING if verdict.reasons else logging.INFO,
630 )
631 if not verdict.reasons:
632 return None
633 self._metrics.inc_counter(
634 "astrocyte_noisy_bank_total",
635 {"bank_id": bank_id, "action": verdict.action},
636 )
637 if verdict.action == "throttle":
638 raise RateLimited(bank_id=bank_id, operation="retain", retry_after_seconds=60.0)
639 if verdict.action == "reject":
640 return RetainResult(stored=False, error=f"Bank flagged as noisy ({', '.join(verdict.reasons)})")
641 return None
643 async def _make_recall_request(
644 self,
645 query: str,
646 bank_id: str,
647 max_results: int,
648 max_tokens: int | None,
649 tags: list[str] | None,
650 params: RecallParams,
651 ) -> RecallRequest:
652 from astrocyte.recall.proxy import merge_manual_and_proxy_hits
654 ext = await merge_manual_and_proxy_hits(
655 self._config,
656 query=query,
657 bank_id=bank_id,
658 manual=params.external_context,
659 metrics=self._metrics,
660 )
661 return RecallRequest(
662 query=query,
663 bank_id=bank_id,
664 max_results=max_results,
665 max_tokens=max_tokens,
666 tags=tags,
667 fact_types=params.fact_types,
668 time_range=params.time_range,
669 include_sources=params.include_sources,
670 layer_weights=params.layer_weights,
671 detail_level=params.detail_level,
672 external_context=ext,
673 as_of=params.as_of, # M9
674 query_reference_date=params.query_reference_date,
675 session_id=params.session_id, # M31 Fix 2
676 )
678 async def recall(
679 self,
680 query: str,
681 bank_id: str | None = None,
682 *,
683 banks: list[str] | None = None,
684 strategy: Literal["cascade", "parallel", "first_match"] | MultiBankStrategy | None = None,
685 max_results: int = 10,
686 max_tokens: int | None = None,
687 tags: list[str] | None = None,
688 context: AstrocyteContext | None = None,
689 external_context: list[MemoryHit] | None = None,
690 fact_types: list[str] | None = None,
691 time_range: tuple[datetime, datetime] | None = None,
692 include_sources: bool = False,
693 layer_weights: dict[str, float] | None = None,
694 detail_level: str | None = None,
695 as_of: datetime | None = None,
696 query_reference_date: datetime | None = None,
697 session_id: str | None = None,
698 ) -> RecallResult:
699 """Retrieve relevant memories for a query.
701 With multiple ``banks``, use ``strategy`` (or a :class:`MultiBankStrategy`) to choose
702 ``parallel`` (default), ``cascade`` (widen until enough hits), or ``first_match``.
704 Args:
705 external_context: Extra :class:`MemoryHit` items merged with retrieval via RRF.
706 fact_types: Filter by fact type (e.g. ``["preference", "event"]``).
707 time_range: ``(start, end)`` datetime tuple to scope retrieval.
708 include_sources: Include source metadata in results.
709 layer_weights: Per-layer scoring weights for tiered retrieval.
710 detail_level: Granularity hint (``"summary"`` / ``"full"``).
712 When ``tiered_retrieval.enabled`` is set and a pipeline is configured, recall uses
713 :class:`~astrocyte.pipeline.tiered_retrieval.TieredRetriever` for cheap tiers.
714 """
715 # Resolve bank(s)
716 bank_ids = self._policy.resolve_read_bank_ids(bank_id, banks, context)
718 max_tokens = max_tokens or self._policy.token_budget(bank_ids, "recall")
720 with span(
721 "astrocyte.recall",
722 {
723 "astrocyte.bank_count": len(bank_ids),
724 "astrocyte.bank_id": bank_ids[0] if len(bank_ids) == 1 else f"{bank_ids[0]}+{len(bank_ids) - 1}",
725 },
726 ):
727 # Access control for all banks
728 for bid in bank_ids:
729 self._policy.check_access(bid, "read", context)
731 # Rate limiting + quota (atomic per-bank)
732 for bid in bank_ids:
733 self._policy.check_rate_and_quota(bid, "recall")
735 # Build typed params for internal helpers
736 _rp = RecallParams(
737 external_context=external_context,
738 fact_types=fact_types,
739 time_range=time_range,
740 include_sources=include_sources,
741 layer_weights=layer_weights,
742 detail_level=detail_level,
743 as_of=as_of, # M9
744 query_reference_date=query_reference_date,
745 session_id=session_id, # M31 Fix 2
746 )
748 # Single bank — direct
749 if len(bank_ids) == 1:
750 request = await self._make_recall_request(
751 query,
752 bank_ids[0],
753 max_results,
754 max_tokens,
755 tags,
756 _rp,
757 )
758 try:
759 self._policy.check_circuit(self._provider_name)
760 with timed() as t:
761 result = await self._dispatcher.recall(request)
762 self._policy.record_success()
763 self._metrics.inc_counter(
764 "astrocyte_recall_total",
765 {"bank_id": bank_ids[0], "provider": self._provider_name, "status": "ok"},
766 )
767 self._metrics.observe_histogram(
768 "astrocyte_recall_duration_seconds",
769 t["elapsed_ms"] / 1000,
770 {"bank_id": bank_ids[0], "provider": self._provider_name},
771 )
772 top_score = result.hits[0].score if result.hits else 0.0
773 self._analytics.record_recall(
774 bank_ids[0],
775 len(result.hits),
776 top_score,
777 )
778 await self._hook_manager.fire(
779 "on_recall",
780 bank_id=bank_ids[0],
781 data={
782 "query_length": len(query),
783 "result_count": len(result.hits),
784 },
785 )
786 if self._config.dlp.scan_recall_output:
787 result = self._output_scanner.scan_recall(result)
788 return apply_recall_authority(result, self._config.recall_authority)
789 except ProviderUnavailable:
790 return self._policy.handle_degraded_recall(self._provider_name)
791 except Exception:
792 self._policy.record_failure()
793 raise
795 strat = _normalize_multi_bank_strategy(strategy)
796 result = await self._multi_bank.recall(
797 query,
798 bank_ids,
799 max_results,
800 max_tokens,
801 tags,
802 _rp,
803 strat,
804 )
805 if self._config.dlp.scan_recall_output:
806 result = self._output_scanner.scan_recall(result)
807 return apply_recall_authority(result, self._config.recall_authority)
809 async def reflect(
810 self,
811 query: str,
812 bank_id: str | None = None,
813 *,
814 banks: list[str] | None = None,
815 strategy: Literal["cascade", "parallel", "first_match"] | MultiBankStrategy | None = None,
816 max_tokens: int | None = None,
817 tags: list[str] | None = None,
818 context: AstrocyteContext | None = None,
819 include_sources: bool = True,
820 dispositions: Any | None = None,
821 as_of: datetime | None = None,
822 query_reference_date: datetime | None = None,
823 ) -> ReflectResult:
824 """Synthesize an answer from memory.
826 Args:
827 tags: Filter recall by tags during reflect.
828 include_sources: Include source memories in the result.
829 dispositions: Emotional/tonal dispositions for synthesis.
830 as_of: M9 time-travel anchor — filters underlying recall to
831 ``retained_at <= as_of``. For legal-hold / audit-replay
832 use cases ("show me what the system knew at time X"). Do
833 NOT pass for relative-phrase anchoring on benchmarks
834 whose retain happens after the dataset's question dates;
835 use ``query_reference_date`` instead.
836 query_reference_date: Reference date for resolving relative
837 temporal phrases (``"yesterday"``, ``"last week"``,
838 ``"3 days ago"``). Required when running a benchmark
839 whose dataset predates the wall-clock — e.g. LongMemEval
840 (2023-vintage) measured in 2026 must pass the question's
841 contemporaneous date here. Unlike ``as_of``, setting
842 this alone does NOT filter out memories retained after
843 this date — the entire corpus stays in scope.
845 Supports multi-bank reflect: pass ``banks`` (and optionally ``strategy``) to
846 recall across multiple banks and synthesize over the fused results.
847 """
848 # Resolve bank(s)
849 bank_ids = self._policy.resolve_read_bank_ids(bank_id, banks, context)
851 max_tokens = max_tokens or self._policy.token_budget(bank_ids, "reflect")
852 primary_bank = bank_ids[0]
854 with span(
855 "astrocyte.reflect",
856 {
857 "astrocyte.bank_count": len(bank_ids),
858 "astrocyte.bank_id": bank_ids[0] if len(bank_ids) == 1 else f"{bank_ids[0]}+{len(bank_ids) - 1}",
859 },
860 ):
861 # Access control for all banks
862 for bid in bank_ids:
863 self._policy.check_access(bid, "read", context)
865 for bid in bank_ids:
866 self._policy.check_rate_and_quota(bid, "reflect")
868 # ── Single bank: delegate to provider/pipeline reflect ──
869 if len(bank_ids) == 1:
870 request = ReflectRequest(
871 query=query,
872 bank_id=primary_bank,
873 max_tokens=max_tokens,
874 include_sources=include_sources,
875 dispositions=dispositions,
876 tags=tags,
877 as_of=as_of,
878 query_reference_date=query_reference_date,
879 )
880 try:
881 self._policy.check_circuit(self._provider_name)
882 with timed() as t:
883 result = await self._dispatcher.reflect(request)
884 self._policy.record_success()
885 except ProviderUnavailable:
886 return ReflectResult(answer="Memory unavailable", sources=[])
887 except Exception:
888 self._policy.record_failure()
889 raise
891 # ── Multi-bank: recall across banks, then synthesize ──
892 else:
893 strat = _normalize_multi_bank_strategy(strategy)
894 with timed() as t:
895 _rp = RecallParams(
896 include_sources=include_sources,
897 as_of=as_of,
898 query_reference_date=query_reference_date,
899 )
900 recall_result = await self._multi_bank.recall(
901 query,
902 bank_ids,
903 max_results=20, # Larger set for synthesis context
904 max_tokens=None, # Budget applied after synthesis
905 tags=tags,
906 params=_rp,
907 strategy=strat,
908 )
909 auth_ctx: str | None = None
910 ra = self._config.recall_authority
911 if ra.enabled and ra.apply_to_reflect:
912 recall_result = apply_recall_authority(recall_result, ra)
913 auth_ctx = recall_result.authority_context
914 result = await self._dispatcher.reflect_from_hits(
915 query=query,
916 hits=recall_result.hits,
917 bank_id=primary_bank,
918 max_tokens=max_tokens,
919 dispositions=dispositions,
920 authority_context=auth_ctx,
921 )
923 self._analytics.record_reflect(
924 primary_bank,
925 success=bool(result.answer.strip()),
926 )
927 self._policy.record_quota(primary_bank, "reflect")
928 self._metrics.inc_counter(
929 "astrocyte_reflect_total",
930 {"bank_id": ",".join(bank_ids), "provider": self._provider_name, "status": "ok"},
931 )
932 self._metrics.observe_histogram(
933 "astrocyte_reflect_duration_seconds",
934 t["elapsed_ms"] / 1000,
935 {"bank_id": ",".join(bank_ids), "provider": self._provider_name},
936 )
937 await self._hook_manager.fire(
938 "on_reflect",
939 bank_id=primary_bank,
940 data={
941 "query_length": len(query),
942 "answer_length": len(result.answer),
943 "bank_count": len(bank_ids),
944 },
945 )
946 if self._config.dlp.scan_reflect_output:
947 result = self._output_scanner.scan_reflect(result)
948 return result
950 # ── M21 — Mental model CRUD (Hindsight parity) ────────────────────
952 def _mental_model_service(self) -> Any:
953 """Return the configured MentalModelService or raise.
955 Mental-model CRUD requires a store to have been set via
956 :meth:`set_mental_model_store`; otherwise these methods raise
957 :class:`RuntimeError` with a hint. Cleanest place to fail —
958 attempts to call the methods without storage would otherwise
959 crash deeper inside the pipeline.
960 """
961 if self._mental_model_store is None:
962 raise RuntimeError(
963 "mental_model_store not configured — call "
964 "Astrocyte.set_mental_model_store(store) before using "
965 "mental-model CRUD."
966 )
967 from astrocyte.pipeline.mental_model import MentalModelService
969 return MentalModelService(self._mental_model_store) # type: ignore[arg-type]
971 async def list_mental_models(
972 self,
973 bank_id: str | None = None,
974 *,
975 scope: str | None = None,
976 context: AstrocyteContext | None = None,
977 ) -> "list[MentalModel]":
978 """List mental models in a bank. Optionally filter by scope."""
979 bid = self._policy.resolve_read_bank_ids(bank_id, None, context)[0]
980 self._policy.check_access(bid, "read", context)
981 return await self._mental_model_service().list(bid, scope=scope)
983 async def get_mental_model(
984 self,
985 model_id: str,
986 bank_id: str | None = None,
987 *,
988 context: AstrocyteContext | None = None,
989 ) -> "MentalModel | None":
990 """Fetch one mental model by id. Returns None if not found / deleted."""
991 bid = self._policy.resolve_read_bank_ids(bank_id, None, context)[0]
992 self._policy.check_access(bid, "read", context)
993 return await self._mental_model_service().get(bank_id=bid, model_id=model_id)
995 async def create_mental_model(
996 self,
997 *,
998 model_id: str,
999 title: str,
1000 content: str | None = None,
1001 sections: list[dict] | None = None,
1002 scope: str = "bank",
1003 source_ids: list[str] | None = None,
1004 bank_id: str | None = None,
1005 context: AstrocyteContext | None = None,
1006 ) -> "MentalModel":
1007 """Create or refresh a mental model.
1009 Two content shapes (mutually exclusive — provide one):
1011 - ``content``: raw markdown string. Parsed on first refresh into
1012 a structured doc; legacy-shape path.
1013 - ``sections``: list of section dicts matching
1014 :class:`astrocyte.pipeline.structured_doc.Section` JSON shape.
1015 Modern path — the structured doc is populated immediately and
1016 ``content`` is rendered from it.
1018 Pass exactly one. If both are provided, ``sections`` wins.
1019 """
1020 bid = self._policy.resolve_read_bank_ids(bank_id, None, context)[0]
1021 self._policy.check_access(bid, "write", context)
1022 self._policy.check_rate_and_quota(bid, "retain")
1024 service = self._mental_model_service()
1026 if sections is not None:
1027 from astrocyte.pipeline.structured_doc import (
1028 StructuredDocument,
1029 render_document,
1030 )
1032 doc = StructuredDocument.model_validate({"sections": sections})
1033 rendered = render_document(doc)
1034 return await service.create(
1035 bank_id=bid,
1036 model_id=model_id,
1037 title=title,
1038 content=rendered,
1039 scope=scope,
1040 source_ids=source_ids,
1041 structured_doc=doc.model_dump(),
1042 )
1044 if content is None:
1045 raise ValueError("create_mental_model requires either `content` or `sections`")
1047 return await service.create(
1048 bank_id=bid,
1049 model_id=model_id,
1050 title=title,
1051 content=content,
1052 scope=scope,
1053 source_ids=source_ids,
1054 )
1056 async def update_mental_model(
1057 self,
1058 *,
1059 model_id: str,
1060 operations: list[dict],
1061 bank_id: str | None = None,
1062 context: AstrocyteContext | None = None,
1063 ) -> "tuple[MentalModel, dict] | None":
1064 """Apply structured delta operations to a mental model.
1066 See :func:`astrocyte.pipeline.delta_ops.apply_operations` for
1067 the operation schema. Conservative-failure contract — invalid
1068 ops drop with a logged reason; the document never gets worse
1069 than its input.
1071 Returns ``(updated_model, summary)`` where summary is
1072 ``{"applied": [...], "skipped": [...], "changed": bool}`` —
1073 the audit trail of which ops landed. Returns ``None`` when
1074 the model doesn't exist.
1075 """
1076 bid = self._policy.resolve_read_bank_ids(bank_id, None, context)[0]
1077 self._policy.check_access(bid, "write", context)
1078 self._policy.check_rate_and_quota(bid, "retain")
1079 return await self._mental_model_service().update_via_ops(
1080 bank_id=bid,
1081 model_id=model_id,
1082 operations=operations,
1083 )
1085 async def delete_mental_model(
1086 self,
1087 model_id: str,
1088 bank_id: str | None = None,
1089 *,
1090 context: AstrocyteContext | None = None,
1091 ) -> bool:
1092 """Soft-delete a mental model. Returns True iff it existed."""
1093 bid = self._policy.resolve_read_bank_ids(bank_id, None, context)[0]
1094 self._policy.check_access(bid, "forget", context)
1095 return await self._mental_model_service().delete(bank_id=bid, model_id=model_id)
1097 async def refresh_mental_model(
1098 self,
1099 model_id: str,
1100 new_source_ids: list[str],
1101 bank_id: str | None = None,
1102 *,
1103 context: AstrocyteContext | None = None,
1104 ) -> "MentalModel | None":
1105 """M28 — re-derive a mental model from new/expanded source memories.
1107 Hindsight parity for ``mental_model.refresh()``: after retain
1108 adds memories that pertain to an existing model, this folds
1109 them into the model's ``source_ids`` and bumps the revision.
1111 The store-level implementation merges ``new_source_ids`` into
1112 the existing set (order-preserving dedup); future enhancement
1113 invokes the LLM compile pipeline against the merged source set
1114 to also rewrite ``content`` and ``structured_doc`` — the SPI
1115 surface is the same.
1117 Returns the refreshed :class:`MentalModel` or ``None`` if the
1118 model doesn't exist.
1119 """
1120 bid = self._policy.resolve_read_bank_ids(bank_id, None, context)[0]
1121 self._policy.check_access(bid, "write", context)
1122 self._policy.check_rate_and_quota(bid, "retain")
1123 return await self._mental_model_service().refresh_from_sources(
1124 bank_id=bid,
1125 model_id=model_id,
1126 new_source_ids=list(new_source_ids),
1127 )
1129 async def create_directive(
1130 self,
1131 *,
1132 rule_text: str,
1133 directive_id: str | None = None,
1134 scope: str = "bank",
1135 bank_id: str | None = None,
1136 context: AstrocyteContext | None = None,
1137 ) -> "MentalModel":
1138 """Create a user-authored hard rule (M18b/M19 deferred item).
1140 Directives are stored as :class:`MentalModel` rows with
1141 ``kind="directive"``. They participate in the agentic reflect
1142 loop's ``search_mental_models`` tool exactly like ``general``
1143 / ``preference`` models, but the ``directive`` discriminator
1144 signals to the answerer that the rule should be applied as a
1145 hard preference override rather than as one input among many.
1147 Mirrors Hindsight's ``mental_models.subtype="directive"``
1148 rows. Architecturally replaces the M18a-2 ``directive_compile``
1149 auto-extraction path that was deprecated in M19 (it replicated
1150 a −30pp SSP regression because compressed auto-directives
1151 overrode original preference nuance).
1152 """
1153 import uuid as _uuid
1155 bid = self._policy.resolve_read_bank_ids(bank_id, None, context)[0]
1156 self._policy.check_access(bid, "write", context)
1157 self._policy.check_rate_and_quota(bid, "retain")
1159 if not rule_text.strip():
1160 raise ValueError("create_directive requires non-empty rule_text")
1162 mid = directive_id or f"directive:{_uuid.uuid4().hex[:12]}"
1163 service = self._mental_model_service()
1164 # Author the directive with a single-paragraph structured doc so
1165 # subsequent refreshes can use delta_ops cleanly.
1166 from astrocyte.pipeline.structured_doc import (
1167 ParagraphBlock,
1168 Section,
1169 StructuredDocument,
1170 render_document,
1171 )
1173 doc = StructuredDocument(
1174 sections=[
1175 Section(
1176 id="rule",
1177 heading="Rule",
1178 level=2,
1179 blocks=[ParagraphBlock(text=rule_text.strip())],
1180 ),
1181 ],
1182 )
1183 rendered = render_document(doc)
1184 return await service.create(
1185 bank_id=bid,
1186 model_id=mid,
1187 title="Directive",
1188 content=rendered,
1189 scope=scope,
1190 kind="directive",
1191 structured_doc=doc.model_dump(),
1192 )
1194 # ── M21 — Observation CRUD (Hindsight parity) ────────────────────
1196 async def list_observations(
1197 self,
1198 bank_id: str | None = None,
1199 *,
1200 scope: str | None = None,
1201 trend: "str | None" = None,
1202 limit: int = 100,
1203 context: AstrocyteContext | None = None,
1204 ) -> list[dict]:
1205 """List observations in a bank.
1207 Returns a list of dicts ``{id, text, trend, proof_count,
1208 source_ids, confidence, updated_at}`` so callers (incl. the
1209 MCP tool) get a JSON-shaped output without round-tripping
1210 through ``VectorHit``.
1212 Args:
1213 scope: Filter by ``_obs_scope`` metadata (default ``"bank"``
1214 if observations were created without a tag).
1215 trend: Filter by computed trend (``"new"`` / ``"stable"`` /
1216 ``"strengthening"`` / ``"weakening"`` / ``"stale"``).
1217 Trend is computed at call time from each observation's
1218 source-timestamp metadata; see
1219 :func:`astrocyte.pipeline.observation.compute_observation_trend`.
1220 limit: Max rows. Hard cap 1000.
1221 """
1222 from astrocyte.pipeline.observation import (
1223 compute_observation_trend,
1224 obs_bank_id,
1225 )
1227 bid = self._policy.resolve_read_bank_ids(bank_id, None, context)[0]
1228 self._policy.check_access(bid, "read", context)
1230 if self._pipeline is None or not hasattr(self._pipeline, "vector_store"):
1231 raise RuntimeError("vector_store not configured on the pipeline")
1232 store = self._pipeline.vector_store # type: ignore[attr-defined]
1233 if store is None:
1234 raise RuntimeError("vector_store is not set")
1236 obs_bank = obs_bank_id(bid)
1237 cap = max(1, min(int(limit), 1000))
1238 items = await store.list_vectors(
1239 obs_bank,
1240 offset=0,
1241 limit=cap,
1242 )
1243 out: list[dict] = []
1244 for item in items:
1245 meta = item.metadata or {}
1246 if scope is not None and str(meta.get("_obs_scope", "")) != scope:
1247 continue
1248 t = compute_observation_trend(dict(meta)).value
1249 if trend is not None and t != trend:
1250 continue
1251 out.append(
1252 {
1253 "id": item.id,
1254 "text": item.text,
1255 "trend": t,
1256 "proof_count": int(meta.get("_obs_proof_count", 1)),
1257 "source_ids": meta.get("_obs_source_ids", "[]"),
1258 "confidence": meta.get("_obs_confidence"),
1259 "updated_at": meta.get("_obs_updated_at"),
1260 "scope": meta.get("_obs_scope"),
1261 }
1262 )
1263 return out
1265 async def get_observation(
1266 self,
1267 observation_id: str,
1268 bank_id: str | None = None,
1269 *,
1270 context: AstrocyteContext | None = None,
1271 ) -> dict | None:
1272 """Fetch one observation by id. Returns None if not found."""
1273 rows = await self.list_observations(
1274 bank_id=bank_id,
1275 limit=1000,
1276 context=context,
1277 )
1278 for r in rows:
1279 if r["id"] == observation_id:
1280 return r
1281 return None
1283 async def create_observation(
1284 self,
1285 *,
1286 text: str,
1287 source_ids: list[str] | None = None,
1288 scope: str | None = None,
1289 confidence: float = 0.9,
1290 bank_id: str | None = None,
1291 context: AstrocyteContext | None = None,
1292 ) -> dict:
1293 """User-authored observation.
1295 Bypasses the autonomous consolidator — for cases where an
1296 agent or user wants to assert a fact directly without going
1297 through retain → consolidate. The observation is stored in
1298 the dedicated ``::obs`` bank with proof_count=1 and the given
1299 source_ids (may be empty for hand-authored entries).
1300 """
1301 from astrocyte.pipeline.observation import (
1302 ObservationConsolidator,
1303 observation_scope,
1304 )
1306 bid = self._policy.resolve_read_bank_ids(bank_id, None, context)[0]
1307 self._policy.check_access(bid, "write", context)
1308 self._policy.check_rate_and_quota(bid, "retain")
1310 if self._pipeline is None or not hasattr(self._pipeline, "vector_store"):
1311 raise RuntimeError("vector_store not configured on the pipeline")
1312 store = self._pipeline.vector_store # type: ignore[attr-defined]
1313 llm = self._pipeline.llm_provider # type: ignore[attr-defined]
1314 if store is None or llm is None:
1315 raise RuntimeError("vector_store / llm_provider not set")
1317 consolidator = ObservationConsolidator()
1318 obs_scope_value = scope or observation_scope(bid)
1319 ok = await consolidator._apply_create( # noqa: SLF001 — single-step CRUD
1320 {"text": text.strip(), "confidence": confidence},
1321 bid,
1322 store,
1323 llm,
1324 source_ids=list(source_ids or []),
1325 now_iso=datetime.now(timezone.utc).isoformat(),
1326 scope=obs_scope_value,
1327 )
1328 if not ok:
1329 raise ValueError("observation create rejected (empty text or confidence below threshold)")
1330 # The freshly-created row is at the top of the obs bank — list and
1331 # find it. (For multi-writer setups, callers can use the returned
1332 # text + scope as the identity check.)
1333 rows = await self.list_observations(bank_id=bid, scope=obs_scope_value, context=context)
1334 for r in rows:
1335 if r["text"] == text.strip():
1336 return r
1337 # Fall back: return a minimal dict so the caller doesn't get None
1338 return {"text": text.strip(), "scope": obs_scope_value, "trend": "new"}
1340 async def delete_observation(
1341 self,
1342 observation_id: str,
1343 bank_id: str | None = None,
1344 *,
1345 context: AstrocyteContext | None = None,
1346 ) -> bool:
1347 """Delete one observation from the dedicated obs bank."""
1348 from astrocyte.pipeline.observation import obs_bank_id
1350 bid = self._policy.resolve_read_bank_ids(bank_id, None, context)[0]
1351 self._policy.check_access(bid, "forget", context)
1353 if self._pipeline is None or not hasattr(self._pipeline, "vector_store"):
1354 raise RuntimeError("vector_store not configured on the pipeline")
1355 store = self._pipeline.vector_store # type: ignore[attr-defined]
1356 if store is None:
1357 return False
1358 obs_bank = obs_bank_id(bid)
1359 deleted = await store.delete([observation_id], obs_bank)
1360 return deleted > 0
1362 async def clear_bank(
1363 self,
1364 bank_id: str,
1365 *,
1366 context: AstrocyteContext | None = None,
1367 ) -> ForgetResult:
1368 """Delete all memories in a bank. Requires admin access if ACL enabled."""
1369 return await self.forget(bank_id, scope="all", context=context)
1371 async def forget(
1372 self,
1373 bank_id: str,
1374 *,
1375 memory_ids: list[str] | None = None,
1376 tags: list[str] | None = None,
1377 scope: str | None = None,
1378 context: AstrocyteContext | None = None,
1379 compliance: bool = False,
1380 reason: str | None = None,
1381 before_date: datetime | None = None,
1382 ) -> ForgetResult:
1383 """Remove memories.
1385 Args:
1386 scope: ``"all"`` to delete everything in a bank (requires admin).
1387 compliance: Bypass legal holds for right-to-forget (requires context).
1388 reason: Audit reason when ``compliance=True``.
1389 before_date: Delete memories created before this date.
1390 """
1391 validate_bank_id(bank_id)
1392 with span("astrocyte.forget", {"astrocyte.bank_id": bank_id}):
1393 # scope="all" requires admin permission
1394 if scope == "all":
1395 if self._config.access_control.enabled:
1396 self._policy.check_access(bank_id, "admin", context)
1397 else:
1398 self._policy.check_access(bank_id, "forget", context)
1400 # Legal hold check — compliance=True bypasses for right-to-forget.
1401 # Even when access_control is disabled, compliance bypass requires
1402 # explicit context (caller must identify themselves).
1403 # MIP forget policy resolution (Phase 4) — runs BEFORE the existing
1404 # legal-hold check so that a rule with respect_legal_hold=True wins
1405 # over the caller's compliance bypass, and so audit logs fire even
1406 # on policy refusal.
1407 await self._load_holds(bank_id)
1408 mip_forget = self._mip_router.resolve_forget_for_bank(bank_id) if self._mip_router else None
1409 if mip_forget is not None:
1410 # max_per_call: cap blast radius for selective deletes by id
1411 if (
1412 mip_forget.max_per_call is not None
1413 and memory_ids is not None
1414 and len(memory_ids) > mip_forget.max_per_call
1415 ):
1416 raise MipRoutingError(
1417 f"forget rejected: {len(memory_ids)} memory_ids exceeds "
1418 f"forget.max_per_call={mip_forget.max_per_call} for bank {bank_id!r}"
1419 )
1421 # min_age_days: refuse forget when any targeted record is younger.
1422 # Best-effort: requires the engine to populate `_created_at` in
1423 # metadata at retain time. Records lacking this stamp are skipped
1424 # with a warning (degraded enforcement, not a hard failure).
1425 if mip_forget.min_age_days is not None and mip_forget.min_age_days > 0 and memory_ids:
1426 too_young = await self._collect_too_young_ids(
1427 bank_id,
1428 memory_ids,
1429 mip_forget.min_age_days,
1430 )
1431 if too_young:
1432 raise MipRoutingError(
1433 f"forget rejected: {len(too_young)} record(s) in {bank_id!r} "
1434 f"younger than forget.min_age_days={mip_forget.min_age_days} "
1435 f"({sorted(too_young)[:5]}{'...' if len(too_young) > 5 else ''})"
1436 )
1438 # audit: emit a structured log line before any deletion occurs
1439 if mip_forget.audit in ("required", "recommended"):
1440 self._logger.log(
1441 "astrocyte.mip.forget.audit",
1442 bank_id=bank_id,
1443 data={
1444 "mode": str(mip_forget.mode or "soft"),
1445 "audit": str(mip_forget.audit),
1446 "cascade": bool(mip_forget.cascade) if mip_forget.cascade is not None else True,
1447 "memory_ids_count": len(memory_ids) if memory_ids else 0,
1448 "scope": scope or "selective",
1449 "actor": context_principal_label(context) if context else "anonymous",
1450 "reason": str(reason) if reason else None,
1451 },
1452 level=logging.WARNING,
1453 )
1455 # respect_legal_hold: when True, override the compliance bypass.
1456 # The MIP rule is the source of truth; compliance flag cannot
1457 # circumvent a rule that explicitly demands legal hold respect.
1458 if mip_forget.respect_legal_hold:
1459 self._lifecycle.check_forget_allowed(bank_id)
1461 if not compliance:
1462 # Skip duplicate check if MIP already enforced legal hold above.
1463 if mip_forget is None or not mip_forget.respect_legal_hold:
1464 self._lifecycle.check_forget_allowed(bank_id)
1465 else:
1466 if context is None:
1467 raise AccessDenied("anonymous", bank_id, "compliance_forget")
1468 # Log compliance forget with actor identity for audit trail
1469 principal_label = context_principal_label(context)
1470 audit_reason = reason or "compliance_forget_request"
1471 self._logger.log(
1472 "astrocyte.compliance.forget",
1473 bank_id=bank_id,
1474 data={
1475 "actor": principal_label,
1476 "reason": str(audit_reason),
1477 "scope": scope or "selective",
1478 },
1479 level=logging.WARNING,
1480 )
1481 # When access control is enabled, also require admin permission
1482 if self._config.access_control.enabled:
1483 self._policy.check_access(bank_id, "admin", context)
1485 # Soft-delete path (mip_forget.mode == "soft"): mark records with
1486 # `_deleted: true` instead of physically removing. Recall is
1487 # responsible for filtering them out. Falls through to hard delete
1488 # with a warning if the engine doesn't expose `soft_delete`.
1489 soft_mode = mip_forget is not None and mip_forget.mode == "soft" and memory_ids is not None
1490 if soft_mode:
1491 soft_fn = getattr(self._engine_provider, "soft_delete", None)
1492 if soft_fn is not None:
1493 deleted = await soft_fn(bank_id, memory_ids)
1495 # Same dedup-cache invalidation as the hard-delete path —
1496 # a soft-deleted memory is no longer reachable on recall,
1497 # so its embedding shouldn't keep blocking re-retains.
1498 if self._pipeline is not None and deleted > 0:
1499 self._pipeline.invalidate_dedup_cache(bank_id, memory_ids)
1501 await self._hook_manager.fire(
1502 "on_forget",
1503 bank_id=bank_id,
1504 data={"deleted_count": deleted, "archived_count": 0, "soft": True},
1505 )
1506 return ForgetResult(deleted_count=deleted)
1507 logging.getLogger("astrocyte.mip").warning(
1508 "forget.mode=soft requested for bank=%s but engine %s does not "
1509 "implement soft_delete(); falling back to hard delete",
1510 _safe_log(bank_id),
1511 type(self._engine_provider).__name__ if self._engine_provider else "pipeline",
1512 )
1514 request = ForgetRequest(
1515 bank_id=bank_id,
1516 memory_ids=memory_ids,
1517 tags=tags,
1518 before_date=before_date,
1519 scope=scope,
1520 )
1521 result = await self._dispatcher.forget(request)
1523 # Drop the forgotten memories from the in-memory dedup cache so
1524 # a re-retain of similar content produces a fresh row instead of
1525 # silently no-op'ing with "All chunks are near-duplicates". The
1526 # DedupDetector lives on the pipeline; safe to skip when no
1527 # pipeline is configured (provider-direct deployments).
1528 if self._pipeline is not None and result.deleted_count > 0:
1529 self._pipeline.invalidate_dedup_cache(bank_id, memory_ids)
1531 await self._hook_manager.fire(
1532 "on_forget",
1533 bank_id=bank_id,
1534 data={
1535 "deleted_count": result.deleted_count,
1536 "archived_count": result.archived_count,
1537 },
1538 )
1539 return result
1541 async def _collect_too_young_ids(
1542 self,
1543 bank_id: str,
1544 memory_ids: list[str],
1545 min_age_days: int,
1546 ) -> list[str]:
1547 """Return the subset of ``memory_ids`` younger than ``min_age_days``.
1549 Best-effort enforcement of ``forget.min_age_days``: relies on records
1550 carrying a ``_created_at`` ISO timestamp in metadata (stamped by the
1551 engine at retain time). Records lacking the stamp are skipped with a
1552 warning — degraded enforcement, not a hard failure, since older data
1553 predating the stamp shouldn't permanently block legitimate forgets.
1554 """
1555 from datetime import timedelta
1557 wanted = set(memory_ids)
1558 try:
1559 result = await self._dispatcher.recall(
1560 RecallRequest(query="*", bank_id=bank_id, max_results=10000),
1561 )
1562 except Exception as exc: # pragma: no cover — defensive
1563 logger.warning(
1564 "min_age_days check skipped: recall failed for bank=%s: %s",
1565 _safe_log(bank_id),
1566 _safe_log(exc),
1567 )
1568 return []
1570 now = datetime.now(timezone.utc)
1571 threshold = timedelta(days=min_age_days)
1572 too_young: list[str] = []
1573 seen: set[str] = set()
1574 missing_stamp = 0
1576 for hit in result.hits:
1577 if hit.memory_id is None or hit.memory_id not in wanted:
1578 continue
1579 seen.add(hit.memory_id)
1580 stamp = (hit.metadata or {}).get("_created_at")
1581 if stamp is None:
1582 missing_stamp += 1
1583 continue
1584 if isinstance(stamp, str):
1585 try:
1586 created_at = datetime.fromisoformat(stamp)
1587 except ValueError:
1588 missing_stamp += 1
1589 continue
1590 elif isinstance(stamp, datetime):
1591 created_at = stamp
1592 else:
1593 missing_stamp += 1
1594 continue
1595 if created_at.tzinfo is None:
1596 created_at = created_at.replace(tzinfo=timezone.utc)
1597 if (now - created_at) < threshold:
1598 too_young.append(hit.memory_id)
1600 if missing_stamp:
1601 mip_logger = logging.getLogger("astrocyte.mip")
1602 mip_logger.warning(
1603 "min_age_days enforcement degraded: %d/%d record(s) in bank=%s "
1604 "lack `_created_at` metadata and were skipped",
1605 missing_stamp,
1606 len(seen),
1607 _safe_log(bank_id),
1608 )
1609 # One increment per forget call that encountered unstamped records.
1610 # The warning log above carries the exact count; this counter
1611 # fires dashboards / alerts when the condition occurs at all.
1612 self._metrics.inc_counter(
1613 "astrocyte_forget_unstamped_records_total",
1614 {"bank_id": bank_id},
1615 )
1616 return too_young
1618 async def compile(
1619 self,
1620 bank_id: str,
1621 scope: str | None = None,
1622 *,
1623 context: AstrocyteContext | None = None,
1624 ) -> CompileResult:
1625 """Compile raw memories into WikiPages for ``bank_id`` (M8).
1627 Synthesises a structured wiki page for each detected topic scope using
1628 an LLM. Compiled pages are stored in the WikiStore and embedded back
1629 into the VectorStore (``memory_layer="compiled"``) so the recall pipeline
1630 can surface them ahead of raw memory fragments.
1632 Args:
1633 bank_id: Bank to compile.
1634 scope: If provided, compile only memories tagged with this scope string.
1635 If ``None``, trigger full scope discovery:
1637 1. Tagged memories are grouped by tag (each tag = one page).
1638 2. Untagged memories are clustered by embedding similarity
1639 (DBSCAN); each cluster is labelled with a lightweight LLM
1640 call. Noise points are held for the next compile cycle.
1642 Returns:
1643 :class:`~astrocyte.types.CompileResult` with pages created/updated,
1644 noise count, and token usage.
1646 Raises:
1647 :class:`~astrocyte.errors.ConfigError`: If no WikiStore has been
1648 configured (call ``set_wiki_store()`` before ``compile()``).
1649 :class:`~astrocyte.errors.ProviderUnavailable`: If the Tier 1 pipeline
1650 has not been configured.
1652 Example::
1654 # Explicit — compile memories tagged "incident-response"
1655 result = await brain.compile("eng-team", scope="incident-response")
1657 # Automatic — discover scopes from tags and embedding clusters
1658 result = await brain.compile("eng-team")
1660 print(result.pages_created, result.pages_updated, result.tokens_used)
1661 """
1662 validate_bank_id(bank_id)
1663 # Authorization parity with recall/reflect: compiling a bank reads
1664 # and rewrites its memories, so a caller needs at least read access.
1665 self._policy.check_access(bank_id, "read", context)
1667 if self._wiki_store is None:
1668 raise ConfigError(
1669 "brain.compile() requires a WikiStore. Call brain.set_wiki_store(store) before compiling."
1670 )
1672 if self._pipeline is None:
1673 raise ProviderUnavailable(
1674 "brain.compile() requires a Tier 1 pipeline. Call brain.set_pipeline(pipeline) before compiling."
1675 )
1677 from astrocyte.pipeline.compile import CompileEngine
1679 engine = CompileEngine(
1680 vector_store=self._pipeline.vector_store,
1681 llm_provider=self._pipeline.llm_provider,
1682 wiki_store=self._wiki_store, # type: ignore[arg-type]
1683 )
1685 with span("compile", {"bank_id": bank_id, "scope": scope or "auto"}):
1686 result = await engine.run(bank_id, scope=scope)
1688 if result.error:
1689 logger.warning(
1690 "compile failed for bank %s scope %s: %s",
1691 _safe_log(bank_id),
1692 _safe_log(scope or "auto"),
1693 _safe_log(result.error),
1694 )
1695 else:
1696 logger.info(
1697 "compile complete bank=%s scope=%s pages_created=%d pages_updated=%d noise=%d tokens=%d elapsed_ms=%d",
1698 _safe_log(bank_id),
1699 _safe_log(scope or "auto"),
1700 result.pages_created,
1701 result.pages_updated,
1702 result.noise_memories,
1703 result.tokens_used,
1704 result.elapsed_ms,
1705 )
1707 return result
1709 # ---------------------------------------------------------------------------
1710 # Graph traversal (public API)
1711 # ---------------------------------------------------------------------------
1713 async def graph_search(
1714 self,
1715 query: str,
1716 bank_id: str,
1717 limit: int = 10,
1718 *,
1719 context: AstrocyteContext | None = None,
1720 ) -> list[Entity]:
1721 """Search the knowledge graph for entities matching *query*.
1723 Performs a name/alias search against the graph store and returns
1724 matching :class:`~astrocyte.types.Entity` objects. Useful for
1725 resolving entity references before calling :meth:`graph_neighbors`.
1727 Args:
1728 query: Name or partial name to search for.
1729 bank_id: Memory bank whose graph to search.
1730 limit: Maximum number of entities to return.
1732 Returns:
1733 Matching entities ordered by relevance.
1735 Raises:
1736 :class:`~astrocyte.errors.ConfigError`: If no graph store has been
1737 configured (set ``graph_store`` on the pipeline).
1738 """
1739 from astrocyte.errors import ConfigError
1741 validate_bank_id(bank_id)
1742 # Reading a bank's knowledge graph is a read over that bank's memories.
1743 self._policy.check_access(bank_id, "read", context)
1744 graph_store = getattr(self._pipeline, "graph_store", None) if self._pipeline else None
1745 if graph_store is None:
1746 raise ConfigError(
1747 "graph_search() requires a GraphStore. Configure a graph_store provider in astrocyte.yaml."
1748 )
1749 return await graph_store.query_entities(query, bank_id, limit=limit)
1751 async def graph_neighbors(
1752 self,
1753 entity_ids: list[str],
1754 bank_id: str,
1755 max_depth: int = 2,
1756 limit: int = 20,
1757 *,
1758 context: AstrocyteContext | None = None,
1759 ) -> list[GraphHit]:
1760 """Traverse the knowledge graph from *entity_ids* and return connected memories.
1762 Walks the entity graph up to *max_depth* hops from each seed entity
1763 and returns the memories attached to discovered entities, scored by
1764 proximity.
1766 Args:
1767 entity_ids: Seed entity IDs to start traversal from.
1768 bank_id: Memory bank whose graph to traverse.
1769 max_depth: Maximum traversal depth (default 2).
1770 limit: Maximum number of memory hits to return.
1772 Returns:
1773 :class:`~astrocyte.types.GraphHit` objects sorted by relevance.
1775 Raises:
1776 :class:`~astrocyte.errors.ConfigError`: If no graph store has been
1777 configured.
1778 """
1779 from astrocyte.errors import ConfigError
1781 validate_bank_id(bank_id)
1782 # Traversing a bank's graph returns that bank's memories — read access.
1783 self._policy.check_access(bank_id, "read", context)
1784 if not entity_ids:
1785 return []
1786 graph_store = getattr(self._pipeline, "graph_store", None) if self._pipeline else None
1787 if graph_store is None:
1788 raise ConfigError(
1789 "graph_neighbors() requires a GraphStore. Configure a graph_store provider in astrocyte.yaml."
1790 )
1791 return await graph_store.query_neighbors(entity_ids, bank_id, max_depth=max_depth, limit=limit)
1793 async def history(
1794 self,
1795 query: str,
1796 bank_id: str,
1797 as_of: datetime,
1798 *,
1799 max_results: int = 10,
1800 max_tokens: int | None = None,
1801 tags: list[str] | None = None,
1802 context: AstrocyteContext | None = None,
1803 ) -> HistoryResult:
1804 """Reconstruct what the agent knew at a past point in time (M9 time travel).
1806 Returns memories that existed in *bank_id* at the moment *as_of* — i.e.
1807 only memories whose ``retained_at`` timestamp is on or before *as_of*.
1808 Memories retained after *as_of* are excluded, giving a faithful snapshot
1809 of the agent's knowledge at that instant.
1811 Args:
1812 query: The recall query to run against the historical snapshot.
1813 bank_id: Bank to query.
1814 as_of: UTC datetime. Memories retained after this moment are hidden.
1815 max_results: Maximum number of hits to return.
1816 max_tokens: Optional token budget for the result set.
1817 tags: Optional tag filter (applied on top of the time filter).
1819 Returns:
1820 :class:`~astrocyte.types.HistoryResult` with hits and the ``as_of``
1821 timestamp embedded for traceability.
1823 Raises:
1824 ConfigError: If no pipeline is configured (no vector store to query).
1826 Example::
1828 from datetime import datetime, UTC
1829 snapshot = await brain.history(
1830 "What did we know about Alice?",
1831 bank_id="user-alice",
1832 as_of=datetime(2025, 1, 1, tzinfo=UTC),
1833 )
1834 for hit in snapshot.hits:
1835 print(hit.retained_at, hit.text)
1836 """
1837 recall_result = await self.recall(
1838 query,
1839 bank_id=bank_id,
1840 max_results=max_results,
1841 max_tokens=max_tokens,
1842 tags=tags,
1843 as_of=as_of,
1844 context=context,
1845 )
1846 return HistoryResult(
1847 hits=recall_result.hits,
1848 total_available=recall_result.total_available,
1849 truncated=recall_result.truncated,
1850 as_of=as_of,
1851 bank_id=bank_id,
1852 trace=recall_result.trace,
1853 )
1855 async def audit(
1856 self,
1857 scope: str,
1858 bank_id: str,
1859 *,
1860 max_memories: int = 50,
1861 max_tokens: int | None = None,
1862 tags: list[str] | None = None,
1863 context: AstrocyteContext | None = None,
1864 ) -> AuditResult:
1865 """Identify knowledge gaps for a topic in a memory bank (M10 gap analysis).
1867 Recalls up to *max_memories* relevant memories for *scope*, then calls
1868 an LLM audit judge to assess what is missing or under-covered. The
1869 result includes a list of :class:`~astrocyte.types.GapItem` objects and
1870 a ``coverage_score`` between 0 (empty bank) and 1 (comprehensive).
1872 This is a diagnostic operation — it does not modify any stored memory.
1873 It does consume LLM tokens proportional to the number of memories scanned.
1875 Args:
1876 scope: Natural-language description of the topic to audit
1877 (e.g. ``"Alice's employment history"``).
1878 bank_id: Bank to audit.
1879 max_memories: Maximum number of memories to retrieve and pass to
1880 the audit judge. Defaults to ``50``.
1881 max_tokens: Optional token budget applied to retrieved memories
1882 before the judge call.
1883 tags: Optional tag filter to narrow which memories are retrieved.
1885 Returns:
1886 :class:`~astrocyte.types.AuditResult` with gaps and coverage score.
1888 Raises:
1889 ConfigError: If no pipeline is configured.
1891 Example::
1893 result = await brain.audit(
1894 "Alice's employment history",
1895 bank_id="user-alice",
1896 )
1897 print(f"Coverage: {result.coverage_score:.0%}")
1898 for gap in result.gaps:
1899 print(f"[{gap.severity}] {gap.topic}: {gap.reason}")
1900 """
1901 from astrocyte.pipeline.audit import run_audit
1903 recall_result = await self.recall(
1904 scope,
1905 bank_id=bank_id,
1906 max_results=max_memories,
1907 max_tokens=max_tokens,
1908 tags=tags,
1909 context=context,
1910 )
1912 pipeline = self._pipeline
1913 if pipeline is None:
1914 from astrocyte.exceptions import ConfigError
1916 raise ConfigError("No pipeline configured — call set_pipeline() first.")
1918 return await run_audit(
1919 scope=scope,
1920 bank_id=bank_id,
1921 memories=recall_result.hits,
1922 llm_provider=pipeline.llm_provider,
1923 trace=recall_result.trace,
1924 )
1926 async def health(self) -> HealthStatus:
1927 """Check system health."""
1928 with timed() as t:
1929 if self._engine_provider:
1930 status = await self._engine_provider.health()
1931 elif self._pipeline:
1932 status = await self._pipeline.vector_store.health()
1933 else:
1934 status = HealthStatus(healthy=True, message="No provider configured")
1935 status.latency_ms = t["elapsed_ms"]
1936 return status
1938 # ---------------------------------------------------------------------------
1939 # Lifecycle — legal hold + TTL
1940 # ---------------------------------------------------------------------------
1942 def set_legal_hold(self, bank_id: str, hold_id: str, reason: str, *, set_by: str = "user:api") -> LegalHold:
1943 """Place a bank under legal hold in this process only. Blocks forget() until released.
1945 Kept in memory: lost on restart and unseen by other processes. Use
1946 :meth:`place_legal_hold` to persist a hold in the store."""
1947 return self._lifecycle.set_legal_hold(bank_id, hold_id, reason, set_by=set_by)
1949 def _hold_store(self) -> Any | None:
1950 """The vector store when it persists legal holds (optional methods)."""
1951 store = self._pipeline.vector_store if self._pipeline is not None else None
1952 if store is None or getattr(store, "list_legal_holds", None) is None:
1953 return None
1954 return store
1956 async def _load_holds(self, bank_id: str) -> None:
1957 """Refresh the bank's persisted holds before a check: another process
1958 or replica may have placed or lifted one since."""
1959 store = self._hold_store()
1960 if store is not None:
1961 self._lifecycle.set_stored_holds(bank_id, await store.list_legal_holds(bank_id))
1963 async def place_legal_hold(
1964 self, bank_id: str, hold_id: str, reason: str, *, set_by: str = "user:api"
1965 ) -> LegalHold:
1966 """Place a bank under legal hold, persisted in the store: it survives
1967 restarts and every process and replica sharing the store honours it.
1968 Blocks forget() and lifecycle deletion until lifted; a compliance
1969 forget (right to erasure) is not blocked unless a MIP rule says
1970 ``respect_legal_hold``. Without a store that persists holds (the
1971 optional ``save_legal_hold`` / ``delete_legal_hold`` /
1972 ``list_legal_holds``), the hold is kept in this process only."""
1973 validate_bank_id(bank_id)
1974 store = self._hold_store()
1975 if store is None:
1976 return self._lifecycle.set_legal_hold(bank_id, hold_id, reason, set_by=set_by)
1977 hold = LegalHold(hold_id=hold_id, bank_id=bank_id, reason=reason,
1978 set_at=datetime.now(timezone.utc), set_by=set_by)
1979 await store.save_legal_hold(hold)
1980 await self._load_holds(bank_id)
1981 self._logger.log("astrocyte.legal_hold.placed", bank_id=bank_id,
1982 data={"hold_id": hold_id, "set_by": set_by, "reason": reason}, level=logging.WARNING)
1983 return hold
1985 async def lift_legal_hold(self, bank_id: str, hold_id: str) -> bool:
1986 """Lift a legal hold, persisted or set in this process. Returns
1987 whether it existed."""
1988 validate_bank_id(bank_id)
1989 released = self._lifecycle.release_legal_hold(bank_id, hold_id)
1990 store = self._hold_store()
1991 if store is not None:
1992 released = await store.delete_legal_hold(bank_id, hold_id) or released
1993 await self._load_holds(bank_id)
1994 if released:
1995 self._logger.log("astrocyte.legal_hold.lifted", bank_id=bank_id, data={"hold_id": hold_id},
1996 level=logging.WARNING)
1997 return released
1999 async def legal_holds(self, bank_id: str) -> list[LegalHold]:
2000 """The bank's active legal holds, persisted and in-process."""
2001 await self._load_holds(bank_id)
2002 return self._lifecycle.get_holds(bank_id)
2004 def release_legal_hold(self, bank_id: str, hold_id: str) -> bool:
2005 """Release a legal hold from a bank. Returns True if hold existed."""
2006 return self._lifecycle.release_legal_hold(bank_id, hold_id)
2008 def is_under_hold(self, bank_id: str) -> bool:
2009 """Check if bank is under legal hold."""
2010 return self._lifecycle.is_under_hold(bank_id)
2012 async def run_lifecycle(self, bank_id: str) -> LifecycleRunResult:
2013 """Run TTL lifecycle check on a bank. Scan memories, archive/delete as needed.
2015 Note: v1 treats "archive" as delete — no separate archive storage yet.
2016 The LifecycleAction.action distinguishes the reason (ttl_unretrieved vs ttl_expired)
2017 so callers can differentiate, but both result in deletion from the provider.
2018 """
2019 from astrocyte.types import LifecycleAction
2021 if not self._config.lifecycle.enabled:
2022 return LifecycleRunResult(archived_count=0, deleted_count=0, skipped_count=0, actions=[])
2024 await self._load_holds(bank_id) # a persisted hold blocks TTL deletion too
2025 now = datetime.now(timezone.utc)
2026 actions: list[LifecycleAction] = []
2027 to_delete: list[str] = []
2029 # Scan memories via paginated list_vectors (avoids query="*" + 10K limit)
2030 vector_store = self._pipeline.vector_store if self._pipeline else None
2031 if vector_store and hasattr(vector_store, "list_vectors"):
2032 scan_offset = 0
2033 scan_batch = 200
2034 while True:
2035 items = await vector_store.list_vectors(bank_id, offset=scan_offset, limit=scan_batch)
2036 if not items:
2037 break
2038 for item in items:
2039 created_at = item.metadata.get("_created_at") if item.metadata else None
2040 last_recalled = item.metadata.get("_last_recalled_at") if item.metadata else None
2041 if isinstance(created_at, str):
2042 created_at = datetime.fromisoformat(created_at)
2043 if isinstance(last_recalled, str):
2044 last_recalled = datetime.fromisoformat(last_recalled)
2045 action = self._lifecycle.evaluate_memory_ttl(
2046 memory_id=item.id,
2047 bank_id=bank_id,
2048 created_at=created_at,
2049 last_recalled_at=last_recalled,
2050 tags=item.metadata.get("_tags", "").split(",")
2051 if item.metadata and item.metadata.get("_tags")
2052 else None,
2053 fact_type=item.metadata.get("_fact_type") if item.metadata else None,
2054 now=now,
2055 )
2056 actions.append(action)
2057 if action.action in ("delete", "archive"):
2058 to_delete.append(item.id)
2059 scan_offset += len(items)
2060 if len(items) < scan_batch:
2061 break
2062 if scan_offset > 100_000:
2063 logger.warning("Lifecycle scan capped at 100k vectors for bank %s", _safe_log(bank_id))
2064 break
2065 else:
2066 # Fallback for engine-only or stores without list_vectors
2067 logger.warning(
2068 "Lifecycle scan using query='*' fallback for bank %s — "
2069 "results may be incomplete. Use a pipeline with list_vectors support for full coverage.",
2070 _safe_log(bank_id),
2071 )
2072 result = await self._dispatcher.recall(RecallRequest(query="*", bank_id=bank_id, max_results=10000))
2073 for hit in result.hits:
2074 created_at = hit.metadata.get("_created_at") if hit.metadata else None
2075 last_recalled = hit.metadata.get("_last_recalled_at") if hit.metadata else None
2076 if isinstance(created_at, str):
2077 created_at = datetime.fromisoformat(created_at)
2078 if isinstance(last_recalled, str):
2079 last_recalled = datetime.fromisoformat(last_recalled)
2080 action = self._lifecycle.evaluate_memory_ttl(
2081 memory_id=hit.memory_id or "",
2082 bank_id=bank_id,
2083 created_at=created_at,
2084 last_recalled_at=last_recalled,
2085 tags=hit.tags,
2086 fact_type=hit.fact_type,
2087 now=now,
2088 )
2089 actions.append(action)
2090 if action.action in ("delete", "archive"):
2091 to_delete.append(hit.memory_id or "")
2093 # Batch delete/archive
2094 deleted = 0
2095 if to_delete:
2096 forget_result = await self._dispatcher.forget(ForgetRequest(bank_id=bank_id, memory_ids=to_delete))
2097 deleted = forget_result.deleted_count
2099 archived = sum(1 for a in actions if a.action == "archive")
2100 skipped = sum(1 for a in actions if a.action == "keep")
2102 # Run dedup consolidation if pipeline has a vector store
2103 consolidation_removed = 0
2104 if self._pipeline and self._pipeline.vector_store:
2105 from astrocyte.pipeline.consolidation import run_consolidation
2107 cons_result = await run_consolidation(
2108 self._pipeline.vector_store,
2109 bank_id,
2110 graph_store=getattr(self._pipeline, "graph_store", None),
2111 )
2112 consolidation_removed = cons_result.duplicates_removed
2114 return LifecycleRunResult(
2115 archived_count=archived,
2116 deleted_count=deleted + consolidation_removed,
2117 skipped_count=skipped,
2118 actions=actions,
2119 )
2121 # ---------------------------------------------------------------------------
2122 # Bank health & analytics
2123 # ---------------------------------------------------------------------------
2125 async def bank_health(self, bank_id: str) -> "BankHealth":
2126 """Compute health score and issues for a bank.
2128 Uses in-memory operation counters collected since process start.
2129 Optionally enriches with memory count from the vector store.
2131 The VectorStore SPI has no count method, so the count pages
2132 through ``list_vectors``: cost grows with bank size.
2133 """
2134 counters = self._analytics.get_counters(bank_id)
2135 memory_count = 0
2136 if self._pipeline and self._pipeline.vector_store:
2137 store = self._pipeline.vector_store
2138 try:
2139 batch = 1000
2140 offset = 0
2141 while True:
2142 page = await store.list_vectors(bank_id, offset=offset, limit=batch)
2143 memory_count += len(page)
2144 offset += len(page)
2145 if len(page) < batch:
2146 break
2147 except Exception:
2148 logger.debug(
2149 "list_vectors failed or unsupported; bank_health memory_count=0",
2150 exc_info=True,
2151 )
2152 memory_count = 0
2153 return compute_bank_health(bank_id, counters, memory_count)
2155 async def all_bank_health(self) -> list["BankHealth"]:
2156 """Compute health for all banks that have recorded operations."""
2157 results = []
2158 for bid in self._analytics.bank_ids():
2159 results.append(await self.bank_health(bid))
2160 return results
2162 def bank_quality_snapshot(self, bank_id: str) -> "QualityDataPoint":
2163 """Return a QualityDataPoint snapshot of current counters for trend tracking."""
2164 counters = self._analytics.get_counters(bank_id)
2165 return counters_to_quality_point(counters)
2167 # ---------------------------------------------------------------------------
2168 # Memory portability
2169 # ---------------------------------------------------------------------------
2171 async def export_bank(
2172 self,
2173 bank_id: str,
2174 path: str,
2175 *,
2176 include_embeddings: bool = False,
2177 include_entities: bool = True,
2178 allowed_roots: list[str] | None = None,
2179 allow_uncontained: bool = False,
2180 context: AstrocyteContext | None = None,
2181 ) -> int:
2182 """Export a memory bank to AMA (Astrocyte Memory Archive) JSONL format.
2184 ``allowed_roots`` / ``allow_uncontained`` propagate to the path
2185 containment check in ``astrocyte.portability``. By default
2186 (no roots configured and ``allow_uncontained=False``), the
2187 export refuses to run — set ``ASTROCYTE_PORTABILITY_ROOTS``,
2188 pass ``allowed_roots=[<dir>]``, or set ``allow_uncontained=True``.
2190 Memories are enumerated with the pipeline vector store's paged
2191 ``list_vectors`` whenever retain writes there (a pipeline and no
2192 engine provider), so every live memory is exported. Engine
2193 providers expose no listing API; for them the export falls back
2194 to one capped ``query="*"`` recall, which is incomplete for banks
2195 larger than one batch (a warning is logged).
2197 Returns the number of memories exported.
2198 """
2199 from astrocyte.portability import export_bank as _export
2201 self._policy.check_access(bank_id, "admin", context)
2203 # Retain only lands in the pipeline's vector store when no engine
2204 # provider is set (see ProviderDispatcher.retain).
2205 list_fn = None
2206 if self._dispatcher.engine_provider is None and self._pipeline is not None:
2207 store = self._pipeline.vector_store
2208 if store is not None and hasattr(store, "list_vectors"):
2209 list_fn = store.list_vectors
2211 count = await _export(
2212 recall_fn=self._dispatcher.recall,
2213 list_fn=list_fn,
2214 bank_id=bank_id,
2215 path=path,
2216 provider_name=self._provider_name,
2217 include_embeddings=include_embeddings,
2218 include_entities=include_entities,
2219 allowed_roots=allowed_roots,
2220 allow_uncontained=allow_uncontained,
2221 )
2222 await self._hook_manager.fire("on_export", bank_id=bank_id, data={"memory_count": count, "path": path})
2223 return count
2225 async def export_okf_bundle(
2226 self,
2227 bank_id: str,
2228 path: str,
2229 *,
2230 scope: str | None = None,
2231 kind: str | None = None,
2232 include_memories: bool = False,
2233 max_memories: int | None = None,
2234 allowed_roots: list[str] | None = None,
2235 allow_uncontained: bool = False,
2236 context: AstrocyteContext | None = None,
2237 ) -> Any:
2238 """Export a bank's compiled wiki pages as an OKF v0.2 bundle.
2240 A directory of markdown files with YAML frontmatter, readable by any
2241 OKF consumer. This is a projection: nothing is written back, and the
2242 bundle is a point-in-time view of the current revision of each page.
2244 Only fields Astrocyte actually persists are emitted — see
2245 :mod:`astrocyte.okf` for the deliberate omissions (``verified``,
2246 ``status``), which are absent rather than invented.
2248 ``include_memories`` additionally exports raw memories as concepts under
2249 ``memory/``. This is off by default because it is one file per memory:
2250 a large bank produces a very large bundle, so pair it with
2251 ``max_memories`` when exploring. When lifecycle is enabled, each memory
2252 carries a derived ``stale_after`` computed from the same
2253 ``_created_at`` value that ``run_lifecycle`` would delete on, so a
2254 consumer sees the shelf life before the deletion happens.
2256 ``allowed_roots`` / ``allow_uncontained`` behave as in
2257 :meth:`export_bank`: by default the export refuses to run unless
2258 containment is configured.
2260 Returns an ``astrocyte.okf.ExportResult``.
2261 """
2262 from astrocyte.okf import export_wiki_bundle
2264 validate_bank_id(bank_id)
2265 # Parity with export_bank: taking a full copy of a bank's compiled
2266 # knowledge out of the system is an admin-level operation.
2267 self._policy.check_access(bank_id, "admin", context)
2269 if self._wiki_store is None:
2270 raise ConfigError(
2271 "brain.export_okf_bundle() requires a WikiStore. Call brain.set_wiki_store(store) before exporting."
2272 )
2274 pages = await self._wiki_store.list_pages(bank_id, scope=scope, kind=kind) # type: ignore[attr-defined]
2276 memories: list[Any] = []
2277 if include_memories:
2278 store = self._pipeline.vector_store if self._pipeline else None
2279 if store is None or not hasattr(store, "list_vectors"):
2280 raise ConfigError(
2281 "export_okf_bundle(include_memories=True) requires a pipeline with a "
2282 "vector store supporting list_vectors()."
2283 )
2284 offset = 0
2285 while True:
2286 batch = await store.list_vectors(bank_id, offset=offset, limit=200)
2287 if not batch:
2288 break
2289 memories.extend(batch)
2290 offset += len(batch)
2291 if max_memories is not None and len(memories) >= max_memories:
2292 memories = memories[:max_memories]
2293 break
2295 result = export_wiki_bundle(
2296 pages,
2297 bank_id=bank_id,
2298 path=path,
2299 memories=memories or None,
2300 lifecycle=self._config.lifecycle,
2301 allowed_roots=allowed_roots,
2302 allow_uncontained=allow_uncontained,
2303 )
2304 await self._hook_manager.fire(
2305 "on_export",
2306 bank_id=bank_id,
2307 data={"format": "okf", "concept_count": result.concept_count, "path": path},
2308 )
2309 return result
2311 async def import_bank(
2312 self,
2313 bank_id: str,
2314 path: str,
2315 *,
2316 on_conflict: str = "skip",
2317 allowed_roots: list[str] | None = None,
2318 allow_uncontained: bool = False,
2319 context: AstrocyteContext | None = None,
2320 progress_fn: Any = None,
2321 ) -> Any:
2322 """Import memories from an AMA JSONL file into a bank.
2324 See ``export_bank`` for ``allowed_roots`` / ``allow_uncontained`` semantics.
2326 Returns an ImportResult with imported/skipped/errors counts.
2327 """
2328 from astrocyte.portability import ImportResult
2329 from astrocyte.portability import import_bank as _import
2331 self._policy.check_access(bank_id, "admin", context)
2333 result: ImportResult = await _import(
2334 retain_fn=self._dispatcher.retain,
2335 bank_id=bank_id,
2336 path=path,
2337 on_conflict=on_conflict,
2338 progress_fn=progress_fn,
2339 allowed_roots=allowed_roots,
2340 allow_uncontained=allow_uncontained,
2341 )
2342 await self._hook_manager.fire(
2343 "on_import",
2344 bank_id=bank_id,
2345 data={
2346 "imported": result.imported,
2347 "skipped": result.skipped,
2348 "errors": result.errors,
2349 },
2350 )
2351 return result
2353 # ---------------------------------------------------------------------------
2354 # Team memory sync (team-memory.md §8)
2355 # ---------------------------------------------------------------------------
2357 def _sync_vector_store(self, capability: str) -> Any:
2358 """The pipeline's vector store, which sync reads and writes directly.
2360 Retain only lands there when no engine provider is set (see
2361 ``ProviderDispatcher.retain``), so a Tier 2 engine can't serve sync.
2362 """
2363 store = None
2364 if self._dispatcher.engine_provider is None and self._pipeline is not None:
2365 store = self._pipeline.vector_store
2366 if store is None:
2367 raise CapabilityNotSupported(self._provider_name, capability)
2368 return store
2370 async def list_changes(
2371 self,
2372 bank_id: str,
2373 *,
2374 cursor: str | None = None,
2375 limit: int = 100,
2376 settle_seconds: float = 0.0,
2377 context: AstrocyteContext | None = None,
2378 ) -> MemoryChangePage:
2379 """Read a bank's changes feed: every change to a synced row, in order.
2381 The read side of team-memory sync (``team-memory.md`` §8, G3; served
2382 as ``GET /v1/banks/{bank_id}/changes``). Returns up to ``limit``
2383 (1–1000) changes strictly after ``cursor``, ordered by
2384 ``(changed_at, id)``, where ``changed_at`` is a row's last change to
2385 any synced field. A live entry carries the row's current values, so a
2386 row changed again later reappears as an upsert of the same id after
2387 the cursor. Pass the returned ``next_cursor`` back to
2388 resume; ``has_more`` says whether to fetch again now. A forgotten
2389 memory appears as a tombstone (``deleted=True``) carrying only its
2390 id and ``changed_at``, so mirrors can erase it.
2392 ``settle_seconds`` holds back changes younger than that. A row's
2393 ``changed_at`` is stamped before its write commits, so a slow commit
2394 can land behind a cursor that has already moved past its timestamp,
2395 and a puller would never see it. Withholding the newest few seconds
2396 lets in-flight writes commit first; the gateway sets it (default 5 s).
2398 Requires ``read`` on the bank. Raises :class:`ValueError` for a
2399 ``limit`` out of range, :class:`~astrocyte.errors.InvalidCursor` (a
2400 ``ValueError``) for a cursor this server did not issue, and
2401 :class:`CapabilityNotSupported` when the vector store has no
2402 ``list_changes`` (or retain goes to an engine provider).
2403 """
2404 from astrocyte._sync import MAX_CHANGES_LIMIT, decode_cursor, encode_cursor
2406 validate_bank_id(bank_id)
2407 if not 1 <= limit <= MAX_CHANGES_LIMIT:
2408 raise ValueError(f"limit must be between 1 and {MAX_CHANGES_LIMIT}")
2409 self._policy.check_access(bank_id, "read", context)
2410 after = decode_cursor(cursor) if cursor else None
2411 list_fn = getattr(self._sync_vector_store("list_changes"), "list_changes", None)
2412 if list_fn is None:
2413 raise CapabilityNotSupported(self._provider_name, "list_changes")
2414 # One extra row answers "is there more?" without a second query.
2415 rows = await list_fn(bank_id, after=after, limit=limit + 1)
2416 if settle_seconds > 0:
2417 horizon = datetime.now(timezone.utc) - timedelta(seconds=settle_seconds)
2418 # Rows come in changed_at order, so the settled ones are a prefix.
2419 settled = 0
2420 while settled < len(rows) and rows[settled].changed_at <= horizon:
2421 settled += 1
2422 rows = rows[:settled]
2423 changes, has_more = rows[:limit], len(rows) > limit
2424 if changes:
2425 next_cursor: str | None = encode_cursor(changes[-1].changed_at, changes[-1].id)
2426 else:
2427 next_cursor = cursor
2428 return MemoryChangePage(changes=changes, next_cursor=next_cursor, has_more=has_more)
2430 async def erase(
2431 self,
2432 bank_id: str,
2433 memory_ids: list[str],
2434 *,
2435 context: AstrocyteContext | None = None,
2436 reason: str | None = None,
2437 ) -> int:
2438 """Erase forgotten memories from storage for good, keeping their tombstones.
2440 ``forget`` is a soft delete on the SQL stores: a forgotten memory's
2441 text stays on disk so ``as_of`` recall can still see the past. Erase
2442 removes it, with its embedding, metadata and what was derived from it
2443 (``VectorStore.erase``), and keeps only the tombstone, so the changes
2444 feed still reports it deleted (mirrors erase it too) and the id is
2445 never stored again. Only memories already forgotten are erased: forget
2446 first. Requires ``forget`` on the bank; logged for the audit trail.
2447 Returns how many were erased. Raises ``CapabilityNotSupported`` when
2448 the vector store has no ``erase``.
2449 """
2450 validate_bank_id(bank_id)
2451 self._policy.check_access(bank_id, "forget", context)
2452 store = self._sync_vector_store("erase")
2453 erase = getattr(store, "erase", None)
2454 if erase is None:
2455 raise CapabilityNotSupported(self._provider_name, "erase")
2456 ids = list(dict.fromkeys(memory_ids))
2457 if not ids:
2458 return 0
2459 erased = await erase(bank_id, ids)
2460 self._logger.log(
2461 "astrocyte.erase",
2462 bank_id=bank_id,
2463 data={"actor": context_principal_label(context) if context else "anonymous",
2464 "reason": reason or "erase", "requested": len(ids), "erased": erased},
2465 level=logging.WARNING,
2466 )
2467 return erased
2469 async def forget_principal(
2470 self,
2471 principal: str,
2472 *,
2473 bank_prefix: str,
2474 context: AstrocyteContext | None = None,
2475 reason: str | None = None,
2476 ) -> list[tuple[str, int, int | None]]:
2477 """Right to erasure for one principal: forget, then erase, what they saved.
2479 In every bank whose id starts with ``bank_prefix`` (the configured
2480 banks, and every bank the vector store holds when it can list them),
2481 the live memories saved by ``principal`` (``_actor``, stamped by an
2482 authenticated retain or push) or tagged ``principal:<principal>`` (the
2483 older convention) are forgotten with ``compliance=True`` (legal holds
2484 do not block an erasure), then erased from storage where the store
2485 can (:meth:`erase`). Returns ``(bank_id, forgotten, erased)`` for
2486 every bank swept; ``erased`` is None where the store can't erase.
2487 Behind an engine provider only the tag is matched. Memories
2488 forgotten before (soft-deleted) are no longer listed, so they are not
2489 found here.
2490 """
2491 if not bank_prefix:
2492 raise ValueError("bank_prefix is required: an erasure is scoped to one tenant's banks")
2493 tag = f"principal:{principal}"
2494 why = reason or f"erasure for {principal}"
2495 banks = {b for b in (self._config.banks or {}) if b.startswith(bank_prefix)}
2496 try:
2497 store = self._sync_vector_store("forget_principal")
2498 except CapabilityNotSupported:
2499 # An engine provider: nothing to list or erase here, so the tag
2500 # convention is all that can be matched (the engine forgets by tag).
2501 out: list[tuple[str, int, int | None]] = []
2502 for bank_id in sorted(banks):
2503 result = await self.forget(bank_id, tags=[tag], compliance=True, reason=why, context=context)
2504 out.append((bank_id, result.deleted_count, None))
2505 return out
2506 lister = getattr(store, "list_banks", None)
2507 if lister is not None:
2508 banks |= {row[0] for row in await lister() if row[0].startswith(bank_prefix)}
2509 out = []
2510 for bank_id in sorted(banks):
2511 ids, offset = [], 0
2512 while True:
2513 page = await store.list_vectors(bank_id, offset=offset, limit=500)
2514 ids += [i.id for i in page if (i.metadata or {}).get("_actor") == principal or tag in (i.tags or ())]
2515 if len(page) < 500:
2516 break
2517 offset += 500
2518 if not ids:
2519 out.append((bank_id, 0, 0 if getattr(store, "erase", None) else None))
2520 continue
2521 result = await self.forget(bank_id, memory_ids=ids, compliance=True, reason=why, context=context)
2522 try:
2523 erased: int | None = await self.erase(bank_id, ids, context=context, reason=why)
2524 except CapabilityNotSupported:
2525 erased = None
2526 out.append((bank_id, result.deleted_count, erased))
2527 return out
2529 async def push_records(
2530 self,
2531 bank_id: str,
2532 records: list[SyncPushRecord],
2533 *,
2534 context: AstrocyteContext | None = None,
2535 ) -> list[SyncPushResult]:
2536 """Store memories pushed by a team-memory client, each as one row with its own id.
2538 The write side of team-memory sync (``team-memory.md`` §8, G2; served
2539 as ``POST /v1/banks/{bank_id}/sync/push``). Each record goes through
2540 the policy layer as a retain does (input limits, content validation,
2541 the PII barrier, metadata sanitization, authoritative ``_actor``),
2542 then becomes exactly one row with the client's id: no chunking, no
2543 extraction, no LLM call; the text is re-embedded with this server's
2544 model. One :class:`SyncPushResult` per record, in order:
2546 - ``stored`` — new row.
2547 - ``unchanged`` — the id already holds this text in this bank (an
2548 idempotent re-push). Text is immutable but other fields are not:
2549 differing metadata is **not** applied (``reason`` says so); those
2550 fields get their own update path.
2551 - ``duplicate`` — near-duplicate of the bank's memory ``duplicate_of``
2552 (the retain dedup check and threshold); nothing stored.
2553 - ``rejected`` — ``reason`` says why: the id holds other text, here or
2554 in **another bank** (never overwritten, moved or revealed), the id
2555 was forgotten here (a forget is not undone by a re-push), or the
2556 policy layer refused the record.
2558 Requires ``write`` on the bank. Rate limits and quotas are checked
2559 once for the request (a push counts as one call) and quota usage is
2560 recorded per stored record. Underscore-prefixed metadata keys other
2561 than ``_created_at``, ``_retain_id``, ``_chunk_index`` are dropped.
2562 Raises :class:`ValueError` for more than 100 records or an id outside
2563 ``[A-Za-z0-9_-]{8,64}``, and :class:`CapabilityNotSupported` when the
2564 vector store lacks ``lookup_ids`` / ``insert_vectors`` (or retain goes
2565 to an engine provider).
2566 """
2567 from astrocyte._sync import MAX_PUSH_RECORDS, PUSH_ID_RE
2568 from astrocyte.identity import format_principal, resolve_actor
2570 validate_bank_id(bank_id)
2571 if len(records) > MAX_PUSH_RECORDS:
2572 raise ValueError(f"at most {MAX_PUSH_RECORDS} records per push ({len(records)} given)")
2573 bad = [r.id for r in records if not isinstance(r.id, str) or not PUSH_ID_RE.fullmatch(r.id)]
2574 if bad:
2575 raise ValueError(f"invalid record id(s) {bad[:5]!r}: expected 8-64 of [A-Za-z0-9_-]")
2576 self._policy.check_access(bank_id, "write", context)
2577 store = self._sync_vector_store("sync_push")
2578 if getattr(store, "lookup_ids", None) is None or getattr(store, "insert_vectors", None) is None:
2579 raise CapabilityNotSupported(self._provider_name, "sync_push")
2580 if not records:
2581 return []
2583 with span("astrocyte.sync_push", {"astrocyte.bank_id": bank_id}):
2584 self._policy.check_rate_and_quota(bank_id, "retain")
2585 noisy = self._policy.check_noisy_bank(bank_id)
2586 if noisy is not None:
2587 refusal = self._apply_noisy_bank_verdict(bank_id, noisy)
2588 if refusal is not None:
2589 return [SyncPushResult(id=r.id, status="rejected", reason=refusal.error) for r in records]
2591 actor_identity = resolve_actor(context) if context else None
2592 actor = format_principal(actor_identity) if actor_identity is not None else None
2593 results: list[SyncPushResult | None] = [None] * len(records)
2594 prepared: list[tuple[int, SyncPushRecord]] = []
2595 for i, record in enumerate(records):
2596 outcome = await self._prepare_push_record(bank_id, record, actor)
2597 if isinstance(outcome, SyncPushResult):
2598 results[i] = outcome
2599 else:
2600 prepared.append((i, outcome))
2602 if prepared:
2603 try:
2604 self._policy.check_circuit(self._provider_name)
2605 pushed = await self._pipeline.push_records(bank_id, [r for _, r in prepared]) # type: ignore[union-attr]
2606 self._policy.record_success()
2607 except ProviderUnavailable:
2608 raise
2609 except Exception:
2610 self._policy.record_failure()
2611 raise
2612 for (i, record), result in zip(prepared, pushed):
2613 results[i] = result
2614 await self._account_pushed(bank_id, record, result)
2615 if self._compile_queue is not None and any(r.status == "stored" for r in pushed):
2616 self._compile_queue.notify_retain(bank_id) # type: ignore[union-attr]
2617 return [r for r in results if r is not None]
2619 async def _prepare_push_record(
2620 self, bank_id: str, record: SyncPushRecord, actor: str | None
2621 ) -> SyncPushRecord | SyncPushResult:
2622 """Run one pushed record through the retain policy layer: the record
2623 to store (text possibly redacted, metadata sanitized and stamped), or
2624 its rejection."""
2625 from astrocyte._sync import CONTENT_HASH_RE, PUSH_SYSTEM_METADATA_KEYS, content_hash
2627 def rejected(reason: str) -> SyncPushResult:
2628 return SyncPushResult(id=record.id, status="rejected", reason=reason)
2630 text = record.text
2631 if not isinstance(text, str) or not text.strip():
2632 return rejected("text is empty")
2633 input_error = self._policy.validate_retain_input(text, record.tags, bank_id=bank_id)
2634 if input_error:
2635 return rejected(input_error)
2636 if record.content_hash is not None:
2637 if not CONTENT_HASH_RE.fullmatch(record.content_hash):
2638 return rejected("content_hash must be sha256:<64 lowercase hex>")
2639 if record.content_hash != content_hash(text):
2640 return rejected("content_hash does not match text")
2641 errors = self._policy.validate_content(text, "text", bank_id=bank_id)
2642 if errors:
2643 return rejected("; ".join(errors))
2644 try:
2645 text, pii_matches = await self._policy.scan_pii(text, bank_id=bank_id)
2646 except PiiRejected as exc:
2647 await self._note_pii(bank_id, exc.pii_types, "reject", "sync_push")
2648 return rejected(f"PII detected ({', '.join(exc.pii_types)})")
2649 if pii_matches:
2650 await self._note_pii(
2651 bank_id, [m.pii_type for m in pii_matches], self._policy.pii_action_for(bank_id), "sync_push"
2652 )
2654 metadata = {
2655 k: v for k, v in (record.metadata or {}).items() if not k.startswith("_") or k in PUSH_SYSTEM_METADATA_KEYS
2656 }
2657 if actor is not None:
2658 # Authoritative provenance, as in retain: the caller's own claim loses.
2659 metadata["_actor"] = actor
2660 sanitized, _warnings = self._policy.sanitize_metadata(metadata or None, bank_id=bank_id)
2661 return SyncPushRecord(
2662 id=record.id,
2663 text=text,
2664 occurred_at=record.occurred_at,
2665 tags=list(record.tags) if record.tags else None,
2666 fact_type=record.fact_type,
2667 metadata=sanitized,
2668 )
2670 async def _note_pii(self, bank_id: str, pii_types: list[str], action: str, operation: str) -> None:
2671 """Log, count and announce PII the barrier found (as retain does)."""
2672 types = ",".join(pii_types)
2673 self._logger.log(
2674 "astrocyte.policy.pii_detected",
2675 bank_id=bank_id,
2676 operation=operation,
2677 data={"pii_types": types, "action": action},
2678 )
2679 self._metrics.inc_counter("astrocyte_pii_detected_total", {"bank_id": bank_id, "action": action})
2680 await self._hook_manager.fire("on_pii_detected", bank_id=bank_id, data={"pii_types": types, "action": action})
2682 async def _account_pushed(self, bank_id: str, record: SyncPushRecord, result: SyncPushResult) -> None:
2683 """Quota, analytics, noisy-bank signal, metrics and hooks for one pushed record."""
2684 if result.status not in ("stored", "duplicate"):
2685 return
2686 deduplicated = result.status == "duplicate"
2687 self._analytics.record_retain(bank_id, len(record.text), deduplicated=deduplicated)
2688 self._policy.record_retain_signal(bank_id, len(record.text), deduplicated)
2689 if deduplicated:
2690 return
2691 self._policy.record_quota(bank_id, "retain")
2692 self._metrics.inc_counter(
2693 "astrocyte_retain_total", {"bank_id": bank_id, "provider": self._provider_name, "status": "ok"}
2694 )
2695 await self._hook_manager.fire(
2696 "on_retain",
2697 bank_id=bank_id,
2698 data={"memory_id": record.id, "content_length": len(record.text), "source": "sync_push"},
2699 )
2701 # ---------------------------------------------------------------------------
2702 # Internal routing
2703 # ---------------------------------------------------------------------------
2705 @property
2706 def _provider_name(self) -> str:
2707 return self._config.provider or "pipeline"
2709 # ---------------------------------------------------------------------------
2710 # Backwards-compat thin wrappers (tests access these directly)
2711 # ---------------------------------------------------------------------------
2713 @property
2714 def _tiered_retriever(self):
2715 """Expose tiered retriever for testing/introspection."""
2716 return self._dispatcher.tiered_retriever
2718 async def _do_retain(self, request: RetainRequest) -> RetainResult:
2719 return await self._dispatcher.retain(request)
2721 async def _do_recall(self, request: RecallRequest) -> RecallResult:
2722 return await self._dispatcher.recall(request)
2724 async def _do_forget(self, request: ForgetRequest) -> ForgetResult:
2725 return await self._dispatcher.forget(request)