Skip to content

symfonic.capabilities.memory.reflection_phase

reflection_phase

The native post-consolidation reflection finalizer.

reflection_handler 

reflection_handler(capability: Any) -> Callable[[Any], Any]

Persist insights after ordinary memory publication, never before egress.

A reflection outage leaves the completed response intact and records a payload-free stage reason. This is a post-consolidation memory action, not a pre-egress metacognitive verdict.

Source code in src/symfonic/capabilities/memory/reflection_phase.py 
15
16
17
18
19
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
48
49
50
51
52
53
54
55
56
57
58
59
60
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
def reflection_handler(capability: Any) -> Callable[[Any], Any]:
    """Persist insights after ordinary memory publication, never before egress.

    A reflection outage leaves the completed response intact and records a
    payload-free stage reason.  This is a post-consolidation memory action, not
    a pre-egress metacognitive verdict.
    """

    async def handle(context: Any) -> StageResult[Any]:
        request = getattr(context, "request", None)
        turn = getattr(context, "turn", None)
        if request is None or turn is None:
            return no_change("reflection received no completed turn")
        if getattr(turn, "tool_requests", ()):
            return no_change("reflection waits for the final model round")
        scope = getattr(request, "scope", None) or capability.scope
        try:
            result = await capability.reflection.reflect(
                ReflectionRequest(
                    scope=scope,
                    user_message=request.prompt,
                    assistant_message=turn.text,
                )
            )
        except Exception as exc:  # noqa: BLE001 - reflection cannot revoke a response
            return no_change(
                f"reflection degraded: {type(exc).__name__}",
                counts={"extracted": 0, "published": 0},
            )
        if getattr(result, "degraded", False):
            return no_change(
                "reflection degraded: collaborator unavailable",
                counts={"extracted": 0, "published": 0},
            )
        records = tuple(getattr(result, "records", ()) or ())
        if not records:
            return no_change(
                getattr(result, "reason", "") or "reflection extracted no durable insights",
                counts={
                    "extracted": 0,
                    "published": 0,
                    "discarded": len(getattr(result, "dropped", ()) or ()),
                },
            )
        try:
            receipt = await capability.writer.write(WriteRequest(scope=scope, records=records))
            if getattr(receipt, "degraded", False):
                return no_change(
                    "reflection degraded: memory store unavailable",
                    counts={"extracted": len(records), "published": 0},
                )
            published = await capability.lifecycle.flush(scope)
        except Exception as exc:  # noqa: BLE001 - same side-effect boundary
            return no_change(
                f"reflection degraded: {type(exc).__name__}",
                counts={"extracted": len(records), "published": 0},
            )
        if getattr(published, "degraded", False):
            return no_change(
                "reflection degraded: memory store unavailable",
                counts={"extracted": len(records), "published": 0},
            )
        committed = tuple(getattr(published, "committed", ()) or ())
        return applied(
            ResolvedInput(capability=HMS_CAPABILITY, value=committed, provenance=scope.path),
            counts={
                "extracted": len(records),
                "published": len(committed),
                "discarded": len(getattr(result, "dropped", ()) or ()),
            },
        )

    return handle