symfonic.capabilities.memory.procedure_publish

procedure_publish

Publish org skills to the platform corpus, reviewed and scrubbed.

The write side of cross-org learning. A skill travels approved -> proposed -> published and every transition is explicit:

• propose: the org's approved+active skill is scrubbed (PII redacted to stable tokens; unresolvable rows refuse the proposal outright) and a proposal node is written in the SOURCE tenant. Nothing crosses yet.
• review: an org owner (or platform admin) approves or refuses. Approve copies the scrubbed snapshot to the platform tenant with provenance (source org/node, reviewer, scrub report); refuse records the reason. Decided proposals cannot be re-decided.
• revoke: the source org (or a platform admin) deactivates the platform copy. Recall filters active rows, so revocation takes effect on the next read; the row stays for audit.

All transitions run under the same review lock as skill review, so a concurrent skill edit cannot interleave with a publish decision. Statuses (pending-platform-review, publication-approved/refused) are deliberately outside the skill lifecycle vocabulary: procedure_lifecycle resolves anything but approved as inactive, so proposals and refused rows can never be recalled as skills.

ProceduralPublishService

ProceduralPublishService(graph: Any, scrubber: Any = None)

Review-gated publication of org skills to the platform corpus.

Parameters:

Name	Type	Description	Default
graph	Any	graph backend (reads/writes procedural nodes).	required
scrubber	Any	redaction engine with scrub_row(row, drop_on_doubt=...) returning {scrubbed, match_counts, unresolved, dropped_reason}. Injected (not imported) so deployments own their pattern set.	None

Source code in src/symfonic/capabilities/memory/procedure_publish.py

def __init__(self, graph: Any, scrubber: Any = None) -> None:
    from symfonic.memory.graph.store import GraphMemoryStore
    from symfonic.memory.privacy import CorpusScrubber

    # Raw backend in, store interface inside -- the same wrap
    # ProcedureAdminService applies, so callers pass backends[0].
    # Default scrubber: publication without redaction is not an
    # option the API offers. A deployment that tuned its own
    # patterns passes its instance instead.
    self._graph = GraphMemoryStore(graph)
    self._scrubber = scrubber if scrubber is not None else CorpusScrubber()

list_proposals async

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

Pending proposals in one tenant (the review queue).

Source code in src/symfonic/capabilities/memory/procedure_publish.py

async def list_proposals(self, scope: Any) -> list[dict[str, Any]]:
    """Pending proposals in one tenant (the review queue)."""
    scoped = tenant_scope(scope)
    nodes = await self._graph.query_nodes(
        scoped, layer=MemoryLayer.PROCEDURAL, limit=None
    )
    return [
        self._row(node)
        for node in nodes
        if (node.properties or {}).get("status") == PROPOSAL_STATUS
    ]

list_published async

list_published() -> list[dict[str, Any]]

Active platform skills (what cross-org recall can see).

Source code in src/symfonic/capabilities/memory/procedure_publish.py

async def list_published(self) -> list[dict[str, Any]]:
    """Active platform skills (what cross-org recall can see)."""
    nodes = await self._graph.query_nodes(
        platform_scope(), layer=MemoryLayer.PROCEDURAL, limit=None
    )
    out = []
    for node in nodes:
        props = node.properties or {}
        status, active = procedure_lifecycle(props)
        if status == "approved" and active and props.get("publication"):
            out.append(self._row(node))
    return out

propose_publication async

propose_publication(scope: Any, node_id: str, *, proposer: str) -> dict[str, Any]

Scrub an approved skill and record a platform proposal for it.

Source code in src/symfonic/capabilities/memory/procedure_publish.py

async def propose_publication(
    self, scope: Any, node_id: str, *, proposer: str
) -> dict[str, Any]:
    """Scrub an approved skill and record a platform proposal for it."""

    scoped = tenant_scope(scope)
    node = await self._graph.get_node(scoped, NodeId(node_id))
    self._require_skill(node, node_id)
    status, active = procedure_lifecycle(node.properties or {})
    if status != "approved" or not active:
        raise PublicationRefused(
            f"only approved active skills can be proposed (node {node_id} "
            f"is {status})"
        )
    scrubbed = self._scrubber.scrub_row(
        {
            "label": node.label,
            "content": (node.properties or {}).get("content", ""),
            "steps": (node.properties or {}).get("steps", []),
            "context": (node.properties or {}).get("context", ""),
        },
        drop_on_doubt=True,
    )
    if scrubbed.scrubbed is None:
        raise PublicationRefused(
            f"scrubber refused the skill: {scrubbed.dropped_reason} "
            f"(matches={dict(scrubbed.match_counts)})"
        )
    from symfonic.memory.layers.procedural.sharing import proposal_node

    proposal = proposal_node(
        scoped, node, scrubbed.scrubbed,
        {
            "match_counts": dict(scrubbed.match_counts),
            "unresolved": scrubbed.unresolved,
        },
        proposer,
    )
    async with review_lock_for(self._graph):
        created = await self._graph.add_node(scoped, proposal)
    return self._row(created)

