Skip to content

symfonic.capabilities.memory.receipt_extraction

receipt_extraction

Validated extracted candidates for one staged PostgreSQL receipt turn.

extracted_records async 

extracted_records(extractor: Any, context: Any, scope: Any) -> tuple[MemoryRecord, ...]

Fail closed before WRITE; the permit binds this batch to its invocation.

Source code in src/symfonic/capabilities/memory/receipt_extraction.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
async def extracted_records(extractor: Any, context: Any, scope: Any) -> tuple[MemoryRecord, ...]:
    """Fail closed before WRITE; the permit binds this batch to its invocation."""
    request = context.request
    run_id, delivered = request.run_id, context.outcome_text
    if (request.scope != scope or type(run_id) is not str or not run_id
            or type(delivered) is not str or type(request.prompt) is not str):
        raise ContractViolationError("receipt extraction turn binding refused")
    result = await extractor.extract(ExtractionRequest(
        scope, user_message=request.prompt, assistant_message=delivered,
    ))
    if asyncio.current_task().cancelling():
        raise asyncio.CancelledError
    if (type(result) is not ExtractionResult or result.scope != scope
            or result.turn != 0 or result.degraded or not result.memories
            or len(result.memories) > 50):
        raise ContractViolationError("receipt extraction produced no valid fact batch")
    records = result.records
    if (not records or len(records) > 50
            or any(type(record) is not MemoryRecord or record.scope_path != scope.path
                   for record in records)):
        raise ContractViolationError("receipt extraction fact scope refused")
    # Extraction's generic turn=0 IDs can recur across invocations. The sealed
    # receipt uses a run-bound identity and retains validated scrubbed content.
    records = tuple(replace(record, record_id=f"{run_id}:fact:{index}",
                            metadata={**record.metadata, "turn_id": run_id})
                    for index, record in enumerate(records))
    WriteRequest(scope, records).validate()
    return records

require_extractor 

require_extractor(extractor: Any, *, staged: bool) -> None

Admit only an exact bound service on the exact staged PG factory path.

Source code in src/symfonic/capabilities/memory/receipt_extraction.py 
13
14
15
16
17
def require_extractor(extractor: Any, *, staged: bool) -> None:
    """Admit only an exact bound service on the exact staged PG factory path."""
    if (not staged or type(extractor) is not MemoryExtractionService
            or extractor._model is None):
        raise ConfigurationError("receipt extraction requires bound staged PostgreSQL service")