Skip to content

symfonic.platform.transport_continuation

transport_continuation

Typed HTTP projection of the host continuation service.

The route deliberately asks for a ticket before it asks the host to resume: ticket recovery authenticates the opaque token and reveals both its bound scope and its registered response schema, but it does not redeem the ticket. Thus a malformed body cannot spend a one-use pause.

mount_continuation_route 

mount_continuation_route(router: Any, *, host: Any, resolve: Any) -> None

Mount POST /resume/{token} over a host continuation service.

resolve authenticates the HTTP request. It is intentionally supplied by the main router so all transport doors share the same tenant policy.

Source code in src/symfonic/platform/transport_continuation.py 
 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
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
141
142
143
144
145
146
def mount_continuation_route(router: Any, *, host: Any, resolve: Any) -> None:
    """Mount ``POST /resume/{token}`` over a host continuation service.

    ``resolve`` authenticates the HTTP request. It is intentionally supplied
    by the main router so all transport doors share the same tenant policy.
    """
    from fastapi import Header, Request
    from fastapi.responses import StreamingResponse

    @router.post("/resume/{pause_token}")  # type: ignore[untyped-decorator]
    async def resume(
        pause_token: str,
        raw_request: Request,
        x_tenant_id: str | None = Header(default=None),  # noqa: B008
        authorization: str | None = Header(default=None),  # noqa: B008
    ) -> Any:
        request_scope = await resolve(x_tenant_id, authorization, request=raw_request)
        try:
            ticket = await host.ticket(pause_token)
        except Exception as exc:  # ticket lookup has no redemption side effect
            status, detail = _status(exc)

            return StreamingResponse(_single_error(status, detail), media_type="text/event-stream")

        cross_scope_allowed = bool(getattr(ticket, "cross_scope_allowed", False))
        if not _same_scope(request_scope, ticket.scope) and not cross_scope_allowed:
            return StreamingResponse(
                _single_error(403, "continuation belongs to another authenticated scope"),
                media_type="text/event-stream",
            )

        try:
            raw_body = await raw_request.json()
        except Exception as exc:  # malformed JSON is still an SSE protocol error
            detail = f"Invalid JSON body: {exc}"

            return StreamingResponse(_single_error(400, detail), media_type="text/event-stream")

        schema = getattr(ticket, "response_schema", None)
        validate = getattr(schema, "model_validate", None)
        if not callable(validate):
            return StreamingResponse(
                _single_error(400, "continuation has no registered response schema"),
                media_type="text/event-stream",
            )
        try:
            answer = validate(raw_body)
        except Exception as exc:  # validation happens before host.resume/redeem
            detail = f"Invalid resume payload: {exc}"

            return StreamingResponse(_single_error(400, detail), media_type="text/event-stream")

        async def events() -> AsyncIterator[str]:
            try:
                resume_stream = getattr(host, "resume_stream", None)
                if callable(resume_stream):
                    source = resume_stream(pause_token, answer, scope=request_scope)
                    async with aclosing(source):
                        async for event in source:
                            yield _typed(event)
                else:
                    result = await host.resume(pause_token, answer, scope=request_scope)
                    if hasattr(result, "__aiter__"):
                        async with aclosing(result):
                            async for event in result:
                                yield _typed(event)
                    else:
                        yield _typed(result)
            except Exception as exc:  # resume failures occur after the SSE handshake
                status, detail = _status(exc)
                yield _error(status, detail)

        return StreamingResponse(events(), media_type="text/event-stream")