Skip to content

symfonic.capabilities.memory.procedure_admin

procedure_admin

Scoped procedure inventory over the graph backend.

The review queue and HTTP list need the stored node, including lifecycle and custom properties. A retrieval projection loses those fields and may stop at 200 rows, so this read port queries the graph without a recall limit.

ProcedureAdminService 

ProcedureAdminService(graph: Any)

List one scope's stored procedures without composing an agent.

Source code in src/symfonic/capabilities/memory/procedure_admin.py 
28
29
def __init__(self, graph: Any) -> None:
    self._graph = GraphMemoryStore(graph)

amend_procedure async 

amend_procedure(scope: MemoryScope, node_id: str, *, steps: Any = None, properties: dict[str, Any] | None = None, active: bool | None = None) -> dict[str, Any] | None

Edit a visible procedure through the locked lifecycle transition.

Source code in src/symfonic/capabilities/memory/procedure_admin.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
async def amend_procedure(
    self, scope: MemoryScope, node_id: str, *, steps: Any = None,
    properties: dict[str, Any] | None = None, active: bool | None = None,
) -> dict[str, Any] | None:
    """Edit a visible procedure through the locked lifecycle transition."""
    scoped = tenant_scope(scope)
    # The layer method guards review state; this port also rejects nodes
    # outside the procedure surface before invoking that transition.
    candidate = await self._graph.backend.get_node(scoped, NodeId(node_id))
    if (candidate is None or candidate.layer != MemoryLayer.PROCEDURAL
            or is_retracted(candidate.properties)):
        return None
    from symfonic.memory.layers.procedural.layer import ProceduralLayer

    try:
        node = await ProceduralLayer(self._graph).amend_skill(
            scoped, NodeId(node_id), steps=steps, properties=properties, active=active,
        )
    except KeyError:
        return None
    props = node.properties or {}
    status, node_active = procedure_lifecycle(props)
    return {
        "id": str(node.id), "label": node.label, "steps": props.get("steps", []),
        "properties": props, "status": status, "active": node_active,
    }

delete_procedure async 

delete_procedure(scope: MemoryScope, node_id: str) -> bool

Atomically delete one exact-owner procedural row, if supported.

Source code in src/symfonic/capabilities/memory/procedure_admin.py 
37
38
39
40
41
42
43
async def delete_procedure(self, scope: MemoryScope, node_id: str) -> bool:
    """Atomically delete one exact-owner procedural row, if supported."""
    backend = self._graph.backend
    delete = getattr(backend, "delete_procedure_atomic", None)
    if not getattr(backend, "supports_atomic_procedure_delete", False) or not callable(delete):
        raise NotImplementedError("atomic procedure deletion unavailable")
    return await delete(tenant_scope(scope), NodeId(node_id))

get_procedure async 

get_procedure(scope: MemoryScope, node_id: str) -> dict[str, Any] | None

Read one procedural node in the authenticated tenant's scope.

Source code in src/symfonic/capabilities/memory/procedure_admin.py 
132
133
134
135
136
137
138
139
async def get_procedure(
    self, scope: MemoryScope, node_id: str
) -> dict[str, Any] | None:
    """Read one procedural node in the authenticated tenant's scope."""
    node = await self._graph.get_node(tenant_scope(scope), NodeId(node_id))
    if node is None or node.layer != MemoryLayer.PROCEDURAL:
        return None
    return self._row(node)

list_draft_procedures async 

list_draft_procedures(scope: MemoryScope) -> list[dict[str, Any]]

Project draft rows to the legacy review-queue wire shape.

Source code in src/symfonic/capabilities/memory/procedure_admin.py 
161
162
163
164
165
166
167
async def list_draft_procedures(self, scope: MemoryScope) -> list[dict[str, Any]]:
    """Project draft rows to the legacy review-queue wire shape."""
    rows = await self.list_procedures(scope, status="draft")
    fields = (
        "id", "label", "content", "steps", "status", "active", "importance", "created_at",
    )
    return [{field: row[field] for field in fields} for row in rows]

list_procedures async 

list_procedures(scope: MemoryScope, *, status: str | None = None) -> list[dict[str, Any]]

Preserve the legacy list row and status filter over all matching nodes.

Source code in src/symfonic/capabilities/memory/procedure_admin.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
async def list_procedures(
    self, scope: MemoryScope, *, status: str | None = None
) -> list[dict[str, Any]]:
    """Preserve the legacy list row and status filter over all matching nodes."""
    nodes = await self._graph.query_nodes(
        tenant_scope(scope),
        layer=MemoryLayer.PROCEDURAL,
        limit=None,
    )
    rows = []
    for node in nodes:
        props = node.properties or {}
        node_status, active = procedure_lifecycle(props)
        if (status is None and node_status != "approved") or (
            status not in (None, "all") and status != node_status
        ):
            continue
        rows.append(self._row(node))
    return rows

toggle_procedure async 

toggle_procedure(scope: MemoryScope, node_id: str) -> dict[str, Any] | None

Toggle a visible procedure under the same lock as review writes.

Source code in src/symfonic/capabilities/memory/procedure_admin.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
async def toggle_procedure(
    self, scope: MemoryScope, node_id: str,
) -> dict[str, Any] | None:
    """Toggle a visible procedure under the same lock as review writes."""
    scoped = tenant_scope(scope)
    async with review_lock_for(self._graph.backend):
        candidate = await self._graph.backend.get_node(scoped, NodeId(node_id))
        if (candidate is None or candidate.layer != MemoryLayer.PROCEDURAL
                or is_retracted(candidate.properties)):
            return None
        node = await self._graph.backend.get_node(scoped, NodeId(node_id))
        if (node is None or node.layer != MemoryLayer.PROCEDURAL
                or is_retracted(node.properties)):
            return None
        props = dict(node.properties or {})
        nested = props.get("record_metadata")
        if isinstance(nested, dict):
            reserved = {
                "status", "active", "procedure_review_authority",
                "procedure_review_actor", "procedure_lifecycle_origin",
            }
            props["record_metadata"] = {
                key: value for key, value in nested.items() if key not in reserved
            }
        status, active = procedure_lifecycle(props)
        props["active"] = not active if status == "approved" else False
        props["_last_edited_by"] = "user_manual_edit"
        try:
            node = await self._graph.update_node(
                scoped, NodeId(node_id), {
                    "properties": props,
                    "access_count": node.access_count + 1,
                },
            )
        except KeyError:
            return None
        props = dict(node.properties or {})
        status, active = procedure_lifecycle(props)
        return {
            "id": str(node.id), "label": node.label, "status": status,
            "active": active, "properties": props,
        }