Skip to content

symfonic.capabilities.memory.binding

binding

Bind a memory capability to one scope and produce its turn records.

as_memory_scope 

as_memory_scope(scope: Any) -> MemoryScope

Refuse paths deeper than this positional memory scope can preserve.

Existing one- to three-level kinds still use the legacy positional key. Their kind-identity migration is tracked separately under CORE-171.

Source code in src/symfonic/capabilities/memory/binding.py 
 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
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def as_memory_scope(scope: Any) -> MemoryScope:
    """Refuse paths deeper than this positional memory scope can preserve.

    Existing one- to three-level kinds still use the legacy positional key.
    Their kind-identity migration is tracked separately under CORE-171.
    """
    if isinstance(scope, MemoryScope):
        return scope
    path_ids: tuple[str, ...] = ()
    levels = getattr(scope, "path", None)
    if levels and not isinstance(levels, (str, bytes)):
        path_levels = tuple(levels)
        if len(path_levels) > 3:
            raise MemoryContractError(
                "scope path has more than three levels and cannot be represented by MemoryScope"
            )
        path_ids = tuple(str(level.id) for level in path_levels if getattr(level, "id", None))
        if len(path_ids) != len(path_levels):
            raise MemoryContractError("scope path contains an empty level")
    segment_ids: tuple[str, ...] = ()
    segments = getattr(scope, "segments", None)
    if segments:
        raw_segments = tuple(segments)
        segment_ids = tuple(str(segment) for segment in raw_segments if segment)
        if len(segment_ids) != len(raw_segments):
            raise MemoryContractError("scope segments contain an empty level")
        if len(segment_ids) > 3:
            raise MemoryContractError(
                "scope path has more than three levels and cannot be represented by MemoryScope"
            )
    if path_ids and segment_ids and path_ids != segment_ids:
        raise MemoryContractError("scope path and segments disagree")
    ids = path_ids or segment_ids
    converted = getattr(scope, "to_memory_scope", None)
    if callable(converted):
        result = converted()
        if isinstance(result, MemoryScope):
            if ids and result.segments != ids:
                raise MemoryContractError(
                    "to_memory_scope lost a scope level before memory composition"
                )
            return result
    if ids:
        return MemoryScope(*ids)
    raise TypeError(
        f"{type(scope).__name__} is neither a MemoryScope nor convertible to "
        "one, so the capability cannot be bound to a scope"
    )

as_native_memory_scope 

as_native_memory_scope(scope: Any) -> MemoryScope

Opt in to complete, kind-aware identity for memory composition.

Source code in src/symfonic/capabilities/memory/binding.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def as_native_memory_scope(scope: Any) -> MemoryScope:
    """Opt in to complete, kind-aware identity for memory composition."""
    if isinstance(scope, MemoryScope):
        if scope.identity_version != 1:
            raise MemoryContractError("native scope requires complete kind/id levels")
        return scope
    path = getattr(scope, "path", None)
    if type(path) is not tuple or not path:
        raise MemoryContractError("native scope requires a nonempty canonical path")
    pairs: list[tuple[str, str]] = []
    for level in path:
        kind = getattr(level, "kind", None)
        id_ = getattr(level, "id", None)
        if type(kind) is not str or type(id_) is not str:
            raise MemoryContractError("native scope path needs complete kind/id levels")
        pairs.append((kind, id_))
    result = MemoryScope.from_levels(tuple(pairs))
    segments = getattr(scope, "segments", None)
    if segments is not None:
        if type(segments) not in (tuple, list):
            raise MemoryContractError("native scope segments must be a complete sequence")
        if tuple(segments) != result.segments:
            raise MemoryContractError("native scope path and segments disagree")
    root_id = getattr(scope, "tenant_id", None)
    if root_id is not None and root_id != result.segments[0]:
        raise MemoryContractError("native scope root and tenant_id disagree")
    child_id = getattr(scope, "sub_tenant_id", None)
    if child_id is not None and (len(result.segments) < 2 or child_id != result.segments[1]):
        raise MemoryContractError("native scope child and sub_tenant_id disagree")
    converted = getattr(scope, "to_memory_scope", None)
    if callable(converted):
        custom = converted()
        if isinstance(custom, MemoryScope):
            if custom.identity_version != 1 or custom.levels != result.levels:
                raise MemoryContractError("to_memory_scope lost native scope kind/id levels")
        elif custom is not scope:
            raise MemoryContractError("to_memory_scope returned a malformed native scope")
    return result

composition_scope 

composition_scope(scope: Any) -> MemoryScope

Keep explicit v0 values; preserve every kind in a supplied path.

Positional conversion remains available for explicitly v0 transports. Public Agent composition must not silently discard a typed owner's kinds.

Source code in src/symfonic/capabilities/memory/binding.py 
14
15
16
17
18
19
20
21
22
23
24
25
def composition_scope(scope: Any) -> MemoryScope:
    """Keep explicit v0 values; preserve every kind in a supplied path.

    Positional conversion remains available for explicitly v0 transports.
    Public Agent composition must not silently discard a typed owner's kinds.
    """
    if isinstance(scope, MemoryScope):
        return scope
    path = getattr(scope, "path", None)
    if path is not None and not isinstance(path, (str, bytes)):
        return as_native_memory_scope(scope)
    return as_memory_scope(scope)

producer_for 

producer_for(scope: MemoryScope) -> Any

Produce the exchange records for the capability's bound scope.

Source code in src/symfonic/capabilities/memory/binding.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def producer_for(scope: MemoryScope) -> Any:
    """Produce the exchange records for the capability's bound scope."""

    def records_from(context: Any) -> tuple[Any, ...]:
        request = getattr(context, "request", None)
        if request is None:  # pragma: no cover - defensive
            return ()
        turn_scope = getattr(request, "scope", None)
        return turn_records(
            query=str(getattr(request, "prompt", "") or ""),
            answer=str(getattr(getattr(context, "turn", None), "text", "") or ""),
            scope=stated_turn_scope(scope, turn_scope),
            # The write stage runs once per model round; deriving the id from
            # the run keeps a tool-using turn from filing one exchange twice.
            turn_id=str(getattr(request, "run_id", "") or "") or None,
            extra_metadata={"session_id": str(getattr(request, "session_id", "") or "")},
            assistant_metadata=(
                completed_tool_calls(getattr(context, "transcript", None))
                if getattr(request, "append_user_message", True) else {}
            ),
        )

    return records_from