authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-08-24 17:50:02-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-08-25 00:50:02+00:00
logc034f51bb727de3d4888c74497b13e878ee65187
tree547c3599c0f4f1b09ec330006f004b38d706c9e3
parent99960383d094b850f0001d3067fa604aa3adf7d6
signature Signed by PGP key B5690EEEBB952194

fix(server): stop routine events from rescanning thread history (#8150)

2 files changed, 98 insertions(+), 1 deletions(-)

apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+72
...@@ -174,6 +174,78 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {...@@ -174,6 +174,78 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
174 assert.equal(row.lastAppliedSequence, 3);174 assert.equal(row.lastAppliedSequence, 3);
175 }175 }
176176
177 yield* sql`CREATE TABLE thread_shell_updates (count INTEGER NOT NULL)`;
178 yield* sql`INSERT INTO thread_shell_updates (count) VALUES (0)`;
179 yield* sql`
180 CREATE TRIGGER count_thread_shell_updates
181 AFTER UPDATE ON projection_threads
182 WHEN NEW.thread_id = 'thread-1'
183 BEGIN
184 UPDATE thread_shell_updates SET count = count + 1;
185 END;
186 `;
187
188 yield* eventStore.append({
189 type: "thread.message-sent",
190 eventId: EventId.make("evt-assistant-update"),
191 aggregateKind: "thread",
192 aggregateId: ThreadId.make("thread-1"),
193 occurredAt: "2026-01-01T00:00:00.100Z",
194 commandId: CommandId.make("cmd-assistant-update"),
195 causationEventId: null,
196 correlationId: CommandId.make("cmd-assistant-update"),
197 metadata: {},
198 payload: {
199 threadId: ThreadId.make("thread-1"),
200 messageId: MessageId.make("message-2"),
201 role: "assistant",
202 text: "more work",
203 turnId: null,
204 streaming: false,
205 createdAt: "2026-01-01T00:00:00.100Z",
206 updatedAt: "2026-01-01T00:00:00.100Z",
207 },
208 });
209 yield* projectionPipeline.bootstrap;
210
211 let threadShellUpdates = yield* sql<{ readonly count: number }>`
212 SELECT count FROM thread_shell_updates
213 `;
214 assert.deepEqual(threadShellUpdates, [{ count: 1 }]);
215
216 yield* sql`UPDATE thread_shell_updates SET count = 0`;
217 yield* eventStore.append({
218 type: "thread.activity-appended",
219 eventId: EventId.make("evt-routine-activity"),
220 aggregateKind: "thread",
221 aggregateId: ThreadId.make("thread-1"),
222 occurredAt: "2026-01-01T00:00:00.200Z",
223 commandId: CommandId.make("cmd-routine-activity"),
224 causationEventId: null,
225 correlationId: CommandId.make("cmd-routine-activity"),
226 metadata: {},
227 payload: {
228 threadId: ThreadId.make("thread-1"),
229 activity: {
230 id: EventId.make("activity-routine"),
231 tone: "tool",
232 kind: "tool.updated",
233 summary: "Tool made progress",
234 payload: {},
235 turnId: null,
236 createdAt: "2026-01-01T00:00:00.200Z",
237 },
238 },
239 });
240 yield* projectionPipeline.bootstrap;
241
242 threadShellUpdates = yield* sql<{ readonly count: number }>`
243 SELECT count FROM thread_shell_updates
244 `;
245 assert.deepEqual(threadShellUpdates, [{ count: 1 }]);
246 yield* sql`DROP TRIGGER count_thread_shell_updates`;
247 yield* sql`DROP TABLE thread_shell_updates`;
248
177 // Settled lifecycle through the DB pipeline: thread.settled writes the249 // Settled lifecycle through the DB pipeline: thread.settled writes the
178 // override + timestamp, thread.unsettled(user) flips to the active pin.250 // override + timestamp, thread.unsettled(user) flips to the active pin.
179 yield* eventStore.append({251 yield* eventStore.append({
apps/server/src/orchestration/Layers/ProjectionPipeline.ts+26-1
...@@ -130,6 +130,29 @@ function isStalePendingApprovalFailureDetail(detail: string | null): boolean {...@@ -130,6 +130,29 @@ function isStalePendingApprovalFailureDetail(detail: string | null): boolean {
130 );130 );
131}131}
132132
133// A full refresh loads all thread history, so skip events that cannot change the summary.
134function shouldRefreshThreadShellSummary(event: OrchestrationEvent): boolean {
135 if (event.type === "thread.message-sent") {
136 return event.payload.role === "user";
137 }
138
139 if (event.type !== "thread.activity-appended") {
140 return true;
141 }
142
143 switch (event.payload.activity.kind) {
144 case "approval.requested":
145 case "approval.resolved":
146 case "provider.approval.respond.failed":
147 case "user-input.requested":
148 case "user-input.resolved":
149 case "provider.user-input.respond.failed":
150 return true;
151 default:
152 return false;
153 }
154}
155
133function derivePendingUserInputCountFromActivities(156function derivePendingUserInputCountFromActivities(
134 activities: ReadonlyArray<ProjectionThreadActivity>,157 activities: ReadonlyArray<ProjectionThreadActivity>,
135): number {158): number {
...@@ -865,7 +888,9 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti...@@ -865,7 +888,9 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
865 ...existingRow.value,888 ...existingRow.value,
866 updatedAt: event.occurredAt,889 updatedAt: event.occurredAt,
867 });890 });
868 yield* refreshThreadShellSummary(event.payload.threadId);891 if (shouldRefreshThreadShellSummary(event)) {
892 yield* refreshThreadShellSummary(event.payload.threadId);
893 }
869 return;894 return;
870 }895 }
871896