Level 5: Production Deployment

Deploy symfonic with offline consolidation, spreading activation, graph traversal, hybrid backends, and full observability. This level covers the complete production topology.

Prerequisites

• Completed Levels 1-4
• Docker and Docker Compose
• Familiarity with Celery (optional, for consolidation scheduling)

Architecture Overview

                          +-------------------+
                          |   FastAPI / SSE   |
                          |  create_agent_    |
                          |    router()       |
                          +--------+----------+
                                   |
                          +--------v----------+
                          |  SymfonicAgent    |
                          |  - run()          |
                          |  - stream_typed() |
                          |  - stream_text()  |
                          |  - load_plugin()  |
                          +--------+----------+
                                   |
              +--------------------+--------------------+
              |                    |                     |
    +---------v-------+  +--------v--------+  +---------v--------+
    | AgentRuntime    |  | MemoryOrch.     |  | Plugins          |
    | (LangGraph)     |  | (5-layer HMS)   |  | (prompt+guards)  |
    +---------+-------+  +--------+--------+  +------------------+
              |                    |
              |          +---------+---------+
              |          |                   |
         +----v----+  +--v---------+  +------v------+
         | LLM     |  | Postgres   |  | Vector      |
         | Provider|  | Graph      |  | (pgvector)  |
         +---------+  +-----+------+  +------+------+
                            |                |
                      +-----v----------------v-----+
                      |  PostgresPoolManager       |
                      |  (shared connection pool)  |
                      +----------------------------+

Docker Compose: Full Stack

# docker-compose.yml
version: "3.9"
services:
  postgres:
    image: pgvector/pgvector:pg16
    environment:
      POSTGRES_USER: symfonic
      POSTGRES_PASSWORD: symfonic_dev
      POSTGRES_DB: symfonic
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  celery-worker:
    build: .
    command: celery -A my_app.celery worker --loglevel=info
    environment:
      POSTGRES_DSN: postgresql://symfonic:symfonic_dev@postgres:5432/symfonic
      REDIS_URL: redis://redis:6379/0
    depends_on:
      - postgres
      - redis

  celery-beat:
    build: .
    command: celery -A my_app.celery beat --loglevel=info
    environment:
      POSTGRES_DSN: postgresql://symfonic:symfonic_dev@postgres:5432/symfonic
      REDIS_URL: redis://redis:6379/0
    depends_on:
      - postgres
      - redis

  app:
    build: .
    command: uvicorn my_app:app --host 0.0.0.0 --port 8000
    ports:
      - "8000:8000"
    environment:
      POSTGRES_DSN: postgresql://symfonic:symfonic_dev@postgres:5432/symfonic
      REDIS_URL: redis://redis:6379/0
      ANTHROPIC_API_KEY: ${ANTHROPIC_API_KEY}
    depends_on:
      - postgres
      - redis

volumes:
  pgdata:

SleepConsolidator: Offline Memory Maintenance

SleepConsolidator runs 11 distinct phases of memory maintenance designed for nightly batch execution. Three entrypoints:

• quick_nap(scope) — lightweight subset (Phase 1 strengthen + optional Phase 5 SOUL + optional Phase 10 episodic). Fires every N turns when FrameworkConfig.quick_nap_interval > 0.
• run(scope) — full roster, returns a ConsolidationReport with phases_run: list[str] listing every phase that fired.
• nightly_nap(scope) — v7.0 T12 full-roster alias gated by FrameworkConfig.nightly_nap_enabled=True. Operators wire their own scheduler; engine does not parse the cron string.

Phase	Name	Effect
1	strengthen	Boost importance for recurring nodes (combines access_count + phase1_spreading_weight * spreading_access_count)
1.5	cooccurrence	Create CO_OCCURRED edges between co-temporal nodes
2	tag_risk	Mark nodes with negative-sentiment context
2.5	pending_edges	Write inferred pending_connections as real edges (conditional)
3	prune_orphans	Delete stale low-confidence orphan nodes; TTL-expired pruned unconditionally
4	meta_nodes	Cluster raw domain nodes by label prefix (excludes existing META: to prevent cascade)
5	soul_corrections	Apply SOUL schema updates (conditional on soul_schema= arg)
8	working_ttl	Prune WORKING-layer nodes with expired TTL
9	decay_importance	Reduce importance of stale nodes; skips importance >= 8.0 and SOUL: / AGENT_IDENTITY: labels
10	episodic_summary	Compress old episodes into semantic meta-nodes (gated by episodic_summarization_max_entries)
11	synthetic_links	Emit SYNTHETIC_LINK edges between semantic nodes that co-occur in episodic content (gated by synthetic_link_min_co_count)
12	procedural_promotion	Promote recurring episodic patterns to draft procedural skills (gated by promotion_min_pattern_count)

