def create_agent_router(
agent: SymfonicAgent,
prefix: str = "/api/v1",
app: Any = None,
) -> APIRouter:
"""Create a FastAPI router wrapping a SymfonicAgent.
Endpoints:
POST {prefix}/chat -- synchronous chat, returns AgentResponse JSON
POST {prefix}/stream -- SSE streaming, yields StreamChunk events
Args:
agent: The SymfonicAgent instance to expose.
prefix: URL prefix for the router (default ``/api/v1``).
app: The ``FastAPI`` app this router will be mounted on. Optional, and
only read by the production auth gate: an injected scope resolver
lives on ``app.state``, so without the app the gate cannot tell an
authenticated deployment from an unauthenticated one and refuses.
Pass it whenever you use ``install_scope_resolver``.
Pass the app the router is actually mounted on. Mounting it on a
different one no longer bypasses the check —
``get_tenant_scope`` re-runs the gate against ``request.app``
before resolving any scope — but passing the right app is what
keeps the failure at startup instead of on the first request,
which is the whole point of a fail-fast gate.
Returns:
Configured APIRouter.
Raises:
RuntimeError: If a production environment is detected (via
``SYMFONIC_ENV``, ``APP_ENV``, ``ENVIRONMENT``, or ``NODE_ENV``
set to ``production``/``prod``) but no tenant-auth verifier is
registered via :func:`set_tenant_auth_verifier` **and** no
satisfying resolver was injected into ``app``. Set
``ALLOW_INSECURE_PROD=true`` to bypass (NOT recommended —
logs a CRITICAL warning).
"""
# Fail-fast in production when auth verifier isn't wired. The previous
# dev-mode fallback (one-shot WARN then trust X-Tenant-ID blindly) was
# a critical tenant-isolation bug waiting for the next deployer to
# forget ``register_tenant_auth_verifier()``.
_check_production_auth_gate(app)
router = APIRouter(prefix=prefix, tags=["agent"])
@router.post("/chat", response_model=AgentResponse)
async def chat(
request: AgentRequest,
raw_request: Request,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
_budget: None = Depends(check_budget_before_llm), # noqa: B008
) -> AgentResponse:
"""Execute the agent and return the full response."""
invocation = _CODEC.decode_invocation(
request,
scope=scope,
is_admin=request_is_admin(raw_request),
principal_id=request_user_id(raw_request),
)
try:
# TA8.43 (C1-K): the admin bit is bound, not passed. This route
# already derived it from the request and never from the body, so
# nothing about the HTTP behaviour changes -- what changes is that
# the facade now refuses the keyword, so a caller that reaches it
# without authenticating anybody cannot grant itself the bypass.
with bind_admin_authority(invocation.admin_claim()):
result = await agent.run(**invocation.as_kwargs())
except Exception as exc:
raise raise_mapped(exc, CHAT_POLICY) from exc
return _CODEC.encode_result(result)
@router.post("/stream")
async def stream(
request: AgentRequest,
raw_request: Request,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
_budget: None = Depends(check_budget_before_llm), # noqa: B008
) -> Any:
"""Stream agent execution as Server-Sent Events."""
from sse_starlette.sse import EventSourceResponse
invocation = _CODEC.decode_invocation(
request,
scope=scope,
is_admin=request_is_admin(raw_request),
principal_id=request_user_id(raw_request),
)
adapter = SseFrameAdapter(STREAM_POLICY)
def encode(chunk: Any) -> dict[str, str]:
# STR-2: the frozen ``/stream`` frame shape. The route owns the
# shape, the adapter owns the lifecycle, and neither owns both.
return {"event": chunk.event_type, "data": chunk.model_dump_json()}
return EventSourceResponse(
adapter.frames(
_bound(invocation, agent.stream(**invocation.as_kwargs())),
encode,
),
)
@router.post("/stream/typed")
async def stream_typed(
request: AgentRequest,
raw_request: Request,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
_budget: None = Depends(check_budget_before_llm), # noqa: B008
) -> Any:
"""Stream agent execution as typed Server-Sent Events.
Each SSE event's ``event`` field is the StreamEvent class name
(e.g. ``TextDeltaEvent``, ``ToolCallStartEvent``). The ``data``
field is the JSON-serialized event payload.
"""
from sse_starlette.sse import EventSourceResponse
invocation = _CODEC.decode_invocation(
request,
scope=scope,
is_admin=request_is_admin(raw_request),
principal_id=request_user_id(raw_request),
)
adapter = SseFrameAdapter(STREAM_POLICY)
return EventSourceResponse(
adapter.frames(
_bound(
invocation, agent.stream_typed(**invocation.as_kwargs())
),
_encode_typed_event,
),
)
@router.post("/resume/{pause_token}")
async def resume(
pause_token: str,
raw_request: Request,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> Any:
"""Resume a paused agent execution.
Dispatches on the token's ``name`` claim:
* ``name == "ask_user"`` -> :meth:`SymfonicAgent.resume` with an
``AskUserResponse`` body (existing public API; URL unchanged).
* other registered names -> :meth:`SymfonicAgent.resume_interrupt`
with a body validated against the registered ``response_schema``
(Roadmap Item 9, behind ``experimental_interrupt=True``).
Returns a typed Server-Sent Events stream as the run continues.
**Which axes this door verifies (TA8.45).** It states the tenant scope
and nothing else: the router has the URL, the POSTed body and
``get_tenant_scope``'s scope, and no fact that says which session, run
or call the consumer believes it is answering. So over HTTP the scope
comparison is real on every redemption and session / run / call are
filled from the authenticated claims inside
:meth:`SymfonicAgent._continuation_kernel_impl`, which makes those three
tautologies *on this surface*. Not a forgery risk — the fill happens
only after the envelope's signature and expiry are verified, so the
values compared are the token's own — but a narrower claim than "all
four verified at the entrance", and §5.1 of the continuation
certificate is where it is written down.
They are deliberately not threaded from here: the only sources
available are client-supplied (query string or body), which would let a
caller choose the axes its own token is checked against. A transport
that genuinely knows an axis should state it on
:meth:`SymfonicAgent.resume`, where each is separately refusable by
name.
"""
from sse_starlette.sse import EventSourceResponse
# Which registration does this token belong to, and what validates its
# body? One facade call (T4.1.2) instead of a preview decode against
# the agent's private config plus a lookup in its private registry.
# The answer is deliberately non-authoritative: the engine re-decodes
# and raises the typed error, so token validation stays in the layer
# that owns it (TRN-2, and the reason T3.4.5's suites can still see it).
dispatch = agent.describe_resume_dispatch(pause_token)
try:
raw_body = await raw_request.json()
except Exception as exc: # noqa: BLE001 - any parse failure is a 400
return EventSourceResponse(
_single_error_event(f"Invalid JSON body: {exc}", 400),
)
try:
if dispatch.is_ask_user:
response_obj: Any = AskUserResponse.model_validate(raw_body)
elif dispatch.response_schema is None:
return EventSourceResponse(
_single_error_event(
f"Token references unknown interrupt {dispatch.name!r}",
400,
),
)
else:
response_obj = dispatch.response_schema.model_validate(raw_body)
except Exception as exc: # noqa: BLE001 - any validation failure is a 400
return EventSourceResponse(
_single_error_event(f"Invalid resume payload: {exc}", 400),
)
events = (
agent.resume(
pause_token=pause_token, response=response_obj, scope=scope,
)
if dispatch.is_ask_user
else agent.resume_interrupt(
pause_token=pause_token, response=response_obj, scope=scope,
)
)
adapter = SseFrameAdapter(RESUME_POLICY)
return EventSourceResponse(adapter.frames(events, _encode_typed_event))
@router.get("/chats")
async def list_chats(
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> list[dict[str, Any]]:
"""List active chat sessions for the current tenant."""
try:
return agent.list_sessions(scope.tenant_id)
except Exception as exc:
raise raise_mapped(exc, CRUD_POLICY) from exc
@router.get("/memories/status", response_model=list[MemoryBlockStatus])
async def memory_status(
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> list[MemoryBlockStatus]:
"""Return the status of known memory blocks for the current tenant."""
try:
return await agent.describe_memory_blocks(scope)
except Exception as exc:
raise raise_mapped(exc, CRUD_POLICY) from exc
# ------------------------------------------------------------------
# Memory CRUD
# ------------------------------------------------------------------
@router.get("/memories")
async def list_memories(
layer: str | None = None,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> list[dict[str, Any]]:
"""List memory nodes for the tenant, grouped by label prefix.
Query params:
layer: Optional memory layer filter (semantic, episodic,
procedural, prospective, working). When omitted, all
available layers are merged.
"""
memory_scope = scope.to_memory_scope()
# Determine which layers to query.
if layer is not None:
try:
target_layer = MemoryLayer(layer)
except ValueError:
raise HTTPException( # noqa: B904
status_code=400,
detail=f"unknown layer: {layer!r}",
)
target_layers = [target_layer]
else:
target_layers = list(MemoryLayer)
all_nodes: list[Any] = []
result: list[dict[str, Any]] = []
for ml in target_layers:
store = agent._orchestrator.get_layer(ml)
if store is None:
continue
# EpisodicLayer is vector-backed (no _graph); enumerate via its
# public list_events() API and emit result dicts directly.
if ml == MemoryLayer.EPISODIC:
try:
entries = await store.list_events(memory_scope) # type: ignore[union-attr]
for entry in entries:
content = entry.content or ""
label = content[:80] if content else "episodic-event"
md = entry.metadata or {}
group = (
str(md.get("group", ""))
or str(md.get("action_type", ""))
or "episodic"
)
created_at_str = (
entry.created_at.isoformat()
if entry.created_at
else None
)
result.append({
"id": str(entry.id) if entry.id else "",
"label": label,
"group": group,
"layer": ml.value,
"properties": md,
"importance": entry.importance,
"created_at": created_at_str,
"updated_at": None,
})
except Exception:
logger.debug("Failed to query episodic layer", exc_info=True)
continue
graph = getattr(store, "_graph", None)
if graph is None:
continue
try:
nodes = await graph.query_nodes( # type: ignore[union-attr]
memory_scope, layer=ml,
)
all_nodes.extend((ml, n) for n in nodes)
except Exception:
logger.debug("Failed to query layer %s", ml, exc_info=True)
for ml, node in all_nodes:
label = str(node.label)
# Prefer explicit metadata group/type, fall back to label prefix.
props = node.properties or {}
if ml == MemoryLayer.PROCEDURAL:
# Procedural nodes carry description-length labels (e.g.
# "Standard app deployment workflow triggered by ..."), so
# the default split-on-":" logic would leak prose into the
# group column. Use a stable, short group derived from the
# skill id or a constant fallback.
group = (
str(props.get("group", ""))
or str(props.get("id", ""))
or "procedure"
)
else:
group = (
str(props.get("group", ""))
or str(props.get("type", ""))
or (label.split(":")[0].strip() if ":" in label else label)
)
result.append({
"id": str(node.id),
"label": label,
"group": group,
"layer": ml.value,
"properties": node.properties,
"importance": node.importance,
"created_at": str(node.created_at) if node.created_at else None,
"updated_at": str(node.updated_at) if node.updated_at else None,
})
return result
@router.get("/memories/{node_id}")
async def get_memory(
node_id: str,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> dict[str, Any]:
"""Return the full details of a single semantic memory node."""
semantic = agent._orchestrator.get_layer(MemoryLayer.SEMANTIC)
if semantic is None:
raise HTTPException(404, "Semantic layer not available")
memory_scope = scope.to_memory_scope()
node = await semantic._graph.get_node( # type: ignore[union-attr]
memory_scope, NodeId(node_id),
)
if node is None:
raise HTTPException(404, f"Node {node_id} not found")
return {
"id": str(node.id),
"label": str(node.label),
"layer": node.layer.value,
"properties": node.properties,
"importance": node.importance,
"access_count": node.access_count,
"created_at": str(node.created_at) if node.created_at else None,
"updated_at": str(node.updated_at) if node.updated_at else None,
}
@router.patch("/memories/{node_id}")
async def update_memory(
node_id: str,
request: Request,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> dict[str, Any]:
"""Update a memory node's content or properties."""
body = await request.json()
semantic = agent._orchestrator.get_layer(MemoryLayer.SEMANTIC)
if semantic is None:
raise HTTPException(404, "Semantic layer not available")
memory_scope = scope.to_memory_scope()
updates: dict[str, object] = {}
if "content" in body:
updates["label"] = body["content"]
if "properties" in body:
updates["properties"] = body["properties"]
if "importance" in body:
updates["importance"] = body["importance"]
# Audit trail
props = updates.get("properties", {})
if isinstance(props, dict):
props["_last_edited_by"] = "user_manual_edit"
updates["properties"] = props
try:
node = await semantic._graph.update_node( # type: ignore[union-attr]
memory_scope, NodeId(node_id), updates,
)
_emit_router_audit(
request, scope,
action="update_memory",
resource_type="memory_node",
resource_id=node_id,
metadata={"fields": sorted(updates.keys())},
)
return {
"id": str(node.id),
"label": node.label,
"properties": node.properties,
}
except KeyError as exc:
raise HTTPException(404, f"Node {node_id} not found") from exc
@router.delete("/memories/{node_id}")
async def delete_memory(
node_id: str,
request: Request,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> dict[str, str]:
"""Delete a memory node and cascade-delete its edges."""
semantic = agent._orchestrator.get_layer(MemoryLayer.SEMANTIC)
if semantic is None:
raise HTTPException(404, "Semantic layer not available")
memory_scope = scope.to_memory_scope()
# Verify node exists before deleting (backends silently ignore missing)
existing = await semantic._graph.get_node( # type: ignore[union-attr]
memory_scope, NodeId(node_id),
)
if existing is None:
raise HTTPException(404, f"Node {node_id} not found")
await semantic._graph.delete_node( # type: ignore[union-attr]
memory_scope, NodeId(node_id),
)
_emit_router_audit(
request, scope,
action="delete_memory",
resource_type="memory_node",
resource_id=node_id,
metadata={"label": str(existing.label)},
)
return {"deleted": node_id}
# ------------------------------------------------------------------
# Procedural CRUD (Workflows)
# ------------------------------------------------------------------
@router.get("/procedures")
async def list_procedures(
status: str | None = None,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> list[dict[str, Any]]:
"""List procedural nodes (learned workflows/skills).
By default only approved (active) skills are returned.
Use ``?status=draft`` or ``?status=rejected`` to filter, or
``?status=all`` to return everything.
"""
procedural = agent._orchestrator.get_layer(MemoryLayer.PROCEDURAL)
if procedural is None:
return []
memory_scope = scope.to_memory_scope()
nodes = await procedural._graph.query_nodes( # type: ignore[union-attr]
memory_scope, layer=MemoryLayer.PROCEDURAL,
)
result: list[dict[str, Any]] = []
for node in nodes:
props = node.properties or {}
node_status = props.get("status", "approved")
# Default filter: only approved. With ?status=<val>: match that.
if status is None:
if node_status != "approved":
continue
elif status != "all" and node_status != status:
continue
result.append({
"id": str(node.id),
"label": str(node.label),
"content": props.get("content", node.label),
"steps": props.get("steps", []),
"properties": props,
"status": node_status,
"active": props.get("active", True),
"importance": node.importance,
"created_at": str(node.created_at) if node.created_at else None,
"updated_at": str(node.updated_at) if node.updated_at else None,
})
return result
@router.get("/procedures/drafts")
async def list_draft_procedures(
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> list[dict[str, Any]]:
"""List all draft (pending review) procedural skills."""
procedural = agent._orchestrator.get_layer(MemoryLayer.PROCEDURAL)
if procedural is None:
return []
memory_scope = scope.to_memory_scope()
nodes = await procedural._graph.query_nodes( # type: ignore[union-attr]
memory_scope, layer=MemoryLayer.PROCEDURAL,
)
return [
{
"id": str(node.id),
"label": str(node.label),
"content": (node.properties or {}).get("content", node.label),
"steps": (node.properties or {}).get("steps", []),
"status": "draft",
"active": False,
"importance": node.importance,
"created_at": str(node.created_at) if node.created_at else None,
}
for node in nodes
if (node.properties or {}).get("status") == "draft"
]
@router.post("/procedures/{node_id}/approve")
async def approve_procedure(
node_id: str,
request: Request,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> dict[str, Any]:
"""Approve a draft skill -- makes it active."""
procedural = agent._orchestrator.get_layer(MemoryLayer.PROCEDURAL)
if procedural is None:
raise HTTPException(404, "Procedural layer not available")
memory_scope = scope.to_memory_scope()
try:
node = await procedural.approve_skill(
memory_scope, NodeId(node_id),
)
except KeyError as exc:
raise HTTPException(404, str(exc)) from exc
_emit_router_audit(
request, scope,
action="approve_procedure",
resource_type="procedure",
resource_id=node_id,
)
props = node.properties or {}
return {
"id": str(node.id),
"label": str(node.label),
"status": props.get("status", "approved"),
"active": props.get("active", True),
}
@router.post("/procedures/{node_id}/reject")
async def reject_procedure(
node_id: str,
request: Request,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> dict[str, Any]:
"""Reject a draft skill -- archives it."""
procedural = agent._orchestrator.get_layer(MemoryLayer.PROCEDURAL)
if procedural is None:
raise HTTPException(404, "Procedural layer not available")
memory_scope = scope.to_memory_scope()
try:
node = await procedural.reject_skill(
memory_scope, NodeId(node_id),
)
except KeyError as exc:
raise HTTPException(404, str(exc)) from exc
_emit_router_audit(
request, scope,
action="reject_procedure",
resource_type="procedure",
resource_id=node_id,
)
props = node.properties or {}
return {
"id": str(node.id),
"label": str(node.label),
"status": props.get("status", "rejected"),
"active": props.get("active", False),
}
@router.get("/procedures/{node_id}")
async def get_procedure(
node_id: str,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> dict[str, Any]:
"""Return a single procedural node's full details."""
procedural = agent._orchestrator.get_layer(MemoryLayer.PROCEDURAL)
if procedural is None:
raise HTTPException(404, "Procedural layer not available")
memory_scope = scope.to_memory_scope()
node = await procedural._graph.get_node( # type: ignore[union-attr]
memory_scope, NodeId(node_id),
)
if node is None:
raise HTTPException(404, f"Node {node_id} not found")
props = node.properties or {}
return {
"id": str(node.id),
"label": str(node.label),
"content": props.get("content", node.label),
"steps": props.get("steps", []),
"properties": props,
"active": props.get("active", True),
"importance": node.importance,
"created_at": str(node.created_at) if node.created_at else None,
"updated_at": str(node.updated_at) if node.updated_at else None,
}
@router.patch("/procedures/{node_id}")
async def update_procedure(
node_id: str,
request: Request,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> dict[str, Any]:
"""Update a procedural node's steps, properties, or active flag."""
body = await request.json()
procedural = agent._orchestrator.get_layer(MemoryLayer.PROCEDURAL)
if procedural is None:
raise HTTPException(404, "Procedural layer not available")
memory_scope = scope.to_memory_scope()
# Fetch existing node to merge properties
existing = await procedural._graph.get_node( # type: ignore[union-attr]
memory_scope, NodeId(node_id),
)
if existing is None:
raise HTTPException(404, f"Node {node_id} not found")
merged_props: dict[str, object] = dict(existing.properties or {})
touched_fields: list[str] = []
if "steps" in body:
merged_props["steps"] = body["steps"]
touched_fields.append("steps")
if "properties" in body:
merged_props.update(body["properties"])
touched_fields.append("properties")
if "active" in body:
merged_props["active"] = body["active"]
touched_fields.append("active")
# Audit trail
merged_props["_last_edited_by"] = "user_manual_edit"
try:
node = await procedural._graph.update_node( # type: ignore[union-attr]
memory_scope, NodeId(node_id), {"properties": merged_props},
)
_emit_router_audit(
request, scope,
action="update_procedure",
resource_type="procedure",
resource_id=node_id,
metadata={"fields": sorted(touched_fields)},
)
props = node.properties or {}
return {
"id": str(node.id),
"label": node.label,
"steps": props.get("steps", []),
"properties": props,
"active": props.get("active", True),
}
except KeyError as exc:
raise HTTPException(404, f"Node {node_id} not found") from exc
@router.delete("/procedures/{node_id}")
async def delete_procedure(
node_id: str,
request: Request,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> dict[str, str]:
"""Delete a procedural node and cascade-delete its edges."""
procedural = agent._orchestrator.get_layer(MemoryLayer.PROCEDURAL)
if procedural is None:
raise HTTPException(404, "Procedural layer not available")
memory_scope = scope.to_memory_scope()
existing = await procedural._graph.get_node( # type: ignore[union-attr]
memory_scope, NodeId(node_id),
)
if existing is None:
raise HTTPException(404, f"Node {node_id} not found")
await procedural._graph.delete_node( # type: ignore[union-attr]
memory_scope, NodeId(node_id),
)
_emit_router_audit(
request, scope,
action="delete_procedure",
resource_type="procedure",
resource_id=node_id,
metadata={"label": str(existing.label)},
)
return {"deleted": node_id}
@router.post("/procedures/deduplicate")
async def deduplicate_procedures(
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> dict[str, object]:
"""Merge semantically similar procedural nodes for the current tenant.
Scans all procedural nodes and greedily merges pairs whose textual
similarity is >= 0.75 (the default SemanticMerge threshold). Absorbed
duplicate nodes are deleted; canonical nodes are updated in-place.
Returns:
JSON object with ``merged`` (count of absorbed nodes) and
``remaining`` (list of surviving canonical procedure dicts).
"""
from symfonic.memory.janitor import SemanticMerge
procedural = agent._orchestrator.get_layer(MemoryLayer.PROCEDURAL)
if procedural is None:
raise HTTPException(404, "Procedural layer not available")
memory_scope = scope.to_memory_scope()
try:
janitor = SemanticMerge()
merged_count, remaining_nodes = await janitor.deduplicate_all(
memory_scope, procedural._graph # type: ignore[union-attr]
)
remaining = [
{
"id": str(n.id),
"label": str(n.label),
"content": n.properties.get("content", n.label),
"steps": n.properties.get("steps", []),
"active": n.properties.get("active", True),
"importance": n.importance,
}
for n in remaining_nodes
]
return {"merged": merged_count, "remaining": remaining}
except Exception as exc:
raise HTTPException(status_code=500, detail="Deduplication failed") from exc
@router.post("/procedures/{node_id}/toggle")
async def toggle_procedure(
node_id: str,
request: Request,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> dict[str, Any]:
"""Toggle a procedure's active flag (true <-> false)."""
procedural = agent._orchestrator.get_layer(MemoryLayer.PROCEDURAL)
if procedural is None:
raise HTTPException(404, "Procedural layer not available")
memory_scope = scope.to_memory_scope()
existing = await procedural._graph.get_node( # type: ignore[union-attr]
memory_scope, NodeId(node_id),
)
if existing is None:
raise HTTPException(404, f"Node {node_id} not found")
merged_props = dict(existing.properties or {})
current_active = merged_props.get("active", True)
merged_props["active"] = not current_active
merged_props["_last_edited_by"] = "user_toggle"
node = await procedural._graph.update_node( # type: ignore[union-attr]
memory_scope, NodeId(node_id),
{"properties": merged_props},
)
_emit_router_audit(
request, scope,
action="update_procedure",
resource_type="procedure",
resource_id=node_id,
metadata={"fields": ["active"], "new_active": not current_active},
)
props = node.properties or {}
return {
"id": str(node.id),
"label": node.label,
"active": props.get("active", True),
"properties": props,
}
# ------------------------------------------------------------------
# Memory consolidation (Deep Sleep)
# ------------------------------------------------------------------
@router.post("/memory/consolidate")
async def consolidate_memory(
request: Request,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
_budget: None = Depends(check_budget_before_llm), # noqa: B008
) -> dict[str, Any]:
"""Trigger manual deep-sleep consolidation for the tenant.
Replays recent traces, strengthens recurring nodes, prunes
orphans, generates meta-nodes from clusters, promotes
user-corrected profile facts onto the tenant's SOUL node, and
writes any pending inferred edges to the graph.
Accepts an optional JSON body::
{
"pending_connections": [
{"source": "NodeA", "target": "NodeB", "relationship": "ASSOCIATED"}
]
}
``pending_connections`` is typically obtained from the
``activation_log.pending_connections`` field returned by ``/chat``.
"""
from symfonic.core.learning.consolidation import SleepConsolidator
semantic = agent._orchestrator.get_layer(MemoryLayer.SEMANTIC)
if semantic is None or not hasattr(semantic, "_graph"):
raise HTTPException(404, "Semantic layer not available")
# Parse optional body (pending_connections from spreading activation)
pending_connections: list[dict[str, Any]] | None = None
try:
body = await request.json()
if isinstance(body, dict):
raw = body.get("pending_connections")
if isinstance(raw, list):
pending_connections = raw
logger.info(
"Consolidate request received %d pending_connections for tenant=%s",
len(pending_connections),
scope.tenant_id,
)
except (json.JSONDecodeError, KeyError, TypeError, AttributeError):
pass # Body is optional; proceed without pending connections
memory_scope = scope.to_memory_scope()
# Pass episodic layer if available (enables Phase 10)
episodic = agent._orchestrator.get_layer(MemoryLayer.EPISODIC)
# Pass procedural layer if available (enables Phase 12)
procedural = agent._orchestrator.get_layer(MemoryLayer.PROCEDURAL)
consolidator = SleepConsolidator(
semantic._graph, # type: ignore[union-attr]
episodic_layer=episodic,
procedural_layer=procedural,
promotion_min_pattern_count=getattr(
agent._config, "promotion_min_pattern_count", 3,
),
promotion_recency_days=getattr(
agent._config, "promotion_recency_days", 30,
),
promotion_max_drafts_per_run=getattr(
agent._config, "promotion_max_drafts_per_run", 5,
),
# v7.4.7 adopter Ask 9 PR-A: tool-call fallback opt-in.
promotion_use_tool_calls_fallback=getattr(
agent._config, "phase_12_action_type_from_tool_calls", False,
),
# v7.7.2 adopter fabrication-promotion ask: default-off
# filter for assistant-narrated promotion.
promotion_promote_assistant_content=getattr(
agent._config, "phase_12_promote_assistant_content", False,
),
# v7.6 adopter Ask 9 Option A: LLM-backed Phase 12 extractor.
# Default-off; wire the agent's bound chat model so the
# factory can instantiate ``LlmProceduralExtractor``.
phase_12_use_llm_extractor=getattr(
agent._config, "phase_12_use_llm_extractor", False,
),
phase_12_llm_model=getattr(
agent._config, "phase_12_llm_model", "claude-haiku-4-5",
),
phase_12_llm_max_episodes_per_run=getattr(
agent._config, "phase_12_llm_max_episodes_per_run", 100,
),
phase_12_llm_max_drafts_per_run=getattr(
agent._config, "phase_12_llm_max_drafts_per_run", 5,
),
chat_model=agent.get_chat_model("consolidation"),
episodic_summarization_max_entries=getattr(
agent._config, "episodic_summarization_max_entries", 100,
),
episodic_summarization_batch_size=getattr(
agent._config, "episodic_summarization_batch_size", 50,
),
phase1_spreading_weight=getattr(
agent._config, "phase1_spreading_weight", 0.5,
),
synthetic_link_min_co_count=getattr(
agent._config, "synthetic_link_min_co_count", 2,
),
# v7.2 Roadmap Item 12 -- EntityLinker phase (default-off).
enable_entity_linker=getattr(
agent._config, "enable_entity_linker", False,
),
entity_linker_extractor_kind=getattr(
agent._config, "entity_linker_extractor", "regex",
),
entity_linker_min_mention_count=getattr(
agent._config, "entity_linker_min_mention_count", 2,
),
entity_linker_max_episodics_per_run=getattr(
agent._config, "entity_linker_max_episodics_per_run", 200,
),
entity_linker_confidence_threshold=getattr(
agent._config, "entity_linker_confidence_threshold", 0.5,
),
)
# frozenset (not dict): the schema is only READ here to learn which
# field names count as "profile" -- never written. See
# symfonic.core.learning.phases_profile for why the old
# dict-copy-and-mutate-in-place shape was unsafe.
profile_fields = (
frozenset(agent._config.domain.soul_schema)
if agent._config.domain.soul_schema
else None
)
try:
report = await consolidator.run(
memory_scope,
profile_fields=profile_fields,
pending_connections=pending_connections,
)
return report.to_dict()
except Exception as exc:
raise HTTPException(
status_code=500,
detail="Consolidation failed",
) from exc
# ------------------------------------------------------------------
# Memory-create sub-router (v6.1.4 — POST /memories/{layer})
# ------------------------------------------------------------------
from symfonic.agent.fastapi.memory_create_router import (
create_memory_create_router,
)
memory_create_router = create_memory_create_router(agent)
router.include_router(memory_create_router)
# ------------------------------------------------------------------
# Graph sub-router (neighborhood, pathfinding, edges, clusters)
# ------------------------------------------------------------------
from symfonic.agent.fastapi.graph_router import create_graph_router
graph_router = create_graph_router(agent)
router.include_router(graph_router)
# ------------------------------------------------------------------
# Graph maintenance sub-router (bulk ops, audit, prune)
# ------------------------------------------------------------------
from symfonic.agent.fastapi.graph_maintenance_router import (
create_graph_maintenance_router,
)
maintenance_router = create_graph_maintenance_router(agent)
router.include_router(maintenance_router)
# ------------------------------------------------------------------
# Tenant privacy sub-router (GDPR Art. 17 + Art. 20)
# ------------------------------------------------------------------
from symfonic.agent.fastapi.tenant_privacy_router import (
create_tenant_privacy_router,
)
privacy_router = create_tenant_privacy_router(agent)
router.include_router(privacy_router)
# ------------------------------------------------------------------
# Metrics sub-router (token usage / cost dashboard) -- opt-in
# ------------------------------------------------------------------
if agent.metrics_collector is not None:
from symfonic.agent.fastapi.metrics_router import create_metrics_router
metrics_router = create_metrics_router(agent.metrics_collector, prefix="")
router.include_router(metrics_router)
return router