Skip to content

symfonic.capabilities.memory.positional_rows

positional_rows

Classify stored positional v0 rows for an explicit kind-aware migration.

Owner decision (CORE-171, 2026-10-13): a v0 row moves to native v1 identity only when its owner is recoverable without guessing. A row whose kinds were lost is quarantined for explicit deletion. Nothing here dual-reads positional keys as v1; classification is pure and never touches a backend.

A v0 row is one of two shapes:

  • Explicit kinds -- a legacy TenantScope path whose stored kinds are not the positional stamp. The row names its whole owner, so it migrates as stored, at any depth.
  • Positional stamp -- tenant[/sub_tenant[/session]] written by the v0 capability, or a pre-v8 row with no path (tenant-global). Those kinds are what the v0 codec wrote, not what the owner was. The adopter must declare which native kinds the store ever wrote at that depth; exactly one candidate migrates, none or several quarantine. A depth-three row also quarantines when deeper scopes were ever composed against the store, because the pre-guard converter truncated them onto depth three.

PositionalOwnerDeclaration dataclass 

PositionalOwnerDeclaration(kinds_by_depth: Mapping[int, tuple[tuple[str, ...], ...]], max_written_depth: int)

The adopter's attestation of the kinds a store ever wrote positionally.

kinds_by_depth maps a positional depth (1-3) to every native kind path composed at that depth against this store. max_written_depth is the deepest typed scope ever composed against it, including scopes the old converter truncated. An undeclared depth is never inferred.

RowDisposition

Bases: StrEnum

What a planned migration does with one stored row.

RowVerdict dataclass 

RowVerdict(disposition: RowDisposition, target: str | None, reason: str)

A row's disposition; target is a storage key, never row content.

claim_verdict 

claim_verdict(record: str | None, stored: object, verdict: RowVerdict, owners: Mapping[str, str | None]) -> RowVerdict

A claim follows its record node's resolved owner, never its own path alone.

It migrates only when that node migrates to, or already holds, exactly the claim's own target; a missing, quarantined or different owner quarantines it.

Source code in src/symfonic/capabilities/memory/positional_rows.py 
220
221
222
223
224
225
226
227
228
229
230
231
def claim_verdict(record: str | None, stored: object, verdict: RowVerdict,
                  owners: Mapping[str, str | None]) -> RowVerdict:
    """A claim follows its record node's resolved owner, never its own path alone.

    It migrates only when that node migrates to, or already holds, exactly the
    claim's own target; a missing, quarantined or different owner quarantines it.
    """
    if verdict.disposition is not RowDisposition.MIGRATE:
        return verdict
    if record is None or owners.get(record) != verdict.target:
        return _quarantine(stored, "claim record does not migrate to the same owner")
    return verdict

classify_stored_path 

classify_stored_path(stored: str | None, tenant_id: str, declaration: PositionalOwnerDeclaration) -> RowVerdict

Decide one row from its stored owner path alone.

Source code in src/symfonic/capabilities/memory/positional_rows.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def classify_stored_path(
    stored: str | None, tenant_id: str, declaration: PositionalOwnerDeclaration
) -> RowVerdict:
    """Decide one row from its stored owner path alone."""
    if type(declaration) is not PositionalOwnerDeclaration:
        raise MemoryContractError("classification needs a PositionalOwnerDeclaration")
    if stored is not None and type(stored) is not str:
        return _quarantine(stored, "stored owner is not a string")
    # A v0 kind is adopter text and may spell either header; only a path
    # without the v0 delimiter can be a native or quarantine key.
    settled = bool(stored) and LEGACY_SCOPE_DELIMITER not in stored
    if settled and stored.startswith(QUARANTINE_HEADER):
        return RowVerdict(RowDisposition.UNTOUCHED, None, "already quarantined")
    if settled and stored.startswith(_NATIVE_HEADER):
        try:
            StorageScopeKey.parse(stored)
        except MemoryContractError:
            return RowVerdict(RowDisposition.UNTOUCHED, None, "unparseable native key")
        return RowVerdict(RowDisposition.UNTOUCHED, None, "already native")
    if not stored:
        if type(tenant_id) is not str or not tenant_id:
            return _quarantine(stored, "pre-v8 row has no tenant")
        pairs: tuple[tuple[str, str], ...] = (("tenant", tenant_id),)
    else:
        parts = stored.split(LEGACY_SCOPE_DELIMITER)
        if len(parts) % 2 or not all(parts):
            return _quarantine(stored, "malformed positional path")
        pairs = tuple(zip(parts[0::2], parts[1::2], strict=True))
        if pairs[0][1] != tenant_id:
            return _quarantine(stored, "stored root disagrees with the row tenant")
    kinds = tuple(kind for kind, _ in pairs)
    ids = tuple(identifier for _, identifier in pairs)
    if kinds != LEGACY_SCOPE_KINDS[: len(kinds)]:
        return _migrate(stored, pairs, "explicit legacy kinds")
    candidates = declaration.kinds_by_depth.get(len(kinds), ())
    if not candidates:
        return _quarantine(stored, "kind lost: depth not declared")
    if len(candidates) > 1:
        return _quarantine(stored, "kind lost: several kinds declared at this depth")
    if len(kinds) == len(LEGACY_SCOPE_KINDS) and declaration.max_written_depth > len(kinds):
        return _quarantine(stored, "owner lost: deeper scopes were truncated here")
    pairs = tuple(zip(candidates[0], ids, strict=True))
    return _migrate(stored, pairs, "declared positional kinds")

