symfonic.cli.native_hms

native_hms

One durable, tenant-scoped CLI HMS turn through the public Agent.

The pool belongs to this invocation. A later CLI process opens the same DSN and recalls published rows from Postgres; no process-local store is involved.

NativeHmsOutcome dataclass

NativeHmsOutcome(text: str, admitted_contribution_tokens: int, recalled_records_count: int, stage_counts: dict[str, dict[str, int]] = dict(), stages: dict[str, NativeStageEvidence] = dict())

Public reply and counts taken from the stages that performed the turn.

admitted_contribution_tokens is the prompt compiler's estimate of its admitted contribution set, not provider token usage or legacy system tokens. recalled_records_count is retrieval's admitted record count, not a post-turn database inventory or legacy memory metric.

NativeStageEvidence dataclass

NativeStageEvidence(outcome: str, reason: str, counts: dict[str, int])

Count-only public stage record, including an extraction refusal reason.

run_native_hms_turn async

run_native_hms_turn(*, query: str, tenant_id: str, dsn: str, domain: str, provider: Any, model: str | ModelConfig | None = None, retrieval_min_relevance: float = 0.0) -> NativeHmsOutcome

Run one HMS turn and close its Postgres pool after lifecycle flush.

A DSN is mandatory: a one-shot CLI process cannot promise recall across processes from an in-memory backend. Scope validation precedes I/O. The native relevance floor is a per-app choice over scored recall. Its neutral default awaits calibration; legacy raw-score 0.2 is not equivalent.

Source code in src/symfonic/cli/native_hms.py

async def run_native_hms_turn(
    *,
    query: str,
    tenant_id: str,
    dsn: str,
    domain: str,
    provider: Any,
    model: str | ModelConfig | None = None,
    retrieval_min_relevance: float = 0.0,
) -> NativeHmsOutcome:
    """Run one HMS turn and close its Postgres pool after lifecycle flush.

    A DSN is mandatory: a one-shot CLI process cannot promise recall across
    processes from an in-memory backend. Scope validation precedes I/O.
    The native relevance floor is a per-app choice over scored recall. Its
    neutral default awaits calibration; legacy raw-score 0.2 is not equivalent.
    """
    if not isinstance(dsn, str) or not dsn.strip():
        raise ValueError("Native HMS requires an explicit Postgres DSN")
    scope = MemoryScope(tenant_id)
    model_config = (
        provider_default_config(provider)
        if model is None
        else ModelConfig(model_name=model)
        if isinstance(model, str)
        else model
    )
    retrieval_policy = RetrievalPolicy(min_relevance=retrieval_min_relevance)
    if not isinstance(model_config, ModelConfig):
        raise TypeError("model must be a model name, ModelConfig, or None")
    state = build_hms_prompt_state(
        domain=DomainTemplate(name=domain),
        tenant=tenant_id,
        tool_names=(),
        domain_description="",
    )
    async with PostgresPoolManager(dsn=dsn) as pool:
        graph = PostgresGraphBackend(pool)
        await graph.ensure_schema()
        store = GraphBackedHms(graph)
        capabilities = [
            *memory_capabilities(
                store,
                scope,
                retrieval_policy=retrieval_policy,
                extractor=MemoryExtractionService(_ExtractionModel(provider, model_config)),
            ),
            PromptingCapability(
                sources=[
                    hms_contribution(HmsSystemSource(state=state)),
                    extraction_contribution(ExtractionDirectiveSource(scope_path=scope.path)),
                ]
            ),
        ]
        stage_counts: dict[str, dict[str, int]] = {}
        stages: dict[str, NativeStageEvidence] = {}
        answer: str | None = None
        async with Agent(provider, model=model, capabilities=capabilities) as agent:
            async for event in agent.stream(query):
                if event.kind == "stage" and event.stage is not None:
                    stage = event.stage
                    counts = dict(stage.counts)
                    stage_counts[stage.stage_id] = counts
                    stages[stage.stage_id] = NativeStageEvidence(
                        outcome=stage.outcome,
                        reason=stage.reason,
                        counts=counts,
                    )
                elif event.kind == "done" and event.result is not None:
                    answer = event.result.text
        if answer is None:
            raise RuntimeError("Native HMS turn ended without a result")
        # Memory side-effect failures are reported as no-change by the
        # capability so an ordinary Agent turn can still answer. This CLI
        # promises durable HMS; it must not exit successfully on those cases.
        for stage_id in (RETRIEVAL_STAGE, "memory.consolidation", "memory.lifecycle"):
            stage = stages.get(stage_id)
            if stage is None:
                raise RuntimeError(f"Native HMS stage did not run: {stage_id}")
            if stage.reason.startswith(EXTRACTION_DEGRADED) or (
                "memory store was unreachable" in stage.reason
            ):
                raise RuntimeError(f"Native HMS stage degraded: {stage_id}")
        prompt = stage_counts.get(PROMPTING_STAGE, {})
        recall = stage_counts.get(RETRIEVAL_STAGE, {})
        return NativeHmsOutcome(
            text=answer,
            admitted_contribution_tokens=prompt.get("admitted_contribution_tokens", 0),
            recalled_records_count=recall.get("admitted", 0),
            stage_counts=stage_counts,
            stages=stages,
        )