Stream Temporal engine turn progress over /invoke instead of going silent until completion - #208
Merged
Merged
Conversation
…ead of going silent until completion TemporalEngine.stream() previously just awaited invoke() (a full accept/poll loop to completion) and yielded exactly one final chunk -- its own doc comment recorded this as a deliberate simplification, not a bug. In practice that meant a streaming chat turn on this engine showed nothing at all until the whole turn finished: no SSE status events, no incremental content, just silence followed by the answer. The engine's own gateway already solves this for its native /v1/chat/completions SSE endpoint: streamTurn polls workflows.TurnProgressQuery (a Temporal query exposing the workflow's own in-flight narration buffer) every 700ms and streams new lines as `status` events. agent-orchestrator never talks to that endpoint though -- it always goes through the generic accept/poll /invoke contract via TemporalEngine.ts, which had no equivalent. Fix: surface the same TurnProgressQuery data over /invoke too. GET /invoke/:id's existing "pending" response now includes a `progress` field (the same narration lines, best-effort -- a query failure or an inactive turn just means an empty list, never an error, since the pending status itself is still accurate). TemporalEngine.poll() tracks how many lines it has already relayed and calls the turn's progressListener for each new one as it arrives, exactly mirroring what the engine's own native endpoint already does -- so a streaming chat caller sees the same live status updates regardless of which engine is driving the turn. Verified: new unit tests cover relaying new-lines-only (no repeats) and confirm nothing is called when the caller has no live channel (no progressListener set). Full engines/temporal and agent-orchestrator suites clean; e2e (bridged-agent-workflow, happy-path, caller-tools) still passes.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Reported live: a chat turn routed through the Temporal engine showed no text streaming and no SSE progress events — nothing until the whole turn completed.
TemporalEngine.stream()was the cause, and it was a documented, deliberate simplification, not a regression: it just awaitedinvoke()(a full accept/poll loop to completion) and yielded exactly one final chunk. Its own comment recorded the tradeoff — "a streaming client on this engine therefore sees the reply rather than the running commentary."The engine's own Go gateway already solves this problem for its native
/v1/chat/completionsSSE endpoint:streamTurnpollsworkflows.TurnProgressQuery(a Temporal query exposing the workflow's own in-flight narration buffer) every 700ms and streams new lines asstatusevents.agent-orchestratornever talks to that endpoint, though — Open WebUI'sopenaiBaseApiUrlpoints atagent-orchestrator, which always routes through the generic accept/poll/invokecontract viaTemporalEngine.ts, and that contract had no equivalent capability.Fix
Surfaces the same
TurnProgressQuerydata over/invoketoo, rather than building a second mechanism:GET /invoke/:id's existing"pending"response now includes aprogressfield — the same narration lines the native endpoint already streams from. Best-effort: a query failure or an inactive turn just means an empty list, never an error, since the pending status itself is still accurate.TemporalEngine.poll()tracks how many lines it has already relayed and calls the turn'sprogressListenerfor each new one as it arrives — the exact callbackagent-orchestrator's own SSE writer (handleChatCompletionsStreaming) already listens to for the in-process LangGraph engine, so no changes were needed on the consuming side at all.Test plan
progressListenerset)go build ./...,go vet ./...,gofmt -l .,go test ./...clean inengines/temporalnpx tsc --noEmit, fullagent-orchestratorvitest suite clean (2 pre-existing, unrelated failures innats-agent-channel*.test.ts— a different subsystem, confirmed unaffected by this change)bridged-agent-workflow,happy-path,caller-tools) still passes against minikube after rebuilding both the gateway and orchestrator images