Running Consolidation Manually

import asyncio
from symfonic.core.learning.consolidation import SleepConsolidator
from symfonic.memory.graph.store import GraphMemoryStore
from symfonic.memory.types import TenantScope

async def consolidate(graph_store: GraphMemoryStore, tenant_id: str):
    consolidator = SleepConsolidator(
        graph_store=graph_store,
        lookback_hours=24,
    )
    scope = TenantScope(tenant_id=tenant_id)
    report = await consolidator.run(scope)

    print(f"Phases run: {report.phases_run}")
    print(f"Strengthened: {report.nodes_strengthened}")
    print(f"Pruned: {report.nodes_pruned}")
    print(f"Meta-nodes created: {report.meta_nodes_created}")
    print(f"SOUL updates: {report.soul_updates}")
    print(f"Working TTL pruned: {report.working_ttl_pruned}")
    print(f"Nodes decayed: {report.nodes_decayed}")
    print(f"Episodes summarized: {report.episodes_summarized}")
    print(f"Draft skills promoted: {report.draft_skills_promoted}")
    return report

report.phases_run lists every phase that actually executed (including zero-mutation phases). Use it to distinguish "phase ran and found nothing" from "phase was short-circuited".

Scheduling with Celery Beat

Use the CeleryTaskProvider adapter to schedule consolidation as a recurring background task:

from celery import Celery
from examples.full_demo.infra.celery_adapter import CeleryTaskProvider

celery_app = Celery("symfonic", broker="redis://localhost:6379/0")

# Register the consolidation task
@celery_app.task(name="execute_skill")
def execute_skill(skill_id: str, **kwargs):
    """Universal skill dispatcher."""
    if skill_id == "deep_sleep_consolidation":
        import asyncio
        from symfonic.core.learning.consolidation import SleepConsolidator

        tenant_id = kwargs.get("tenant_id", "default")
        # graph_store initialized from app context
        consolidator = SleepConsolidator(graph_store=graph_store, lookback_hours=24)
        scope = TenantScope(tenant_id=tenant_id)
        report = asyncio.run(consolidator.run(scope))
        return report.to_dict()

# Schedule via CeleryTaskProvider
async def setup_scheduling():
    provider = CeleryTaskProvider(celery_app)
    schedule_id = await provider.schedule_recurring(
        task_name="deep_sleep_consolidation",
        interval_seconds=86400,  # daily
        params={"tenant_id": "acme-corp"},
    )
    print(f"Scheduled: {schedule_id}")

The InProcessScheduler is available for single-process deployments without Celery:

from symfonic.agent.scheduler import InProcessScheduler

scheduler = InProcessScheduler()
config = FrameworkConfig(scheduler=scheduler)
agent = SymfonicAgent(model_provider=provider, config=config)

Spreading Activation and ActivationEvent

When spreading_activation=True, the agent automatically fetches 1st-degree neighbor nodes when semantic retrieval produces hits. This gives the LLM richer context without explicit queries.

Each activated node emits an ActivationEvent during streaming:

from symfonic.core.contracts.types import ActivationEvent

async for event in agent.stream_typed(query, scope=scope):
    if isinstance(event, ActivationEvent):
        print(
            f"Activated: {event.node_label} "
            f"(score={event.score:.2f}, layer={event.layer})"
        )
        if event.source_node_id:
            print(f"  triggered by: {event.source_node_id}")

ActivationEvent fields:

Field	Type	Description
node_id	str	Unique node identifier
node_label	str	Human-readable label (e.g. "SOUL: brand_voice")
score	float	Activation score 0.0 to 1.0
layer	str	Memory layer (semantic, working, episodic, etc.)
source_node_id	str | None	Parent node that triggered activation

Graph Traversal API

The GraphTraversal class provides BFS and shortest-path queries over the memory graph:

from symfonic.memory.graph.traversal import GraphTraversal
from symfonic.memory.types import TenantScope

graph_store = agent.orchestrator.graph_store
traversal = GraphTraversal(graph_store)
scope = TenantScope(tenant_id="acme-corp")

# BFS: find all nodes within 2 hops of a starting node
neighborhood = await traversal.bfs(scope, start="node-123", max_depth=2)
for node in neighborhood:
    print(f"  {node.label}: {node.content[:60]}")

# Shortest path between two nodes
path = await traversal.shortest_path(scope, source="node-A", target="node-B")
if path:
    print(f"Path: {' -> '.join(str(n) for n in path)}")
else:
    print("No path found")

