Skip to content

symfonic.capabilities.memory.legacy_extraction

legacy_extraction

Explicit, safe migration for the bundled graph-operation extraction schema.

The bundled MEMORY_EXTRACT and GRAPH_OPERATIONS templates predate the native extraction wire contract. They emit type and payload rather than action and properties. This adapter translates only the four legacy operation kinds that become durable native records. It never performs graph mutations itself, and it never accepts a caller-supplied scope or provenance: the delegated native service mints those from ExtractionRequest.

Unsupported or malformed legacy operations deliberately produce a dropped result. The post-response stage treats that as a signal to run its configured extraction-model fallback, rather than allowing a stripped but unusable tagged reply to suppress durable extraction for the turn.

LegacyGraphOperationsAdapter 

LegacyGraphOperationsAdapter(native: MemoryExtractionService)

Make a native extractor accept the bundled legacy directive explicitly.

Compose this around :class:MemoryExtractionService only when rendering a bundled legacy extraction template. Native action/properties operations retain their normal service path, which lets a template migrate independently without a behavior switch in the post-response stage.

Source code in src/symfonic/capabilities/memory/legacy_extraction.py 
54
55
def __init__(self, native: MemoryExtractionService) -> None:
    self._native = native

extract async 

extract(request: ExtractionRequest) -> ExtractionResult

Delegate fallback extraction without changing its provider contract.

Source code in src/symfonic/capabilities/memory/legacy_extraction.py 
57
58
59
async def extract(self, request: ExtractionRequest) -> ExtractionResult:
    """Delegate fallback extraction without changing its provider contract."""
    return await self._native.extract(request)

parse_inline 

parse_inline(operations: tuple[Mapping[str, Any], ...], request: ExtractionRequest) -> ExtractionResult

Translate an all-legacy envelope or retain the native inline path.

Source code in src/symfonic/capabilities/memory/legacy_extraction.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 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
def parse_inline(
    self, operations: tuple[Mapping[str, Any], ...], request: ExtractionRequest
) -> ExtractionResult:
    """Translate an all-legacy envelope or retain the native inline path."""
    if not operations:
        return self._native.parse_inline(operations, request)

    legacy = [
        isinstance(operation, Mapping)
        and "type" in operation
        and "action" not in operation
        for operation in operations
    ]
    native = [
        isinstance(operation, Mapping)
        and "action" in operation
        and "type" not in operation
        for operation in operations
    ]
    if all(native):
        return self._native.parse_inline(operations, request)
    if not all(legacy):
        return self._rejected(
            request,
            "the tagged operations mix native, legacy, or malformed schemas",
        )

    translated: list[Mapping[str, Any]] = []
    dropped: list[tuple[str, str]] = []
    for index, operation in enumerate(operations):
        converted, reason = _translate(operation)
        if reason:
            dropped.append((f"op[{index}]", reason))
        elif converted is not None:
            translated.append(converted)
    if dropped:
        return ExtractionResult(
            scope=request.scope,
            turn=request.turn,
            dropped=tuple(dropped),
            family=ProviderFamily.TEXT,
            reason="the inline legacy extraction schema was rejected",
        )
    return self._native.parse_inline(tuple(translated), request)