review_publication async

review_publication(scope: Any, proposal_id: str, *, decision: str, reviewer: str, reason: str = '') -> dict[str, Any]

Approve (publish the scrubbed copy) or refuse a proposal.

Only the source tenant's owners may decide; the transport layer enforces that from the verified principal before calling here.

Source code in src/symfonic/capabilities/memory/procedure_publish.py

async def review_publication(
    self,
    scope: Any,
    proposal_id: str,
    *,
    decision: str,
    reviewer: str,
    reason: str = "",
) -> dict[str, Any]:
    """Approve (publish the scrubbed copy) or refuse a proposal.

    Only the source tenant's owners may decide; the transport layer
    enforces that from the verified principal before calling here.
    """

    if decision not in ("approve", "refuse"):
        raise PublicationRefused(f"unknown decision {decision!r}")
    scoped = tenant_scope(scope)
    proposal = await self._graph.get_node(scoped, NodeId(proposal_id))
    if proposal is None:
        raise PublicationNotFound(f"proposal {proposal_id} not found")
    current = (proposal.properties or {}).get("status")
    if current != PROPOSAL_STATUS:
        raise PublicationConflict(
            f"proposal {proposal_id} already decided ({current})"
        )
    publication = dict((proposal.properties or {}).get("publication", {}))
    if decision == "refuse":
        return await self._transition(
            scoped, proposal, REFUSED_STATUS,
            {"reviewed_by": reviewer, "reviewed_at": _now(), "reason": reason},
        )
    snapshot = publication.get("snapshot") or {}
    platform_scope_ = platform_scope()
    # No silent duplicates: an identical label already published resolves
    # to a refusal naming the existing copy (re-approving the same skill
    # -- retry, double review -- must not mint a second platform row).
    # Normalized the same way the read path dedupes tenant-vs-platform.
    wanted = str(snapshot.get("label", "") or "").strip().casefold()
    if wanted:
        existing = await self._graph.query_nodes(
            platform_scope_, layer=MemoryLayer.PROCEDURAL, limit=None
        )
        for node in existing:
            props = node.properties or {}
            status, active = procedure_lifecycle(props)
            if status != "approved" or not active:
                continue
            if str(node.label or "").strip().casefold() == wanted:
                raise PublicationConflict(
                    f"already published as platform node {node.id}"
                )
    from symfonic.memory.layers.procedural.sharing import platform_copy_node

    copy = platform_copy_node(
        snapshot,
        {
            **publication,
            "reviewed_by": reviewer,
            "reviewed_at": _now(),
            "reason": reason,
        },
        proposal.importance,
    )
    async with review_lock_for(self._graph):
        created = await self._graph.add_node(platform_scope_, copy)
        row = await self._transition(
            scoped, proposal, APPROVED_STATUS,
            {
                "reviewed_by": reviewer, "reviewed_at": _now(), "reason": reason,
                "platform_node_id": str(created.id),
            },
            lock_held=True,
        )
    row["platform_node_id"] = str(created.id)
    return row

revoke_publication async

revoke_publication(caller_scope: Any, platform_node_id: str, *, actor: str) -> dict[str, Any]

Deactivate a platform copy. Only the publishing org (or a platform admin calling with its tenant header) may revoke, checked against the copy's own provenance -- never against a caller claim.

Source code in src/symfonic/capabilities/memory/procedure_publish.py

async def revoke_publication(
    self, caller_scope: Any, platform_node_id: str, *, actor: str
) -> dict[str, Any]:
    """Deactivate a platform copy. Only the publishing org (or a platform
    admin calling with its tenant header) may revoke, checked against
    the copy's own provenance -- never against a caller claim."""
    scoped = tenant_scope(caller_scope)
    node = await self._graph.get_node(platform_scope(), NodeId(platform_node_id))
    if node is None:
        raise PublicationNotFound(f"platform node {platform_node_id} not found")
    provenance = (node.properties or {}).get("publication") or {}
    if not provenance.get("source_tenant"):
        raise PublicationNotFound(
            f"node {platform_node_id} is not a platform publication"
        )
    if provenance["source_tenant"] != scoped.tenant_id:
        raise PublicationForbidden(
            "only the publishing org may revoke this publication"
        )
    async with review_lock_for(self._graph):
        updated = await self._graph.update_node(
            platform_scope(), node.id,
            {"active": False, "revoked": {"by": actor, "at": _now()}},
        )
    return self._row(updated)