authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-04-02 16:26:18-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-04-02 16:26:18-07:00
log9013c07f6d679813777462466e9051e4ffc27e03
tree003b4c6a59750f8e7e99bdef551fe70b383daa26
parent085340588ac87d14e8cad7c5a8bd620e0c8826da
signature Signed by PGP key B5690EEEBB952194

Clean up terminal state when threads are archived (#1702)

7 files changed, 144 insertions(+), 8 deletions(-)

apps/server/src/server.test.ts+35
...@@ -1193,6 +1193,41 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -1193,6 +1193,41 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
1193 }).pipe(Effect.provide(NodeHttpServer.layerTest)),1193 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
1194 );1194 );
11951195
1196 it.effect("closes thread terminals after a successful archive command", () =>
1197 Effect.gen(function* () {
1198 const threadId = ThreadId.makeUnsafe("thread-archive");
1199 const closeInputs: Array<Parameters<TerminalManagerShape["close"]>[0]> = [];
1200
1201 yield* buildAppUnderTest({
1202 layers: {
1203 terminalManager: {
1204 close: (input) =>
1205 Effect.sync(() => {
1206 closeInputs.push(input);
1207 }),
1208 },
1209 orchestrationEngine: {
1210 dispatch: () => Effect.succeed({ sequence: 8 }),
1211 },
1212 },
1213 });
1214
1215 const wsUrl = yield* getWsServerUrl("/ws");
1216 const dispatchResult = yield* Effect.scoped(
1217 withWsRpcClient(wsUrl, (client) =>
1218 client[ORCHESTRATION_WS_METHODS.dispatchCommand]({
1219 type: "thread.archive",
1220 commandId: CommandId.makeUnsafe("cmd-thread-archive"),
1221 threadId,
1222 }),
1223 ),
1224 );
1225
1226 assert.equal(dispatchResult.sequence, 8);
1227 assert.deepEqual(closeInputs, [{ threadId }]);
1228 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
1229 );
1230
1196 it.effect(1231 it.effect(
1197 "routes websocket rpc subscribeOrchestrationDomainEvents with replay/live overlap resilience",1232 "routes websocket rpc subscribeOrchestrationDomainEvents with replay/live overlap resilience",
1198 () =>1233 () =>
apps/server/src/ws.ts+14-1
...@@ -85,7 +85,20 @@ const WsRpcLayer = WsRpcGroup.toLayer(...@@ -85,7 +85,20 @@ const WsRpcLayer = WsRpcGroup.toLayer(
85 [ORCHESTRATION_WS_METHODS.dispatchCommand]: (command) =>85 [ORCHESTRATION_WS_METHODS.dispatchCommand]: (command) =>
86 Effect.gen(function* () {86 Effect.gen(function* () {
87 const normalizedCommand = yield* normalizeDispatchCommand(command);87 const normalizedCommand = yield* normalizeDispatchCommand(command);
88 return yield* startup.enqueueCommand(orchestrationEngine.dispatch(normalizedCommand));88 const result = yield* startup.enqueueCommand(
89 orchestrationEngine.dispatch(normalizedCommand),
90 );
91 if (normalizedCommand.type === "thread.archive") {
92 yield* terminalManager.close({ threadId: normalizedCommand.threadId }).pipe(
93 Effect.catch((error) =>
94 Effect.logWarning("failed to close thread terminals after archive", {
95 threadId: normalizedCommand.threadId,
96 error: error.message,
97 }),
98 ),
99 );
100 }
101 return result;
89 }).pipe(102 }).pipe(
90 Effect.mapError((cause) =>103 Effect.mapError((cause) =>
91 Schema.is(OrchestrationDispatchCommandError)(cause)104 Schema.is(OrchestrationDispatchCommandError)(cause)
apps/web/src/lib/terminalStateCleanup.test.ts+31-5
...@@ -9,8 +9,8 @@ describe("collectActiveTerminalThreadIds", () => {...@@ -9,8 +9,8 @@ describe("collectActiveTerminalThreadIds", () => {
9 it("retains non-deleted server threads", () => {9 it("retains non-deleted server threads", () => {
10 const activeThreadIds = collectActiveTerminalThreadIds({10 const activeThreadIds = collectActiveTerminalThreadIds({
11 snapshotThreads: [11 snapshotThreads: [
12 { id: threadId("server-1"), deletedAt: null },12 { id: threadId("server-1"), deletedAt: null, archivedAt: null },
13 { id: threadId("server-2"), deletedAt: null },13 { id: threadId("server-2"), deletedAt: null, archivedAt: null },
14 ],14 ],
15 draftThreadIds: [],15 draftThreadIds: [],
16 });16 });
...@@ -18,15 +18,41 @@ describe("collectActiveTerminalThreadIds", () => {...@@ -18,15 +18,41 @@ describe("collectActiveTerminalThreadIds", () => {
18 expect(activeThreadIds).toEqual(new Set([threadId("server-1"), threadId("server-2")]));18 expect(activeThreadIds).toEqual(new Set([threadId("server-1"), threadId("server-2")]));
19 });19 });
2020
21 it("ignores deleted server threads and keeps local draft threads", () => {21 it("ignores deleted and archived server threads and keeps local draft threads", () => {
22 const activeThreadIds = collectActiveTerminalThreadIds({22 const activeThreadIds = collectActiveTerminalThreadIds({
23 snapshotThreads: [23 snapshotThreads: [
24 { id: threadId("server-active"), deletedAt: null },24 { id: threadId("server-active"), deletedAt: null, archivedAt: null },
25 { id: threadId("server-deleted"), deletedAt: "2026-03-05T08:00:00.000Z" },25 {
26 id: threadId("server-deleted"),
27 deletedAt: "2026-03-05T08:00:00.000Z",
28 archivedAt: null,
29 },
30 {
31 id: threadId("server-archived"),
32 deletedAt: null,
33 archivedAt: "2026-03-05T09:00:00.000Z",
34 },
26 ],35 ],
27 draftThreadIds: [threadId("local-draft")],36 draftThreadIds: [threadId("local-draft")],
28 });37 });
2938
30 expect(activeThreadIds).toEqual(new Set([threadId("server-active"), threadId("local-draft")]));39 expect(activeThreadIds).toEqual(new Set([threadId("server-active"), threadId("local-draft")]));
31 });40 });
41
42 it("does not keep draft-linked terminal state for archived server threads", () => {
43 const archivedThreadId = threadId("server-archived");
44
45 const activeThreadIds = collectActiveTerminalThreadIds({
46 snapshotThreads: [
47 {
48 id: archivedThreadId,
49 deletedAt: null,
50 archivedAt: "2026-03-05T09:00:00.000Z",
51 },
52 ],
53 draftThreadIds: [archivedThreadId, threadId("local-draft")],
54 });
55
56 expect(activeThreadIds).toEqual(new Set([threadId("local-draft")]));
57 });
32});58});
apps/web/src/lib/terminalStateCleanup.ts+10
...@@ -3,6 +3,7 @@ import type { ThreadId } from "@t3tools/contracts";...@@ -3,6 +3,7 @@ import type { ThreadId } from "@t3tools/contracts";
3interface TerminalRetentionThread {3interface TerminalRetentionThread {
4 id: ThreadId;4 id: ThreadId;
5 deletedAt: string | null;5 deletedAt: string | null;
6 archivedAt: string | null;
6}7}
78
8interface CollectActiveTerminalThreadIdsInput {9interface CollectActiveTerminalThreadIdsInput {
...@@ -14,11 +15,20 @@ export function collectActiveTerminalThreadIds(...@@ -14,11 +15,20 @@ export function collectActiveTerminalThreadIds(
14 input: CollectActiveTerminalThreadIdsInput,15 input: CollectActiveTerminalThreadIdsInput,
15): Set<ThreadId> {16): Set<ThreadId> {
16 const activeThreadIds = new Set<ThreadId>();17 const activeThreadIds = new Set<ThreadId>();
18 const snapshotThreadById = new Map(input.snapshotThreads.map((thread) => [thread.id, thread]));
17 for (const thread of input.snapshotThreads) {19 for (const thread of input.snapshotThreads) {
18 if (thread.deletedAt !== null) continue;20 if (thread.deletedAt !== null) continue;
21 if (thread.archivedAt !== null) continue;
19 activeThreadIds.add(thread.id);22 activeThreadIds.add(thread.id);
20 }23 }
21 for (const draftThreadId of input.draftThreadIds) {24 for (const draftThreadId of input.draftThreadIds) {
25 const snapshotThread = snapshotThreadById.get(draftThreadId);
26 if (
27 snapshotThread &&
28 (snapshotThread.deletedAt !== null || snapshotThread.archivedAt !== null)
29 ) {
30 continue;
31 }
22 activeThreadIds.add(draftThreadId);32 activeThreadIds.add(draftThreadId);
23 }33 }
24 return activeThreadIds;34 return activeThreadIds;
apps/web/src/orchestrationEventEffects.test.ts+27-1
...@@ -42,6 +42,7 @@ describe("deriveOrchestrationBatchEffects", () => {...@@ -42,6 +42,7 @@ describe("deriveOrchestrationBatchEffects", () => {
42 it("targets draft promotion and terminal cleanup from thread lifecycle events", () => {42 it("targets draft promotion and terminal cleanup from thread lifecycle events", () => {
43 const createdThreadId = ThreadId.makeUnsafe("thread-created");43 const createdThreadId = ThreadId.makeUnsafe("thread-created");
44 const deletedThreadId = ThreadId.makeUnsafe("thread-deleted");44 const deletedThreadId = ThreadId.makeUnsafe("thread-deleted");
45 const archivedThreadId = ThreadId.makeUnsafe("thread-archived");
4546
46 const effects = deriveOrchestrationBatchEffects([47 const effects = deriveOrchestrationBatchEffects([
47 makeEvent("thread.created", {48 makeEvent("thread.created", {
...@@ -60,11 +61,16 @@ describe("deriveOrchestrationBatchEffects", () => {...@@ -60,11 +61,16 @@ describe("deriveOrchestrationBatchEffects", () => {
60 threadId: deletedThreadId,61 threadId: deletedThreadId,
61 deletedAt: "2026-02-27T00:00:01.000Z",62 deletedAt: "2026-02-27T00:00:01.000Z",
62 }),63 }),
64 makeEvent("thread.archived", {
65 threadId: archivedThreadId,
66 archivedAt: "2026-02-27T00:00:02.000Z",
67 updatedAt: "2026-02-27T00:00:02.000Z",
68 }),
63 ]);69 ]);
6470
65 expect(effects.clearPromotedDraftThreadIds).toEqual([createdThreadId]);71 expect(effects.clearPromotedDraftThreadIds).toEqual([createdThreadId]);
66 expect(effects.clearDeletedThreadIds).toEqual([deletedThreadId]);72 expect(effects.clearDeletedThreadIds).toEqual([deletedThreadId]);
67 expect(effects.removeTerminalStateThreadIds).toEqual([deletedThreadId]);73 expect(effects.removeTerminalStateThreadIds).toEqual([deletedThreadId, archivedThreadId]);
68 expect(effects.needsProviderInvalidation).toBe(false);74 expect(effects.needsProviderInvalidation).toBe(false);
69 });75 });
7076
...@@ -105,4 +111,24 @@ describe("deriveOrchestrationBatchEffects", () => {...@@ -105,4 +111,24 @@ describe("deriveOrchestrationBatchEffects", () => {
105 expect(effects.removeTerminalStateThreadIds).toEqual([]);111 expect(effects.removeTerminalStateThreadIds).toEqual([]);
106 expect(effects.needsProviderInvalidation).toBe(true);112 expect(effects.needsProviderInvalidation).toBe(true);
107 });113 });
114
115 it("does not retain archive cleanup when a thread is unarchived later in the same batch", () => {
116 const threadId = ThreadId.makeUnsafe("thread-1");
117
118 const effects = deriveOrchestrationBatchEffects([
119 makeEvent("thread.archived", {
120 threadId,
121 archivedAt: "2026-02-27T00:00:01.000Z",
122 updatedAt: "2026-02-27T00:00:01.000Z",
123 }),
124 makeEvent("thread.unarchived", {
125 threadId,
126 updatedAt: "2026-02-27T00:00:02.000Z",
127 }),
128 ]);
129
130 expect(effects.clearPromotedDraftThreadIds).toEqual([]);
131 expect(effects.clearDeletedThreadIds).toEqual([]);
132 expect(effects.removeTerminalStateThreadIds).toEqual([]);
133 });
108});134});
apps/web/src/orchestrationEventEffects.ts+18
...@@ -46,6 +46,24 @@ export function deriveOrchestrationBatchEffects(...@@ -46,6 +46,24 @@ export function deriveOrchestrationBatchEffects(
46 break;46 break;
47 }47 }
4848
49 case "thread.archived": {
50 threadLifecycleEffects.set(event.payload.threadId, {
51 clearPromotedDraft: false,
52 clearDeletedThread: false,
53 removeTerminalState: true,
54 });
55 break;
56 }
57
58 case "thread.unarchived": {
59 threadLifecycleEffects.set(event.payload.threadId, {
60 clearPromotedDraft: false,
61 clearDeletedThread: false,
62 removeTerminalState: false,
63 });
64 break;
65 }
66
49 default: {67 default: {
50 break;68 break;
51 }69 }
apps/web/src/routes/__root.tsx+9-1
...@@ -328,7 +328,11 @@ function EventRouter() {...@@ -328,7 +328,11 @@ function EventRouter() {
328 useComposerDraftStore.getState().draftThreadsByThreadId,328 useComposerDraftStore.getState().draftThreadsByThreadId,
329 ) as ThreadId[];329 ) as ThreadId[];
330 const activeThreadIds = collectActiveTerminalThreadIds({330 const activeThreadIds = collectActiveTerminalThreadIds({
331 snapshotThreads: threads.map((thread) => ({ id: thread.id, deletedAt: null })),331 snapshotThreads: threads.map((thread) => ({
332 id: thread.id,
333 deletedAt: null,
334 archivedAt: thread.archivedAt,
335 })),
332 draftThreadIds,336 draftThreadIds,
333 });337 });
334 removeOrphanedTerminalStates(activeThreadIds);338 removeOrphanedTerminalStates(activeThreadIds);
...@@ -497,6 +501,10 @@ function EventRouter() {...@@ -497,6 +501,10 @@ function EventRouter() {
497 }501 }
498 });502 });
499 const unsubTerminalEvent = api.terminal.onEvent((event) => {503 const unsubTerminalEvent = api.terminal.onEvent((event) => {
504 const thread = useStore.getState().threads.find((entry) => entry.id === event.threadId);
505 if (thread && thread.archivedAt !== null) {
506 return;
507 }
500 useTerminalStateStore.getState().recordTerminalEvent(event);508 useTerminalStateStore.getState().recordTerminalEvent(event);
501 const hasRunningSubprocess = terminalRunningSubprocessFromEvent(event);509 const hasRunningSubprocess = terminalRunningSubprocessFromEvent(event);
502 if (hasRunningSubprocess === null) {510 if (hasRunningSubprocess === null) {