Skip to content

symfonic.platform.wire

wire

Values on the wire, and the domain values they name.

Split from :mod:symfonic.platform.transport, which is the router itself. The router decides who may ask; this decides what a header, a query parameter or a stored record looks like as JSON. Neither needs to change when the other does, and one of them is the file a deployment reads when it wants to know what the API returns.

kind_of 

kind_of(event: Any) -> str

The event's kind as a plain string, whatever type carries it.

Source code in src/symfonic/platform/wire.py 
52
53
54
55
def kind_of(event: Any) -> str:
    """The event's kind as a plain string, whatever type carries it."""
    kind = getattr(event, "kind", None)
    return getattr(kind, "value", None) or str(kind or "chunk")

memory_layer 

memory_layer(name: str | None) -> Any

The memory layer a query parameter names, or None for every layer.

Source code in src/symfonic/platform/wire.py 
58
59
60
61
62
63
64
65
66
67
def memory_layer(name: str | None) -> Any:
    """The memory layer a query parameter names, or ``None`` for every layer."""
    if name is None:
        return None
    from symfonic.capabilities.memory import MemoryLayer

    try:
        return MemoryLayer(name)
    except ValueError:
        raise ValueError(f"unknown layer: {name!r}") from None

memory_scope 

memory_scope(scope: Any) -> Any

The memory scope a framework scope names.

Delegated rather than converted here: the memory package owns the one translation between the two spellings of a scope, and a second one in a transport is how they drift.

Source code in src/symfonic/platform/wire.py 
70
71
72
73
74
75
76
77
78
79
def memory_scope(scope: Any) -> Any:
    """The memory scope a framework scope names.

    Delegated rather than converted here: the memory package owns the one
    translation between the two spellings of a scope, and a second one in a
    transport is how they drift.
    """
    from symfonic.capabilities.memory import as_memory_scope

    return as_memory_scope(scope)

record_body 

record_body(record: Any) -> dict[str, Any]

One stored record as JSON a browser can render.

Source code in src/symfonic/platform/wire.py 
82
83
84
85
86
87
88
89
90
91
92
93
def record_body(record: Any) -> dict[str, Any]:
    """One stored record as JSON a browser can render."""
    return {
        "id": record.record_id,
        "layer": getattr(record.layer, "value", str(record.layer)),
        "content": record.text,
        "scope_path": record.scope_path,
        "salience": record.salience,
        "origin": record.origin,
        "revision": record.revision,
        "metadata": dict(record.metadata or {}),
    }

scope_for_tenant 

scope_for_tenant(tenant: str) -> Any

The scope a tenant header names.

One level, because a header names a tenant and nothing deeper. A router that read a principal or a session from headers too would be deriving authorisation from something the caller controls, which is the platform's own rule about where a scope may come from.

Source code in src/symfonic/platform/wire.py 
24
25
26
27
28
29
30
31
32
33
34
35
def scope_for_tenant(tenant: str) -> Any:
    """The scope a tenant header names.

    One level, because a header names a tenant and nothing deeper. A router
    that read a principal or a session from headers too would be deriving
    authorisation from something the caller controls, which is the platform's
    own rule about where a scope may come from.
    """
    from symfonic.agent.types import FrameworkTenantScope
    from symfonic.core.scope import ScopeLevel

    return FrameworkTenantScope.from_path([ScopeLevel(kind="tenant", id=tenant)])

text_of 

text_of(value: Any) -> str

Whatever the turn produced, as text a JSON body can carry.

Structural rather than typed: the router is a mapper and must not learn the result type, which differs between a run and a stream chunk and is free to gain fields without this file changing.

Source code in src/symfonic/platform/wire.py 
38
39
40
41
42
43
44
45
46
47
48
49
def text_of(value: Any) -> str:
    """Whatever the turn produced, as text a JSON body can carry.

    Structural rather than typed: the router is a mapper and must not learn
    the result type, which differs between a run and a stream chunk and is
    free to gain fields without this file changing.
    """
    for attribute in ("content", "text", "final_response", "delta"):
        found = getattr(value, attribute, None)
        if isinstance(found, str) and found:
            return found
    return str(value)