| 107 | }, |
| 108 | |
| 109 | async *streamEvents(input, signal): AsyncIterable<RunEvent> { |
| 110 | const graph = compile(input.selectedAnalysts, input.outputLanguage); |
| 111 | const accumulated = makeState(input); |
| 112 | try { |
| 113 | const stream = await graph.stream(accumulated, { |
| 114 | recursionLimit: config.maxRecurLimit, |
| 115 | streamMode: "updates", |
| 116 | signal, |
| 117 | }); |
| 118 | for await (const chunk of stream) { |
| 119 | // updates mode: { [nodeName]: Partial<AgentState> } |
| 120 | for (const [node, patch] of Object.entries( |
| 121 | chunk as Record<string, Record<string, unknown>>, |
| 122 | )) { |
| 123 | if (patch && typeof patch === "object") { |
| 124 | // `structured` is a merge channel (see workflow/state.ts): a shallow |
| 125 | // Object.assign would let each node clobber prior contributions, so |
| 126 | // accumulate it explicitly to mirror the graph's reducer. |
| 127 | const { structured, ...rest } = patch as { |
| 128 | structured?: Record<string, unknown>; |
| 129 | } & Record<string, unknown>; |
| 130 | Object.assign(accumulated, rest); |
| 131 | if (structured) { |
| 132 | accumulated.structured = { ...accumulated.structured, ...structured }; |
| 133 | } |
| 134 | yield { type: "nodeEnd", node, patch }; |
| 135 | } |
| 136 | } |
| 137 | } |
| 138 | yield { |
| 139 | type: "done", |
| 140 | rating: processSignal(accumulated.finalTradeDecision), |
| 141 | finalState: accumulated, |
| 142 | }; |
| 143 | } catch (e) { |
| 144 | yield { type: "error", message: e instanceof Error ? e.message : String(e) }; |
| 145 | } |
| 146 | }, |
| 147 | }; |
| 148 | } |