covering_keys 

covering_keys(target: str) -> tuple[str, ...]

target's own key and each ancestor key: every key whose forget erased it.

Source code in src/symfonic/capabilities/memory/positional_rows.py 
187
188
189
def covering_keys(target: str) -> tuple[str, ...]:
    """``target``'s own key and each ancestor key: every key whose forget erased it."""
    return tuple(StorageScopeKey.parse(target).ancestor_keys())

edge_verdict 

edge_verdict(source: str | None, target: str | None, stored: object, verdict: RowVerdict, owners: Mapping[str, str | None]) -> RowVerdict

An edge migrates only with both endpoints visible at its new owner.

owners maps a node id to the native owner it resolves to after the migration; a node that is missing, quarantined or unmoved maps to None.

Source code in src/symfonic/capabilities/memory/positional_rows.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def edge_verdict(source: str | None, target: str | None, stored: object, verdict: RowVerdict,
                 owners: Mapping[str, str | None]) -> RowVerdict:
    """An edge migrates only with both endpoints visible at its new owner.

    ``owners`` maps a node id to the native owner it resolves to after the
    migration; a node that is missing, quarantined or unmoved maps to None.
    """
    if verdict.disposition is not RowDisposition.MIGRATE:
        return verdict
    edge_owner = scope_from_storage_path(verdict.target)
    for endpoint in (source, target):
        owner = owners.get(endpoint) if endpoint is not None else None
        if owner is None or not scope_from_storage_path(owner).covers(edge_owner):
            return _quarantine(stored, "edge endpoint does not migrate with it")
    return verdict

erased_verdict 

erased_verdict(stored: object, verdict: RowVerdict, erased: frozenset[str]) -> RowVerdict

A row whose new owner a recorded native forget erased never moves there.

Moving it would put back content the forget of that owner, or of any ancestor of it, was asked to erase (CORE-171 owner decision, 2026-10-14).

Source code in src/symfonic/capabilities/memory/positional_rows.py 
192
193
194
195
196
197
198
199
200
def erased_verdict(stored: object, verdict: RowVerdict, erased: frozenset[str]) -> RowVerdict:
    """A row whose new owner a recorded native forget erased never moves there.

    Moving it would put back content the forget of that owner, or of any
    ancestor of it, was asked to erase (CORE-171 owner decision, 2026-10-14).
    """
    if verdict.disposition is not RowDisposition.MIGRATE or verdict.target not in erased:
        return verdict
    return _quarantine(stored, "owner erased by a native forget")

origin_ids 

origin_ids(stored: object, tenant_id: str) -> tuple[str, ...] | None

The ids a v0 owner names, root first; None when they cannot be read.

Every native owner a declaration could give the row carries exactly these ids, whatever kinds it declares.

Source code in src/symfonic/capabilities/memory/positional_rows.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
def origin_ids(stored: object, tenant_id: str) -> tuple[str, ...] | None:
    """The ids a v0 owner names, root first; None when they cannot be read.

    Every native owner a declaration could give the row carries exactly these
    ids, whatever kinds it declares.
    """
    if stored is None or stored == "":
        return (tenant_id,)
    if type(stored) is not str:
        return None
    parts = stored.split(LEGACY_SCOPE_DELIMITER)
    if len(parts) % 2 or not all(parts):
        return None
    return tuple(parts[1::2])

quarantine_key 

quarantine_key(stored: str | None) -> str

A key no positional or native read can match, carrying its origin.

Source code in src/symfonic/capabilities/memory/positional_rows.py 
109
110
111
112
def quarantine_key(stored: str | None) -> str:
    """A key no positional or native read can match, carrying its origin."""
    origin = "" if stored is None else stored
    return f"{QUARANTINE_HEADER}{origin.encode('utf-8').hex()}/"

quarantined_origin 

quarantined_origin(key: str) -> str | None

The stored path a quarantine key replaced; None for a pre-v8 row.

Source code in src/symfonic/capabilities/memory/positional_rows.py 
115
116
117
118
119
120
121
122
123
124
125
126
def quarantined_origin(key: str) -> str | None:
    """The stored path a quarantine key replaced; ``None`` for a pre-v8 row."""
    if type(key) is not str or not key.startswith(QUARANTINE_HEADER) or not key.endswith("/"):
        raise MemoryContractError("not a quarantined positional row")
    body = key[len(QUARANTINE_HEADER):-1]
    try:
        origin = bytes.fromhex(body).decode("utf-8")
    except (ValueError, UnicodeError) as error:
        raise MemoryContractError("malformed quarantine key") from error
    if quarantine_key(origin or None) != key:
        raise MemoryContractError("noncanonical quarantine key")
    return origin or None