Skip to content

symfonic.capabilities.memory.native_deep_risk_runner

native_deep_risk_runner

Opt-in named native DEEP risk phase over an exact PostgreSQL tenant owner.

This is a single phase, not the generic DEEP roster or generated worker route. The caller supplies verified tenant authority via DeepTenantOwner. The lease, guarded node patches, and journal replay use one pool transaction.

NativeDeepRiskResult dataclass 

NativeDeepRiskResult(tagged: int | None, replayed: bool)

A fresh count or an exact replay with no reconstructed count.

NativeDeepRiskRunner 

NativeDeepRiskRunner(graph: PostgresGraphBackend, owner: DeepTenantOwner, *, lease_ttl_seconds: float = 30.0)

Run only the exact-owner native risk tag phase with durable exclusion.

Source code in src/symfonic/capabilities/memory/native_deep_risk_runner.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def __init__(
    self,
    graph: PostgresGraphBackend,
    owner: DeepTenantOwner,
    *,
    lease_ttl_seconds: float = 30.0,
) -> None:
    if type(owner) is not DeepTenantOwner:
        raise MemoryContractError("native DEEP requires a verified tenant owner")
    if type(graph) is not PostgresGraphBackend:
        raise MemoryContractError("native DEEP requires an exact PostgresGraphBackend")
    if (
        type(lease_ttl_seconds) not in (float, int)
        or not isfinite(lease_ttl_seconds)
        or lease_ttl_seconds <= 0
    ):
        raise MemoryContractError("native DEEP requires a positive lease TTL")
    self._graph = graph
    self._owner = owner
    self._phase = NativeTagRiskPhase(JournalledGraph(graph), owner)
    self._domain = PostgresReceiptDomain(graph.pool)
    self._leases = PostgresReceiptLeases(self._domain)
    self._ttl = float(lease_ttl_seconds)

run async 

run(scope: MemoryScope, operation: MemoryOperation | None = None) -> NativeDeepRiskResult | None

Return a fresh result, exact replay, or None for lease contention.

Source code in src/symfonic/capabilities/memory/native_deep_risk_runner.py 
 80
 81
 82
 83
 84
 85
 86
 87
 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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
async def run(
    self,
    scope: MemoryScope,
    operation: MemoryOperation | None = None,
) -> NativeDeepRiskResult | None:
    """Return a fresh result, exact replay, or None for lease contention."""
    if type(scope) is not MemoryScope or scope != self._owner.scope:
        raise MemoryContractError("native DEEP cycle must match its exact tenant owner")
    require_operation(operation, MemoryOperationVerb.COMMIT_CYCLE, self._owner.scope)
    expected = MemoryOperation(
        operation.operation_id,
        MemoryOperationVerb.COMMIT_CYCLE,
        self._owner.scope,
        NATIVE_DEEP_RISK_INTENT,
    )
    if operation.content_digest != expected.content_digest:
        raise MemoryContractError("native DEEP operation requires the fixed tag-risk intent")
    pool = self._graph.pool
    if pool.raw_pool is None:
        raise MemoryContractError("native DEEP requires an open PostgreSQL pool")
    if bound_connection(pool) is not None:
        raise MemoryContractError("native DEEP cannot join an ambient transaction")
    await _require_lease_ready(pool)
    if await self._domain.reconcile(operation) is MemoryOperationOutcome.COMMITTED:
        return NativeDeepRiskResult(tagged=None, replayed=True)
    lease = await self._leases.acquire(
        scope,
        owner=new_owner_token(),
        ttl_seconds=self._ttl,
    )
    if lease is None:
        return None
    journal = CycleJournal()  # No cycle=deep: that selects v0 WorksetOverlay.
    fence = Fence(
        lambda: self._leases.holds(lease),
        lambda: self._leases.hold_for_update(lease),
    )
    try:
        reserved = await self._domain.reserve(operation, owner=lease.owner)
        if not reserved:
            return NativeDeepRiskResult(tagged=None, replayed=True)
        async with heartbeat(self._leases, lease), held(fence), journalled(journal):
            context = PhaseContext(scope, ConsolidationCycle.DEEP, datetime.now(UTC))
            tagged = await self._phase.run(context)
            await fence.check("finishing native DEEP risk phase")
            await commit_native_deep_risk(
                journal,
                self._graph,
                self._domain,
                self._leases,
                fence,
                operation,
            )
            return NativeDeepRiskResult(tagged=tagged, replayed=False)
    finally:
        journal.discard()
        self._domain.abandon_reservation(operation, owner=lease.owner)
        await self._leases.release(lease)