symfonic.capabilities.memory.positional_audit

positional_audit

Audit evidence for CORE-171 positional migrations and quarantine deletions.

Every operation that moves or deletes a row records itself through the audit= logger its caller passes: any object with an async record(entry), which is the shape of :class:symfonic.core.observability.audit.AuditLogger and of the durable logger the scaffold registers (pass get_audit_logger()). Without one the operation refuses before reading, so a quarantine decision or a deletion never happens without a trail. The entry is :class:PositionalAuditEntry, which carries the field names those loggers read; this layer may not import the core model itself.

initiated is recorded inside the backend transaction, after the rows are re-checked and before any of them changes; a logger that raises there aborts the transaction and nothing moves. That entry is the justification: tenant, each row's kind and id, its prior owner path and the reason it was moved or deleted, and a digest binding the set. completed or failed follows best-effort, as for every audited deletion here: a logger that fails after commit is logged, not raised, because the rows have already changed.

Owner paths are identifiers, not memory content; record text, embeddings and backend error text are never recorded. The actor is whatever the caller names and is not authenticated here. Without one the entry says so and records only the explicit operator operation that ran. Durability is the logger's: an in-memory logger keeps nothing past the process.

PositionalAudit

PositionalAudit(audit: Any, tenant_id: str, actor: Any, operation: str)

One operation's audit trail, bound to its logger, tenant and actor.

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

def __init__(self, audit: Any, tenant_id: str, actor: Any, operation: str) -> None:
    if audit is None or not callable(getattr(audit, "record", None)):
        raise MemoryContractError(NO_AUDIT)
    if actor is not None and (type(actor) is not str or not actor or len(actor) > 256):
        raise MemoryContractError("actor must be a nonempty string of at most 256 chars")
    self._audit, self._tenant, self._actor = audit, tenant_id, actor
    self._operation = operation
    self._digest: str | None = None
    self._count = 0

initiated async

initiated(rows: list[list[Any]], **metadata: Any) -> None

Record the justification; raising here must abort the operation.

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

async def initiated(self, rows: list[list[Any]], **metadata: Any) -> None:
    """Record the justification; raising here must abort the operation."""
    self._digest, self._count = row_digest(self._tenant, rows), len(rows)
    await self._audit.record(self._entry("initiated", {
        "row_fields": ["kind", "row_id", "prior_owner", "new_owner", "reason"],
        "rows": rows, "row_count": len(rows), **metadata,
    }))

settled async

settled(outcome: str, **metadata: Any) -> None

Record the result after commit or abort; never raises.

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

async def settled(self, outcome: str, **metadata: Any) -> None:
    """Record the result after commit or abort; never raises."""
    if self._digest is None:
        return
    try:
        await self._audit.record(self._entry(outcome, {
            "row_count": self._count, **metadata}))
    except Exception:  # noqa: BLE001 - the rows have already settled
        _log.error("positional audit could not record %s %s for digest %s",
                   self._operation, outcome, self._digest)

PositionalAuditEntry dataclass

PositionalAuditEntry(tenant_id: str, user_id: str | None, action: str, resource_type: str, resource_id: str | None, metadata: dict[str, Any], timestamp: datetime = (lambda: datetime.now(UTC))(), ip_address: str | None = None, user_agent: str | None = None)

One audit event, field-compatible with the core AuditLogEntry.

to_dict

to_dict() -> dict[str, Any]

A JSON-safe dict, as AuditLogEntry.to_dict returns.

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

def to_dict(self) -> dict[str, Any]:
    """A JSON-safe dict, as ``AuditLogEntry.to_dict`` returns."""
    return {
        "tenant_id": self.tenant_id, "user_id": self.user_id, "action": self.action,
        "resource_type": self.resource_type, "resource_id": self.resource_id,
        "timestamp": self.timestamp.isoformat(), "metadata": self.metadata,
        "ip_address": self.ip_address, "user_agent": self.user_agent,
    }

prior_owner

prior_owner(stored: Any) -> str | None

The owner path a quarantine key replaced; a malformed key names itself.

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

def prior_owner(stored: Any) -> str | None:
    """The owner path a quarantine key replaced; a malformed key names itself."""
    from symfonic.capabilities.memory.positional_rows import quarantined_origin

    try:
        return quarantined_origin(stored)
    except MemoryContractError:
        return f"malformed quarantine key: {stored!r}"[:512]

row_digest

row_digest(tenant_id: str, rows: list[list[Any]]) -> str

A digest of exactly the rows an operation named, in order.

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

def row_digest(tenant_id: str, rows: list[list[Any]]) -> str:
    """A digest of exactly the rows an operation named, in order."""
    body = json.dumps([tenant_id, rows], ensure_ascii=True, separators=(",", ":"))
    return hashlib.sha256(body.encode()).hexdigest()