Stream Temporal engine turn progress over /invoke instead of going silent until completion

apps/agent-orchestrator/src/engine/temporal-engine.test.ts

@@ -131,8 +131,12 @@ describe("TemporalEngine", () => {
     await expect(engine.invoke(input())).rejects.toThrow(/502/);
   });
 
-  // Streaming yields the answer but no per-node narration: those lines describe
-  // LangGraph node transitions, which do not exist on this engine.
+  // The returned AsyncIterable itself still only ever yields one terminal
+  // update -- there are no LangGraph node transitions on this engine to
+  // report as separate updates. Live narration during the wait rides a
+  // different channel: progressListener, called as a side effect from
+  // poll() below, which is what a streaming caller's SSE writer actually
+  // listens to (see handleChatCompletionsStreaming).
   it("streams a single terminal update", async () => {
     const { impl } = scriptedFetch([{ id: "x", status: "succeeded", result: "done" }]);
     const engine = new TemporalEngine({ baseUrl: BASE, fetchImpl: impl });
@@ -144,4 +148,38 @@ describe("TemporalEngine", () => {
     expect(updates).toHaveLength(1);
     expect(Object.values(updates[0]!)[0]).toMatchObject({ result: "done" });
   });
+
+  // Without this, a streaming chat caller on this engine saw nothing at all
+  // until the whole turn completed -- poll() ran silently, even though the
+  // engine's own gateway already narrates in-flight turns
+  // (workflows.TurnProgressQuery) for its native SSE endpoint. This is that
+  // same narration relayed through the accept/poll /invoke contract instead.
+  it("relays in-flight progress lines to progressListener as they arrive, without repeating them", async () => {
+    const { impl } = scriptedFetch([
+      { id: "x", status: "pending", progress: ["clone: cloning the repo"] },
+      { id: "x", status: "pending", progress: ["clone: cloning the repo", "edit: adding the README"] },
+      { id: "x", status: "succeeded", result: "Opened PR #1" },
+    ]);
+    const progressListener = vi.fn();
+    const engine = new TemporalEngine({ baseUrl: BASE, fetchImpl: impl });
+
+    const state = await engine.invoke(input({ progressListener }));
+
+    expect(state.result).toBe("Opened PR #1");
+    expect(progressListener).toHaveBeenCalledTimes(2);
+    expect(progressListener).toHaveBeenNthCalledWith(1, "", "clone: cloning the repo");
+    expect(progressListener).toHaveBeenNthCalledWith(2, "", "edit: adding the README");
+  });
+
+  it("never calls progressListener when the caller has no live channel", async () => {
+    const { impl } = scriptedFetch([
+      { id: "x", status: "pending", progress: ["clone: cloning the repo"] },
+      { id: "x", status: "succeeded", result: "Opened PR #1" },
+    ]);
+    const engine = new TemporalEngine({ baseUrl: BASE, fetchImpl: impl });
+
+    // input() sets no progressListener -- this must not throw reading it,
+    // and there is nothing to assert a call against.
+    await expect(engine.invoke(input())).resolves.toMatchObject({ result: "Opened PR #1" });
+  });
 });

apps/agent-orchestrator/src/engine/temporal-engine.ts

@@ -87,6 +87,8 @@ interface InvokeRecord {
   result?: string;
   error?: string;
   toolCalls?: { id: string; name: string; arguments: string }[];
+  /** In-flight narration lines (only ever set on a "pending" record) -- see poll()'s use of it. */
+  progress?: string[];
 }
 
 export class TemporalEngine implements AgentGraphLike {
@@ -229,6 +231,12 @@ export class TemporalEngine implements AgentGraphLike {
 
   private async poll(id: string, input: AgentGraphInput): Promise<InvokeRecord> {
     const deadline = Date.now() + (this.options.timeoutMs ?? DEFAULT_TIMEOUT_MS);
+    // Lines already relayed to progressListener -- the engine's /invoke/:id
+    // resends the WHOLE narration buffer each poll (same source the
+    // gateway's own native SSE endpoint streams from, workflows.
+    // TurnProgressQuery), not just what's new, so this is what keeps a
+    // streaming caller from seeing every line repeated on every poll tick.
+    let seen = 0;
     for (;;) {
       const res = await this.fetchImpl(`${this.baseUrl}/invoke/${encodeURIComponent(id)}`, {
         headers: this.headers(),
@@ -239,6 +247,17 @@ export class TemporalEngine implements AgentGraphLike {
       const record = (await res.json()) as InvokeRecord;
       if (record.status !== "pending") return record;
 
+      // Surfaces as real SSE status events the moment they arrive, exactly
+      // like the engine's own native streaming endpoint -- without this, a
+      // streaming chat caller on this engine sees nothing until the whole
+      // turn completes (previously this whole poll loop ran silently).
+      if (input.progressListener && record.progress) {
+        for (const line of record.progress.slice(seen)) {
+          input.progressListener("", line);
+        }
+        seen = Math.max(seen, record.progress.length);
+      }
+
       if (Date.now() >= deadline) {
         // Deliberately not an engine error: the turn is still running and its
         // answer stays collectable, because the record IS the workflow rather
@@ -253,7 +272,6 @@ export class TemporalEngine implements AgentGraphLike {
         };
       }
       await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS));
-      void input;
     }
   }
 

engines/temporal/internal/gateway/invoke.go

@@ -97,6 +97,13 @@ type invokeRecord struct {
 	// array, so a caller has nowhere to put the result. Reported so an adapter
 	// sees a real outcome rather than an empty success.
 	ToolCalls []callertools.PendingCall `json:"toolCalls,omitempty"`
+	// Progress carries the in-flight turn's narration lines (workflows.
+	// TurnProgressQuery), same source streamTurn already polls for the
+	// gateway's own native SSE endpoint -- surfaced here too so a caller
+	// using the accept/poll /invoke contract (agent-orchestrator's
+	// TemporalEngine) can render live status instead of going silent until
+	// the turn completes. Only ever set on a "pending" record.
+	Progress []string `json:"progress,omitempty"`
 }
 
 const (
@@ -289,7 +296,17 @@ func (s *Server) handleInvokeStatus(c *gin.Context) {
 
 	case ctx.Err() != nil && c.Request.Context().Err() == nil:
 		// Our own deadline, not the client's: the turn is simply still running.
-		c.JSON(http.StatusOK, invokeRecord{ID: id, Status: invokeStatusPending})
+		// Best-effort narration alongside it -- a query failure or an inactive
+		// turn (nothing narrated yet) just means an empty Progress, never an
+		// error response, since the pending status itself is still accurate.
+		record := invokeRecord{ID: id, Status: invokeStatusPending}
+		if resp, err := s.temporal.QueryWorkflow(c.Request.Context(), workflowID, "", workflows.TurnProgressQuery); err == nil {
+			var progress workflows.TurnProgress
+			if resp.Get(&progress) == nil && progress.Active {
+				record.Progress = progress.Lines
+			}
+		}
+		c.JSON(http.StatusOK, record)
 
 	case isUnknownUpdate(err):
 		writeError(c, http.StatusNotFound, "unknown invocation")