Hybrid Backends: Per-Layer Storage

Route different memory layers to different backends using HybridBackendConfig. For example, episodic data to a vector-optimized store while semantic data stays in Postgres:

from symfonic.memory.orchestrator.factory import HybridBackendConfig, OrchestratorFactory
from symfonic.memory.types import MemoryLayer

hybrid_config = HybridBackendConfig(
    layer_graph_backends={
        MemoryLayer.SEMANTIC: postgres_graph,
        MemoryLayer.WORKING: in_memory_graph,
    },
    layer_vector_backends={
        MemoryLayer.EPISODIC: postgres_vector,
    },
)

# Build orchestrator with hybrid routing
orchestrator = OrchestratorFactory.build_hybrid(
    config=orchestrator_config,
    default_graph_backend=postgres_graph,
    default_vector_backend=postgres_vector,
    embedding_provider=embedding_provider,
    hybrid_config=hybrid_config,
)

agent = SymfonicAgent(
    model_provider=provider,
    orchestrator=orchestrator,
    config=FrameworkConfig(),
)

Multi-Tenant Isolation

Every operation in symfonic requires a TenantScope. This is enforced at the backend level -- there is no way to accidentally leak data between tenants.

from symfonic.agent.types import FrameworkTenantScope

# Full scope with sub-tenant and namespace
scope = FrameworkTenantScope(
    tenant_id="acme-corp",
    sub_tenant_id="west-coast",
    namespace="west-coast",       # must match sub_tenant_id if both set
)

HTTP headers for the FastAPI bridge:

Header	Required	Maps to
X-Tenant-ID	Yes	scope.tenant_id
X-Sub-Tenant-ID	No	scope.sub_tenant_id
X-Namespace	No	scope.namespace

Production Hardening: v6.1.9+ Security Knobs

Production auth fail-fast

create_agent_router() raises RuntimeError at build time when SYMFONIC_ENV=production (or APP_ENV / ENVIRONMENT / NODE_ENV set to production / prod) and no tenant-auth verifier was registered. Emergency override: ALLOW_INSECURE_PROD=true — logs CRITICAL and data will leak across tenants if left enabled.

from symfonic.agent.fastapi.dependencies import install_scope_resolver
from symfonic.agent.fastapi.router import create_agent_router
from symfonic.platform import (
    AuthenticationError, AuthorizationError, HeaderScopeResolver,
)

async def my_verifier(credentials, tenant_id):
    """Return principal facts, or raise to deny.

    ``credentials`` is a ``RequestCredentials`` -- headers plus connection
    facts. Raise ``AuthenticationError`` for 401, ``AuthorizationError``
    for 403; the framework maps both.
    """
    token = (credentials.header("authorization") or "").removeprefix("Bearer ")
    if not token:
        raise AuthenticationError("Missing Authorization header")
    user = my_decode(token)
    if not my_is_member(user, tenant_id):
        raise AuthorizationError("Not a member of the requested tenant")
    return {"principal_id": user.id, "is_admin": user.role == "admin"}

install_scope_resolver(app, HeaderScopeResolver(verifier=my_verifier))
router = create_agent_router(agent, prefix="/api/v1", app=app)

Both calls matter. The resolver lives on app.state, so passing app=app is what lets the production gate see it — without it the gate finds no auth of either kind and refuses to boot, which is the behaviour you want if the install call is ever dropped.

The gate also runs again, per request, against the app that actually serves it. Mounting the router on a second app therefore cannot serve traffic that app never attested to: it is refused with a 500, and the operator gets the specific reason in a CRITICAL log. Passing the right app= is still what keeps the failure at startup rather than on the first request.

This is what symfonic init now generates. Identity no longer arrives on request.state: the verifier returns facts, the framework publishes the derived principal, and symfonic.agent.fastapi.identity.request_user_id / request_is_admin is where your middleware, rate limiter and audit hooks read it. Reading request.state.user_id directly returns None on this path without raising, so a migration that forgets this bucket-keys authenticated traffic by a spoofable header and nothing fails.

The legacy process-global form (still supported)

from symfonic.agent.fastapi.dependencies import set_tenant_auth_verifier
from symfonic.agent.fastapi.router import create_agent_router

set_tenant_auth_verifier(my_jwt_verifier)  # MUST be called before router build
router = create_agent_router(agent, prefix="/api/v1")

A process global holding a per-deployment fact: two apps in one process share one verifier, and whichever starts second silently redefines authorisation for the first. The two verifier protocols are **not** interchangeable — the legacy one takes `(Request, tenant_id)` and denies by raising `HTTPException` — so moving across is a rewrite, not a renamed import.

