def create_agent_router(
agent: SymfonicAgent,
prefix: str = "/api/v1",
) -> 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``).
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`. 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()
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."""
try:
return await agent.run(
query=request.query,
attachments=request.attachments,
scope=scope,
is_admin=bool(getattr(raw_request.state, "is_admin", False)),
session_id=request.session_id,
)
except SecurityScopeError as exc:
raise HTTPException(
status_code=403,
detail="Access denied: missing or invalid tenant scope",
) from exc
except ScopeValidationError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
except SymfonicAgentError as exc:
# Budget breaker (defence-in-depth at the agent layer) surfaces
# as a typed SymfonicAgentError with "Budget exceeded:" prefix.
# Translate to 429 so clients treat it like the dep-layer 429.
msg = str(exc)
if msg.startswith("Budget exceeded:"):
raise HTTPException(
status_code=429, detail=msg,
headers={"Retry-After": "3600"},
) from exc
# v7.1.3 (Item 10.1): map code -> HTTP status (bad_request->400,
# failed_dependency->424, ...). Unknown codes fall back to 500.
status_code = _status_from_agent_error(exc)
detail = str(exc) if status_code != 500 else "Agent execution failed"
raise HTTPException(status_code=status_code, detail=detail) from exc
except Exception as exc:
raise HTTPException(status_code=500, detail="Internal server error") from exc
@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
is_admin = bool(getattr(raw_request.state, "is_admin", False))
async def event_generator() -> Any:
try:
async for chunk in agent.stream(
query=request.query,
attachments=request.attachments,
scope=scope,
is_admin=is_admin,
session_id=request.session_id,
):
yield {
"event": chunk.event_type,
"data": chunk.model_dump_json(),
}
except SecurityScopeError:
yield {
"event": "error",
"data": '{"detail": "Access denied: missing or invalid tenant scope"}',
}
except ScopeValidationError as exc:
yield {
"event": "error",
"data": json.dumps({"detail": str(exc)}),
}
except SymfonicAgentError as exc:
msg = str(exc)
if msg.startswith("Budget exceeded:"):
yield {
"event": "error",
"data": json.dumps({"detail": msg, "status": 429}),
}
else:
# v7.1.3 (Item 10.1): same status mapping as /chat so
# SSE clients see consistent error semantics.
status_code = _status_from_agent_error(exc)
detail = (
str(exc) if status_code != 500 else "Agent execution failed"
)
yield {
"event": "error",
"data": json.dumps(
{"detail": detail, "status": status_code},
),
}
except Exception:
logger.exception("Stream event generator failed")
yield {
"event": "error",
"data": '{"detail": "Internal server error"}',
}
return EventSourceResponse(event_generator())
@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
from symfonic.core.streaming.transpiler import _event_to_dict
is_admin = bool(getattr(raw_request.state, "is_admin", False))
async def typed_event_generator() -> Any:
try:
async for stream_event in agent.stream_typed(
query=request.query,
attachments=request.attachments,
scope=scope,
is_admin=is_admin,
session_id=request.session_id,
):
event_name = type(stream_event).__name__
yield {
"event": event_name,
# default=str is a defense-in-depth fallback for any
# type _make_json_safe might miss (e.g. nested
# langchain objects).
"data": json.dumps(
_event_to_dict(stream_event), default=str,
),
}
except SecurityScopeError:
yield {
"event": "error",
"data": '{"detail": "Access denied: missing or invalid tenant scope"}',
}
except ScopeValidationError as exc:
yield {
"event": "error",
"data": json.dumps({"detail": str(exc)}),
}
except SymfonicAgentError as exc:
msg = str(exc)
if msg.startswith("Budget exceeded:"):
yield {
"event": "error",
"data": json.dumps({"detail": msg, "status": 429}),
}
else:
# v7.1.3 (Item 10.1): consistent code -> status mapping.
status_code = _status_from_agent_error(exc)
detail = (
str(exc) if status_code != 500 else "Agent execution failed"
)
yield {
"event": "error",
"data": json.dumps(
{"detail": detail, "status": status_code},
),
}
except Exception:
logger.exception("Typed stream event generator failed")
yield {
"event": "error",
"data": '{"detail": "Internal server error"}',
}
return EventSourceResponse(typed_event_generator())
@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.
"""
from sse_starlette.sse import EventSourceResponse
from symfonic.agent.middleware.pause_token import PauseToken
from symfonic.core.streaming.transpiler import _event_to_dict
# Stateless preview decode (HMAC + expiry only) to discover the
# registered interrupt name. Roadmap Item 9 dispatches generic
# interrupts here; ask_user stays on the historical path so the
# existing /resume contract -- including the public test suite
# that monkeypatches ``agent.resume`` -- is unchanged.
#
# If preview decode fails for any reason (malformed token,
# expired, etc.), we DEFAULT TO ASK_USER. This keeps the
# historical behaviour where ``agent.resume`` raises a typed
# error during its own decode pass; that error code maps to
# the right HTTP status via ``_status_from_agent_error``.
# The token is still rejected by ``agent.resume``; we just do
# not pre-empt the rejection here.
token_name = "ask_user"
try:
preview_claims = PauseToken.decode(pause_token, agent._config)
token_name = (
getattr(preview_claims, "name", "ask_user") or "ask_user"
)
except Exception:
# Fall through with the default ask_user assumption; the
# downstream engine call will re-decode and raise the same
# typed error the historical contract returned.
pass
registration = agent._registered_interrupts.get(token_name)
is_ask_user = (
(registration is None)
or (registration.built_in and token_name == "ask_user")
)
# Parse + validate the body against the right schema. For the
# ask_user case we keep the existing AskUserResponse contract;
# for generic interrupts we look up the registered response
# schema and validate against it.
try:
raw_body = await raw_request.json()
except Exception as exc:
return EventSourceResponse(
_single_error_event(f"Invalid JSON body: {exc}", 400),
)
try:
if is_ask_user:
response_obj: Any = AskUserResponse.model_validate(raw_body)
else:
if registration is None:
return EventSourceResponse(
_single_error_event(
f"Token references unknown interrupt "
f"{token_name!r}",
400,
),
)
response_obj = registration.response_schema.model_validate(
raw_body,
)
except Exception as exc:
return EventSourceResponse(
_single_error_event(f"Invalid resume payload: {exc}", 400),
)
async def resume_generator() -> Any:
try:
if is_ask_user:
iterator = agent.resume(
pause_token=pause_token,
response=response_obj,
scope=scope,
)
else:
iterator = agent.resume_interrupt(
pause_token=pause_token,
response=response_obj,
scope=scope,
)
async for stream_event in iterator:
event_name = type(stream_event).__name__
yield {
"event": event_name,
"data": json.dumps(
_event_to_dict(stream_event), default=str,
),
}
except SecurityScopeError:
yield {
"event": "error",
"data": '{"detail": "Access denied: missing or invalid tenant scope"}',
}
except ScopeValidationError as exc:
yield {
"event": "error",
"data": json.dumps({"detail": str(exc)}),
}
except SymfonicAgentError as exc:
# v7.1.3 (Item 10.1): use the centralised mapping so
# ``bad_request`` and ``failed_dependency`` now surface as
# 400 / 424 (previously bad_request fell through to the
# default 400 by coincidence; failed_dependency leaked
# the request_hash-mismatch / MemorySaver-restart path
# as a generic 400). Unknown codes still default to 400
# here -- /resume is more restrictive than /chat because
# an unrecognised error during a token redemption is
# almost certainly a client-side payload bug.
status_code = _status_from_agent_error(exc)
if status_code == 500:
# Preserve the historical /resume default of 400 for
# unmapped codes; only known generic-error codes get 500.
status_code = 400
yield {
"event": "error",
"data": json.dumps({"detail": str(exc), "status": status_code}),
}
except Exception:
logger.exception("Resume event generator failed")
yield {
"event": "error",
"data": '{"detail": "Internal server error"}',
}
return EventSourceResponse(resume_generator())
@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._session_manager.list_sessions(scope.tenant_id)
except SecurityScopeError as exc:
raise HTTPException(
status_code=403,
detail="Access denied: missing or invalid tenant scope",
) from exc
except ScopeValidationError as exc:
raise HTTPException(status_code=400, detail=str(exc)) 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:
discovery = DiscoveryService(agent._orchestrator)
required = agent._config.domain.required_labels or None
blocks = await discovery.scan_status(
scope.to_memory_scope(),
required_labels=required,
)
return [
MemoryBlockStatus(
label=b.label,
exists=b.exists,
entry_count=b.entry_count,
last_updated=b.last_updated,
importance=b.importance,
layer=b.layer.value,
enabled=b.layer.value in agent._config.enabled_layers,
is_required=b.is_required,
)
for b in blocks
]
except SecurityScopeError as exc:
raise HTTPException(
status_code=403,
detail="Access denied: missing or invalid tenant scope",
) from exc
except ScopeValidationError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
except Exception as exc:
raise HTTPException(status_code=500, detail="Internal server error") 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, updates SOUL
schema from user corrections, 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 Jarvio 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 Jarvio 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 Jarvio 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,
),
)
soul_schema = (
dict(agent._config.domain.soul_schema)
if agent._config.domain.soul_schema
else None
)
try:
report = await consolidator.run(
memory_scope,
soul_schema=soul_schema,
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