Skip to content

symfonic.agent.cutover.compat_extraction

compat_extraction

Inline extraction for SymfonicAgent's kernel route (CORE-227).

Legacy _consolidate_impl files the operations a reply carries in its MEMORY_EXTRACT/GRAPH_OPERATIONS block. The kernel route parsed and stripped that block and filed nothing, because the bundle folded no extractor. This is that extractor: the native :class:LegacyGraphOperationsAdapter over a :class:MemoryExtractionService with no model, so its records are minted, scoped and scrubbed by the native service and written through the bundle's own write port.

No extraction model call. Legacy asks no second model when a reply carries no block, so this extractor does not either. extract reads the directive legacy's own scanner would strip from the reply -- a trailing bare {"ops": ...} object, a loose GRAPH_OPERATIONS tail, a fenced JSON tail -- which the kernel's tag filter does not recognise and so leaves in the turn text, and files its operations as :meth:parse_inline does. That is the same :func:~symfonic.agent.response_presentation.present_response the projection uses to report and strip them, so an op the response reports is an op this extractor was offered. A reply with no directive answers the native service's "no extraction model is bound" result.

The legacy dialect's own leniencies, and only those. _consolidate_impl infers layer from type, falls back from label to payload.name then payload.id, reads operation_type/data as aliases, and merges an upsert_node payload's other fields into its properties with payload.properties winning. Those are normalised here, before the adapter sees the operation; the adapter's scope, provenance and classification refusals are unchanged, and a scope or provenance field is never moved into the properties the adapter does not check for one.

Refusals are not silent. The consolidation stage replaces any result that drops an operation with the fallback model's, which here is nothing, so one refused operation would lose the accepted ones beside it. A dropped envelope is therefore re-read one operation at a time: accepted records are kept, and each refusal is logged by index and reason (never content) and carried on the result's reason. The result carries no dropped, so the shared stage keeps it under its unchanged fallback rule. The cost is the stage trace: it reads discarded=0, and it carries the reason only when nothing was accepted; a refusal beside an accepted op is on the log alone. The re-read keeps the envelope's own bounds: records past max_records are refused as the native service refuses them, and an envelope that mixes the native and legacy schemas is never re-read: the adapter's whole-envelope refusal stands, carried on reason like the others, so the stage makes no fallback read of the reply's untagged tail either.

CompatibilityExtractor 

CompatibilityExtractor()

The extractor SymfonicAgent's consolidating kernel bundle folds.

Source code in src/symfonic/agent/cutover/compat_extraction.py 
123
124
def __init__(self) -> None:
    self._adapter = LegacyGraphOperationsAdapter(MemoryExtractionService())

extract async 

extract(request: ExtractionRequest) -> ExtractionResult

The reply's untagged directive, else the native no-model result.

Source code in src/symfonic/agent/cutover/compat_extraction.py 
126
127
128
129
130
131
async def extract(self, request: ExtractionRequest) -> ExtractionResult:
    """The reply's untagged directive, else the native no-model result."""
    ops = present_response(request.assistant_message).ops
    if ops:
        return self.parse_inline(ops, request)
    return await self._adapter.extract(request)

normalise_legacy_operation 

normalise_legacy_operation(operation: Any) -> Any

Apply the legacy dialect's leniencies to one type/payload op.

Anything that is not a legacy-shaped mapping is returned unchanged, so the native action dialect and malformed input reach the adapter as sent.

Source code in src/symfonic/agent/cutover/compat_extraction.py 
 76
 77
 78
 79
 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
def normalise_legacy_operation(operation: Any) -> Any:
    """Apply the legacy dialect's leniencies to one ``type``/``payload`` op.

    Anything that is not a legacy-shaped mapping is returned unchanged, so the
    native ``action`` dialect and malformed input reach the adapter as sent.
    """
    if not isinstance(operation, Mapping) or "action" in operation:
        return operation
    kind = operation.get("type", operation.get("operation_type"))
    if not isinstance(kind, str):
        return operation
    payload = operation.get("payload") or operation.get("data")
    normalised = {
        key: value for key, value in operation.items() if key not in {"operation_type", "data"}
    }
    normalised["type"] = kind
    if isinstance(payload, Mapping):
        payload = dict(payload)
        normalised["payload"] = payload
    if not normalised.get("layer") and kind in _LAYER_BY_TYPE:
        normalised["layer"] = _LAYER_BY_TYPE[kind]
    if not normalised.get("label") and isinstance(payload, Mapping):
        fallback = payload.get("name") or payload.get("id")
        if isinstance(fallback, str) and fallback.strip():
            normalised["label"] = " ".join(fallback.split())
    if kind == "upsert_node" and isinstance(payload, dict):
        inner = payload.get("properties", {})
        # A scope or provenance claim stays in the payload, where the adapter
        # refuses it; folding it into properties would launder it past that.
        extra = {
            k: v
            for k, v in payload.items()
            if k not in _UPSERT_PAYLOAD_FIELDS and k not in _FORBIDDEN_FIELDS
        }
        if extra and isinstance(inner, Mapping):
            # Legacy: ``{payload minus properties} | properties``.
            payload["properties"] = {**extra, **inner}
            for key in extra:
                del payload[key]
    return normalised