Credential scrubbing (three-tier defence-in-depth)

Three layers scrub credential-shaped property keys on every memory op:

1. Extraction-time — HMS / JIT / extraction prompts instruct the LLM to omit credential-shaped properties.
2. Write-time — SymfonicAgent._consolidate scrubs before persist.
3. Read-time — SymfonicAgent._hydrate scrubs before formatting the context string.

Default pattern list: password, secret, token, api_key / api-key, credential, private_key / private-key, bearer, authorization, auth_header, access_key, refresh_token.

from symfonic.agent.config import FrameworkConfig

# Default: None = use built-in v6.1.9 list byte-for-byte
cfg = FrameworkConfig()

# Explicit opt-out (tenants that run their own upstream hygiene):
cfg = FrameworkConfig(credential_patterns=[])

# Custom list REPLACES built-in (invalid regex raises re.error):
cfg = FrameworkConfig(credential_patterns=[
    r"password", r"secret", r"auth_token", r"client_secret",
])

Public entrypoint for non-engine call sites: symfonic.agent.hygiene.scrub_credential_keys.

Intent filter + tool routing (v7.0 / v7.0.1)

For deployments with 10+ bound tools, narrow the per-turn tools= payload via tool_routing_mode. Safe rollout: start in observe mode to collect ToolRoutingDecisionEvent telemetry before flipping to enforce.

from symfonic.agent.config import FrameworkConfig
from symfonic.agent.domain import DomainTemplate

cfg = FrameworkConfig(
    intent_filter_mode="enforce",   # action turns skip semantic/episodic
    tool_routing_mode="enforce",    # narrow tools= payload per intent
    always_include_tools=("hydrate_context", "emit_telemetry"),
    domain=DomainTemplate(
        tool_trigger_keywords={
            "mysql_query": ["mysql", "query", "sql", "database"],
            "bash": ["bash", "shell", "command"],
        },
    ),
)

See Guide 07 for the full recipe.

Fabrication detection (v7.0 / v7.0.3)

In observe mode the detector emits telemetry but never mutates drafts — safe to enable in prod while tuning thresholds. Escalate to refuse once you trust the signal.

cfg = FrameworkConfig(
    fabrication_check_mode="refuse",
    fabrication_refuse_min_confidence=0.8,  # catches UUID (0.95) + kv/verb (0.8)
    # Lower to 0.5 to catch short-hex too; raise to 1.0 for observe-only
)

See Guide 08 for the deployment stance.

Self-Reflection: Insight Extraction

Enable post-interaction insight extraction to automatically create semantic graph nodes from conversations:

config = FrameworkConfig(
    self_reflection=True,
    reflection_prompt="""Analyze the conversation to extract insights.
For each insight, return a JSON array of objects with:
- "subject": entity name
- "relation": semantic link type
- "object": target entity or value
- "fact": 1-sentence summary
- "confidence": 0.0-1.0

Conversation:
{conversation}

Return ONLY a JSON array. If no insights found, return [].""",
)

The {conversation} placeholder is required. The InsightExtractor runs after each interaction and commits extracted facts to the semantic graph.

Monitoring and Observability

FrameworkObservabilityHook

from symfonic.agent.observability import FrameworkObservabilityHook

class MetricsCollector:
    """Collect framework metrics for Prometheus/Datadog/etc."""

    async def on_hydration_complete(self, tenant_id, query, entries_count, latency_ms):
        metrics.histogram("symfonic.hydration.latency", latency_ms, tags=[f"tenant:{tenant_id}"])
        metrics.gauge("symfonic.hydration.entries", entries_count, tags=[f"tenant:{tenant_id}"])

    async def on_consolidation_complete(self, tenant_id, operations_count, latency_ms):
        metrics.histogram("symfonic.consolidation.latency", latency_ms, tags=[f"tenant:{tenant_id}"])

    async def on_routing_complete(self, tenant_id, query, tools_selected, latency_ms):
        metrics.gauge("symfonic.routing.tools", len(tools_selected), tags=[f"tenant:{tenant_id}"])

    async def on_token_budget_computed(self, tenant_id, total_tool_tokens, filtered_tool_tokens, savings_pct):
        metrics.gauge("symfonic.tokens.savings_pct", savings_pct, tags=[f"tenant:{tenant_id}"])

CallbackHandler for Compliance Logging

class ComplianceLogger:
    """Log all LLM interactions for audit trail.

    v7.4.3: every non-react LLM call site (metacognition critic,
    context-window summariser, insight extractor, LLM entity extractor,
    consolidation extractor, procedural router) now fires
    ``on_llm_end`` with a populated ``node_name`` discriminator so the
    audit trail captures the full cost surface, not just the main
    reasoning loop.
    """

    async def on_llm_start(self, event):
        audit_log.write({
            "type": "llm_start",
            "model": event.model,
            "message_count": len(event.messages),
            "timestamp": datetime.utcnow().isoformat(),
        })

    async def on_llm_end(self, event):
        audit_log.write({
            "type": "llm_end",
            # v7.4.3: node_name attributes the call to react /
            # metacognition_critic / context_window_summariser / etc.
            # so compliance dashboards can slice cost by subsystem.
            "node_name": event.node_name,
            "model": event.model,
            "usage": event.usage,
            "timestamp": datetime.utcnow().isoformat(),
        })

    # Implement remaining hooks as no-ops
    async def on_agent_start(self, event): pass
    async def on_agent_end(self, event): pass
    async def on_node_start(self, event): pass
    async def on_node_end(self, event): pass
    async def on_node_error(self, event): pass

Performance Tips

Token Budget

Reduce system prompt size by limiting enabled layers and tool count:

config = FrameworkConfig(
    enabled_layers={"semantic", "episodic"},  # skip working/procedural/prospective
    lazy_tooling=True,                         # only inject relevant tools per query
)

Conversation History

Limit the number of prior messages injected:

from symfonic.memory.orchestrator.config import OrchestratorConfig

config = FrameworkConfig(
    orchestrator=OrchestratorConfig(
        context_budget=ContextBudget(max_history_messages=10),
    ),
)

Streaming Consolidation

Consolidation runs in a background task after streaming completes. Call flush_background_tasks() before shutdown to ensure all writes complete:

# In your FastAPI lifespan
@asynccontextmanager
async def lifespan(app):
    yield
    await agent.flush_background_tasks()

app = FastAPI(lifespan=lifespan)

Full Production Setup

Combining everything from Levels 1-5:

import asyncio
import os
from contextlib import asynccontextmanager

from fastapi import FastAPI

from symfonic.agent.config import FrameworkConfig
from symfonic.agent.domain import DomainTemplate
from symfonic.agent.engine import SymfonicAgent
from symfonic.agent.fastapi.router import create_agent_router
from symfonic.core.providers import AnthropicProvider
from symfonic.memory.backends.pool import PostgresPoolManager
from symfonic.memory.backends.postgres_graph import PostgresGraphBackend
from symfonic.memory.backends.postgres_vector import PostgresVectorBackend


MY_DOMAIN = DomainTemplate(
    name="my-product",
    required_labels=["SOUL", "USER_PROFILE"],
    soul_schema={"name": "str", "plan": "str"},
)

pool = None
agent = None

@asynccontextmanager
async def lifespan(app):
    global pool, agent
    dsn = os.environ["POSTGRES_DSN"]

    pool = PostgresPoolManager(dsn=dsn, min_size=2, max_size=20)
    await pool.open()

    graph = PostgresGraphBackend(pool)
    vector = PostgresVectorBackend(pool)
    await graph.ensure_schema()
    await vector.ensure_schema()

    agent = SymfonicAgent(
        model_provider=AnthropicProvider(),
        embedding_provider=my_embedding_provider,
        graph_backend=graph,
        vector_backend=vector,
        config=FrameworkConfig(
            domain=MY_DOMAIN,
            auto_hydrate=True,
            auto_consolidate=True,
            spreading_activation=True,
            self_reflection=True,
            enable_hms_prompt=True,
            # v7.0 production stance
            intent_filter_mode="observe",          # flip to enforce after burn-in
            tool_routing_mode="observe",
            fabrication_check_mode="observe",
            quick_nap_interval=10,
            nightly_nap_enabled=True,
        ),
    )

    router = create_agent_router(agent, prefix="/api/v1")
    app.include_router(router)

    yield

    await agent.flush_background_tasks()
    await pool.close()

app = FastAPI(title="Production Agent", lifespan=lifespan)

@app.get("/health")
async def health():
    return {"status": "ok"}

Run:

docker compose up -d
ANTHROPIC_API_KEY=sk-ant-... uvicorn my_app:app --host 0.0.0.0 --port 8000

Further Reading

• Feature Catalog -- All 11 capability clusters with opt-in examples
• Guide 07 Intent & Tool Routing -- v7.0.1 tool narrowing recipe
• Guide 08 Fabrication Detection -- Deployment stance and confidence thresholds
• Guide 09 Consolidation & Deep Sleep -- 11 phases in detail
• Integration Guide -- Configuration reference
• Memory Concepts -- Deep dive into 5-layer HMS
• Architecture -- Module map and Thalamic pipeline
• API Reference -- Full endpoint documentation