authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-09-14 22:32:43-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-14 22:32:43-07:00
log2c19283afed1e3b69a9af33360980527aeb37494
tree75cccd7aa1a7466058f8e471aa41440f0e16c650
parent26b8f985d85773e41a2ee637785f7c87d5c1a01e
signature Signed by PGP key B5690EEEBB952194

feat(server): persist the worktree setup send and progress on the thread (#11852)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

29 files changed, 1200 insertions(+), 207 deletions(-)

apps/mobile/src/features/threads/ThreadRouteScreen.tsx+18-1
...@@ -782,9 +782,26 @@ function ThreadRouteContent(...@@ -782,9 +782,26 @@ function ThreadRouteContent(
782 }),782 }),
783 );783 );
784 }, [navigation, routeThreadIdentity, selectedThreadCreation, selectedThreadProject]);784 }, [navigation, routeThreadIdentity, selectedThreadCreation, selectedThreadProject]);
785 // A worktree bootstrap records a running setup on the thread before its
786 // turn, so a thread opened from another device (or after a restart) shows
787 // the same preparing state the sending client does. A starting session is
788 // not enough on its own: an ordinary first turn projects one too.
789 const awaitingBootstrapTurn = useMemo(
790 () =>
791 selectedThreadDetail !== null &&
792 selectedThreadDetail.latestTurn === null &&
793 selectedThreadDetail.activities.some(
794 (activity) =>
795 activity.kind === "worktree-setup" &&
796 typeof activity.payload === "object" &&
797 activity.payload !== null &&
798 (activity.payload as { phase?: unknown }).phase === "running",
799 ),
800 [selectedThreadDetail],
801 );
785 const creationState = ((): ThreadDetailScreenProps["creationState"] => {802 const creationState = ((): ThreadDetailScreenProps["creationState"] => {
786 if (selectedThreadCreation === null) {803 if (selectedThreadCreation === null) {
787 return null;804 return awaitingBootstrapTurn ? { kind: "preparing", preparingWorktree: true } : null;
788 }805 }
789 if (selectedThreadCreation.outcome?.kind === "failed") {806 if (selectedThreadCreation.outcome?.kind === "failed") {
790 return {807 return {
apps/mobile/src/lib/threadActivity.ts+1
...@@ -411,6 +411,7 @@ function deriveWorkLogEntries(...@@ -411,6 +411,7 @@ function deriveWorkLogEntries(
411 const ordered = Arr.sort(activities, activityOrder);411 const ordered = Arr.sort(activities, activityOrder);
412 const entries: DerivedWorkLogEntry[] = [];412 const entries: DerivedWorkLogEntry[] = [];
413 for (const activity of foldUserInputActivities(ordered)) {413 for (const activity of foldUserInputActivities(ordered)) {
414 // Mobile has no setup card, so a failed setup surfaces as an error row.
414 if (activity.tone !== "error" && isWorktreeSetupActivity(activity.kind)) continue;415 if (activity.tone !== "error" && isWorktreeSetupActivity(activity.kind)) continue;
415 if (activity.kind === "tool.started") continue;416 if (activity.kind === "tool.started") continue;
416 // Like web: an agent's task.started row anchors its batch. It has a fixed417 // Like web: an agent's task.started row anchors its batch. It has a fixed
apps/server/src/checkpointing/CheckpointDiffQuery.test.ts+5
...@@ -76,6 +76,7 @@ describe("CheckpointDiffQuery.layer", () => {...@@ -76,6 +76,7 @@ describe("CheckpointDiffQuery.layer", () => {
76 Layer.provideMerge(76 Layer.provideMerge(
77 Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {77 Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
78 getUserInputActivity: () => Effect.die("unused"),78 getUserInputActivity: () => Effect.die("unused"),
79 listActivitiesByKind: () => Effect.die("unused"),
79 getCommandReadModel: () =>80 getCommandReadModel: () =>
80 Effect.die("CheckpointDiffQuery should not request the command read model"),81 Effect.die("CheckpointDiffQuery should not request the command read model"),
81 getSnapshot: () =>82 getSnapshot: () =>
...@@ -191,6 +192,7 @@ describe("CheckpointDiffQuery.layer", () => {...@@ -191,6 +192,7 @@ describe("CheckpointDiffQuery.layer", () => {
191 Layer.provideMerge(192 Layer.provideMerge(
192 Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {193 Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
193 getUserInputActivity: () => Effect.die("unused"),194 getUserInputActivity: () => Effect.die("unused"),
195 listActivitiesByKind: () => Effect.die("unused"),
194 getCommandReadModel: () =>196 getCommandReadModel: () =>
195 Effect.die("CheckpointDiffQuery should not request the command read model"),197 Effect.die("CheckpointDiffQuery should not request the command read model"),
196 getSnapshot: () =>198 getSnapshot: () =>
...@@ -281,6 +283,7 @@ describe("CheckpointDiffQuery.layer", () => {...@@ -281,6 +283,7 @@ describe("CheckpointDiffQuery.layer", () => {
281 Layer.provideMerge(283 Layer.provideMerge(
282 Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {284 Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
283 getUserInputActivity: () => Effect.die("unused"),285 getUserInputActivity: () => Effect.die("unused"),
286 listActivitiesByKind: () => Effect.die("unused"),
284 getCommandReadModel: () =>287 getCommandReadModel: () =>
285 Effect.die("CheckpointDiffQuery should not request the command read model"),288 Effect.die("CheckpointDiffQuery should not request the command read model"),
286 getSnapshot: () =>289 getSnapshot: () =>
...@@ -356,6 +359,7 @@ describe("CheckpointDiffQuery.layer", () => {...@@ -356,6 +359,7 @@ describe("CheckpointDiffQuery.layer", () => {
356 Layer.provideMerge(359 Layer.provideMerge(
357 Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {360 Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
358 getUserInputActivity: () => Effect.die("unused"),361 getUserInputActivity: () => Effect.die("unused"),
362 listActivitiesByKind: () => Effect.die("unused"),
359 getCommandReadModel: () =>363 getCommandReadModel: () =>
360 Effect.die("CheckpointDiffQuery should not request the command read model"),364 Effect.die("CheckpointDiffQuery should not request the command read model"),
361 getSnapshot: () =>365 getSnapshot: () =>
...@@ -416,6 +420,7 @@ describe("CheckpointDiffQuery.layer", () => {...@@ -416,6 +420,7 @@ describe("CheckpointDiffQuery.layer", () => {
416 Layer.provideMerge(420 Layer.provideMerge(
417 Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {421 Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
418 getUserInputActivity: () => Effect.die("unused"),422 getUserInputActivity: () => Effect.die("unused"),
423 listActivitiesByKind: () => Effect.die("unused"),
419 getCommandReadModel: () =>424 getCommandReadModel: () =>
420 Effect.die("CheckpointDiffQuery should not request the command read model"),425 Effect.die("CheckpointDiffQuery should not request the command read model"),
421 getSnapshot: () =>426 getSnapshot: () =>
apps/server/src/orchestration/Layers/CheckpointReactor.ts+4
...@@ -630,8 +630,12 @@ const make = Effect.gen(function* () {...@@ -630,8 +630,12 @@ const make = Effect.gen(function* () {
630 >,630 >,
631 ) {631 ) {
632 if (event.type === "thread.message-sent") {632 if (event.type === "thread.message-sent") {
633 // A bootstrap message lands before the worktree exists; its baseline
634 // would snapshot the project checkout. The turn-start event that
635 // follows captures it against the right cwd.
633 if (636 if (
634 event.metadata.historyImport === true ||637 event.metadata.historyImport === true ||
638 event.metadata.deferredTurn === true ||
635 event.payload.role !== "user" ||639 event.payload.role !== "user" ||
636 event.payload.streaming ||640 event.payload.streaming ||
637 event.payload.turnId !== null641 event.payload.turnId !== null
apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts+1
...@@ -420,6 +420,7 @@ describe("OrchestrationEngine", () => {...@@ -420,6 +420,7 @@ describe("OrchestrationEngine", () => {
420 Layer.provide(420 Layer.provide(
421 Layer.succeed(ProjectionSnapshotQuery, {421 Layer.succeed(ProjectionSnapshotQuery, {
422 getUserInputActivity: () => Effect.die("unused"),422 getUserInputActivity: () => Effect.die("unused"),
423 listActivitiesByKind: () => Effect.die("unused"),
423 getCommandReadModel: () => Effect.succeed(commandReadModel),424 getCommandReadModel: () => Effect.succeed(commandReadModel),
424 getSnapshot: () =>425 getSnapshot: () =>
425 Effect.sync(() => {426 Effect.sync(() => {
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts+48
...@@ -3466,3 +3466,51 @@ it.effect("omits foreign-host PRs from legacy snapshots while preserving native...@@ -3466,3 +3466,51 @@ it.effect("omits foreign-host PRs from legacy snapshots while preserving native
3466 }3466 }
3467 }).pipe(Effect.provide(layer));3467 }).pipe(Effect.provide(layer));
3468});3468});
3469
3470projectionSnapshotLayer("ProjectionSnapshotQuery activities by kind", (it) => {
3471 it.effect("lists one kind across active threads only, without hydrating the threads", () =>
3472 Effect.gen(function* () {
3473 const query = yield* ProjectionSnapshotQuery;
3474 const sql = yield* SqlClient.SqlClient;
3475 const timestamp = "2026-03-02T00:00:00.000Z";
3476 yield* sql`
3477 INSERT INTO projection_projects (
3478 project_id, title, workspace_root, scripts_json, created_at, updated_at
3479 ) VALUES ('project-kinds', 'Project', '/tmp/project-kinds', '[]', ${timestamp}, ${timestamp})
3480 `;
3481 yield* sql`
3482 INSERT INTO projection_threads (
3483 thread_id, project_id, title, model_selection_json, runtime_mode, interaction_mode,
3484 created_at, updated_at, deleted_at
3485 ) VALUES
3486 ('thread-live', 'project-kinds', 'Live', '{"instanceId":"codex","model":"gpt-5"}',
3487 'full-access', 'default', ${timestamp}, ${timestamp}, NULL),
3488 ('thread-gone', 'project-kinds', 'Gone', '{"instanceId":"codex","model":"gpt-5"}',
3489 'full-access', 'default', ${timestamp}, ${timestamp}, ${timestamp}),
3490 ('thread-shelved', 'project-kinds', 'Shelved', '{"instanceId":"codex","model":"gpt-5"}',
3491 'full-access', 'default', ${timestamp}, ${timestamp}, NULL)
3492 `;
3493 yield* sql`UPDATE projection_threads SET archived_at = ${timestamp} WHERE thread_id = 'thread-shelved'`;
3494 yield* sql`
3495 INSERT INTO projection_thread_activities (
3496 activity_id, thread_id, turn_id, tone, kind, summary, payload_json, created_at
3497 ) VALUES
3498 ('setup-live', 'thread-live', NULL, 'info', 'worktree-setup', 'Setting up',
3499 '{"phase":"running"}', ${timestamp}),
3500 ('other-live', 'thread-live', NULL, 'info', 'tool.completed', 'Other',
3501 '{}', ${timestamp}),
3502 ('setup-gone', 'thread-gone', NULL, 'info', 'worktree-setup', 'Setting up',
3503 '{"phase":"running"}', ${timestamp}),
3504 ('setup-shelved', 'thread-shelved', NULL, 'info', 'worktree-setup', 'Setting up',
3505 '{"phase":"running"}', ${timestamp})
3506 `;
3507
3508 const setups = yield* query.listActivitiesByKind("worktree-setup");
3509 assert.deepEqual(
3510 setups.map((activity) => [activity.id, activity.kind, activity.payload]),
3511 [["setup-live", "worktree-setup", { phase: "running" }]],
3512 );
3513 assert.deepEqual(yield* query.listActivitiesByKind("nope"), []);
3514 }),
3515 );
3516});
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts+35
...@@ -1450,6 +1450,40 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {...@@ -1450,6 +1450,40 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
1450 ),1450 ),
1451 );1451 );
14521452
1453 const listActivityRowsByKind = SqlSchema.findAll({
1454 Request: Schema.Struct({ kind: Schema.String }),
1455 Result: ProjectionThreadActivityDbRowSchema,
1456 execute: ({ kind }) => sql`
1457 SELECT
1458 a.activity_id AS "activityId",
1459 a.thread_id AS "threadId",
1460 a.turn_id AS "turnId",
1461 a.tone,
1462 a.kind,
1463 a.summary,
1464 a.payload_json AS "payload",
1465 a.sequence,
1466 a.created_at AS "createdAt"
1467 FROM projection_thread_activities a
1468 JOIN projection_threads t ON t.thread_id = a.thread_id
1469 WHERE a.kind = ${kind}
1470 AND t.deleted_at IS NULL
1471 AND t.archived_at IS NULL
1472 ORDER BY a.created_at ASC, a.activity_id ASC
1473 `,
1474 });
1475
1476 const listActivitiesByKind: ProjectionSnapshotQueryShape["listActivitiesByKind"] = (kind) =>
1477 listActivityRowsByKind({ kind }).pipe(
1478 Effect.map((rows) => rows.map(mapThreadActivityRow)),
1479 Effect.mapError(
1480 toPersistenceSqlOrDecodeError(
1481 "ProjectionSnapshotQuery.listActivitiesByKind:query",
1482 "ProjectionSnapshotQuery.listActivitiesByKind:decodeRow",
1483 ),
1484 ),
1485 );
1486
1453 const listThreadActivityIdsByThread = SqlSchema.findAll({1487 const listThreadActivityIdsByThread = SqlSchema.findAll({
1454 Request: ThreadIdLookupInput,1488 Request: ThreadIdLookupInput,
1455 Result: ProjectionThreadActivityIdRowSchema,1489 Result: ProjectionThreadActivityIdRowSchema,
...@@ -3692,6 +3726,7 @@ pending_approval_requests AS (...@@ -3692,6 +3726,7 @@ pending_approval_requests AS (
3692 return {3726 return {
3693 getCommandReadModel,3727 getCommandReadModel,
3694 getUserInputActivity,3728 getUserInputActivity,
3729 listActivitiesByKind,
3695 getSnapshot,3730 getSnapshot,
3696 getShellSnapshot,3731 getShellSnapshot,
3697 getArchivedShellSnapshot,3732 getArchivedShellSnapshot,
apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts+9
...@@ -83,6 +83,15 @@ export interface ProjectionSnapshotQueryShape {...@@ -83,6 +83,15 @@ export interface ProjectionSnapshotQueryShape {
83 readonly requestId: ApprovalRequestId;83 readonly requestId: ApprovalRequestId;
84 }) => Effect.Effect<Option.Option<OrchestrationThreadActivity>, ProjectionRepositoryError>;84 }) => Effect.Effect<Option.Option<OrchestrationThreadActivity>, ProjectionRepositoryError>;
8585
86 /**
87 * Read every activity of one kind across active (not deleted, not archived)
88 * threads, without hydrating the threads. Used at startup to find state a
89 * crashed process left behind.
90 */
91 readonly listActivitiesByKind: (
92 kind: string,
93 ) => Effect.Effect<ReadonlyArray<OrchestrationThreadActivity>, ProjectionRepositoryError>;
94
86 /**95 /**
87 * Read the lightweight command snapshot used to bootstrap the in-memory96 * Read the lightweight command snapshot used to bootstrap the in-memory
88 * orchestration engine without hydrating message/activity/checkpoint bodies.97 * orchestration engine without hydrating message/activity/checkpoint bodies.
apps/server/src/orchestration/decider.ts+81-23
...@@ -1390,27 +1390,39 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"...@@ -1390,27 +1390,39 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"
1390 detail: `Proposed plan '${sourceProposedPlan?.planId}' belongs to thread '${sourceThread.id}' in a different project.`,1390 detail: `Proposed plan '${sourceProposedPlan?.planId}' belongs to thread '${sourceThread.id}' in a different project.`,
1391 });1391 });
1392 }1392 }
1393 const userMessageEvent: Omit<OrchestrationEvent, "sequence"> = {1393 // A worktree bootstrap persists the message ahead of the turn with
1394 ...(yield* withEventBase({1394 // `thread.message.user.append`; the turn then only references it.
1395 aggregateKind: "thread",1395 const persistedUserMessage = targetThread.messages.find(
1396 aggregateId: command.threadId,1396 (message) =>
1397 occurredAt: command.createdAt,1397 message.id === command.message.messageId &&
1398 commandId: command.commandId,1398 message.role === "user" &&
1399 })),1399 message.turnId === null,
1400 type: "thread.message-sent",1400 );
1401 payload: {1401 const userMessageEvent: Omit<OrchestrationEvent, "sequence"> | null = persistedUserMessage
1402 threadId: command.threadId,1402 ? null
1403 messageId: command.message.messageId,1403 : {
1404 role: "user",1404 ...(yield* withEventBase({
1405 text: command.message.text,1405 aggregateKind: "thread",
1406 attachments: command.message.attachments,1406 aggregateId: command.threadId,
1407 ...(command.message.context !== undefined ? { context: command.message.context } : {}),1407 occurredAt: command.createdAt,
1408 turnId: null,1408 commandId: command.commandId,
1409 streaming: false,1409 })),
1410 createdAt: command.createdAt,1410 type: "thread.message-sent",
1411 updatedAt: command.createdAt,1411 payload: {
1412 },1412 threadId: command.threadId,
1413 };1413 messageId: command.message.messageId,
1414 role: "user",
1415 text: command.message.text,
1416 attachments: command.message.attachments,
1417 ...(command.message.context !== undefined
1418 ? { context: command.message.context }
1419 : {}),
1420 turnId: null,
1421 streaming: false,
1422 createdAt: command.createdAt,
1423 updatedAt: command.createdAt,
1424 },
1425 };
1414 const turnStartRequestedEvent: Omit<OrchestrationEvent, "sequence"> = {1426 const turnStartRequestedEvent: Omit<OrchestrationEvent, "sequence"> = {
1415 ...(yield* withEventBase({1427 ...(yield* withEventBase({
1416 aggregateKind: "thread",1428 aggregateKind: "thread",
...@@ -1418,7 +1430,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"...@@ -1418,7 +1430,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"
1418 occurredAt: command.createdAt,1430 occurredAt: command.createdAt,
1419 commandId: command.commandId,1431 commandId: command.commandId,
1420 })),1432 })),
1421 causationEventId: userMessageEvent.eventId,1433 ...(userMessageEvent ? { causationEventId: userMessageEvent.eventId } : {}),
1422 type: "thread.turn-start-requested",1434 type: "thread.turn-start-requested",
1423 payload: {1435 payload: {
1424 threadId: command.threadId,1436 threadId: command.threadId,
...@@ -1471,7 +1483,53 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"...@@ -1471,7 +1483,53 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"
1471 },1483 },
1472 });1484 });
1473 }1485 }
1474 return [...lifecycleResetEvents, userMessageEvent, turnStartRequestedEvent];1486 return [
1487 ...lifecycleResetEvents,
1488 ...(userMessageEvent ? [userMessageEvent] : []),
1489 turnStartRequestedEvent,
1490 ];
1491 }
1492
1493 case "thread.message.user.append": {
1494 if (isImportedAgentSessionMessageId(command.message.messageId)) {
1495 return yield* new OrchestrationCommandInvariantError({
1496 commandType: command.type,
1497 detail: `Message id '${command.message.messageId}' uses the reserved imported-session namespace.`,
1498 });
1499 }
1500 const thread = yield* requireThread({
1501 readModel,
1502 command,
1503 threadId: command.threadId,
1504 });
1505 if (thread.messages.some((message) => message.id === command.message.messageId)) {
1506 return yield* new OrchestrationCommandInvariantError({
1507 commandType: command.type,
1508 detail: `Message '${command.message.messageId}' already exists on thread '${command.threadId}'.`,
1509 });
1510 }
1511 return {
1512 ...(yield* withEventBase({
1513 aggregateKind: "thread",
1514 aggregateId: command.threadId,
1515 occurredAt: command.createdAt,
1516 commandId: command.commandId,
1517 metadata: { deferredTurn: true },
1518 })),
1519 type: "thread.message-sent",
1520 payload: {
1521 threadId: command.threadId,
1522 messageId: command.message.messageId,
1523 role: "user",
1524 text: command.message.text,
1525 attachments: command.message.attachments,
1526 ...(command.message.context !== undefined ? { context: command.message.context } : {}),
1527 turnId: null,
1528 streaming: false,
1529 createdAt: command.createdAt,
1530 updatedAt: command.createdAt,
1531 },
1532 };
1475 }1533 }
14761534
1477 case "thread.turn.interrupt": {1535 case "thread.turn.interrupt": {
apps/server/src/orchestration/decider.userMessageAppend.test.ts created+137
...@@ -0,0 +1,137 @@
1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { expect, it } from "@effect/vitest";
3import {
4 CommandId,
5 EventId,
6 MessageId,
7 ProjectId,
8 ProviderInstanceId,
9 ThreadId,
10} from "@t3tools/contracts";
11import * as Effect from "effect/Effect";
12
13import { decideOrchestrationCommand } from "./decider.ts";
14import { createEmptyReadModel, projectEvent } from "./projector.ts";
15
16const createdAt = "2026-08-24T10:00:00.000Z";
17const projectId = ProjectId.make("project-1");
18const threadId = ThreadId.make("thread-bootstrap");
19const messageId = MessageId.make("message-bootstrap");
20
21const readModelWithThread = Effect.gen(function* () {
22 const withProject = yield* projectEvent(createEmptyReadModel(createdAt), {
23 sequence: 1,
24 eventId: EventId.make("event-project-created"),
25 aggregateKind: "project",
26 aggregateId: projectId,
27 type: "project.created",
28 occurredAt: createdAt,
29 commandId: CommandId.make("command-project-created"),
30 causationEventId: null,
31 correlationId: CommandId.make("command-project-created"),
32 metadata: {},
33 payload: {
34 projectId,
35 title: "Project",
36 workspaceRoot: "/tmp/project",
37 defaultModelSelection: null,
38 scripts: [],
39 createdAt,
40 updatedAt: createdAt,
41 },
42 });
43 return yield* projectEvent(withProject, {
44 sequence: 2,
45 eventId: EventId.make("event-thread-created"),
46 aggregateKind: "thread",
47 aggregateId: threadId,
48 type: "thread.created",
49 occurredAt: createdAt,
50 commandId: CommandId.make("command-thread-created"),
51 causationEventId: null,
52 correlationId: CommandId.make("command-thread-created"),
53 metadata: {},
54 payload: {
55 threadId,
56 projectId,
57 title: "Bootstrap thread",
58 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" },
59 runtimeMode: "full-access",
60 interactionMode: "default",
61 branch: null,
62 worktreePath: null,
63 createdAt,
64 updatedAt: createdAt,
65 },
66 });
67});
68
69const appendCommand = {
70 type: "thread.message.user.append" as const,
71 commandId: CommandId.make("command-append"),
72 threadId,
73 message: { messageId, text: "Build it", attachments: [] },
74 createdAt,
75};
76
77const turnStartCommand = {
78 type: "thread.turn.start" as const,
79 commandId: CommandId.make("command-turn-start"),
80 threadId,
81 message: { messageId, role: "user" as const, text: "Build it", attachments: [] },
82 runtimeMode: "full-access" as const,
83 interactionMode: "default" as const,
84 createdAt,
85};
86
87it.layer(NodeServices.layer)("thread.message.user.append", (it) => {
88 it.effect("persists a user message without a turn, tagged as deferred", () =>
89 Effect.gen(function* () {
90 const readModel = yield* readModelWithThread;
91 const planned = yield* decideOrchestrationCommand({ command: appendCommand, readModel });
92 const events = Array.isArray(planned) ? planned : [planned];
93 expect(events.map((event) => event.type)).toEqual(["thread.message-sent"]);
94 expect(events[0]?.metadata.deferredTurn).toBe(true);
95 expect(events[0]?.payload).toMatchObject({ messageId, role: "user", turnId: null });
96 }),
97 );
98
99 it.effect("rejects a message id that already exists on the thread", () =>
100 Effect.gen(function* () {
101 const readModel = yield* readModelWithThread;
102 const first = yield* decideOrchestrationCommand({ command: appendCommand, readModel });
103 const firstEvent = Array.isArray(first) ? first[0]! : first;
104 const withMessage = yield* projectEvent(readModel, { ...firstEvent, sequence: 3 });
105 const error = yield* Effect.flip(
106 decideOrchestrationCommand({ command: appendCommand, readModel: withMessage }),
107 );
108 expect(error._tag).toBe("OrchestrationCommandInvariantError");
109 expect(error.message).toContain("already exists");
110 }),
111 );
112
113 it.effect("lets the following turn start reference the message instead of re-sending it", () =>
114 Effect.gen(function* () {
115 const readModel = yield* readModelWithThread;
116 const appended = yield* decideOrchestrationCommand({ command: appendCommand, readModel });
117 const appendedEvent = Array.isArray(appended) ? appended[0]! : appended;
118 const withMessage = yield* projectEvent(readModel, { ...appendedEvent, sequence: 3 });
119
120 const planned = yield* decideOrchestrationCommand({
121 command: turnStartCommand,
122 readModel: withMessage,
123 });
124 const events = Array.isArray(planned) ? planned : [planned];
125 expect(events.map((event) => event.type)).toEqual(["thread.turn-start-requested"]);
126 expect(events[0]?.payload).toMatchObject({ messageId });
127
128 // Without the append the turn start still carries the message itself.
129 const direct = yield* decideOrchestrationCommand({ command: turnStartCommand, readModel });
130 const directEvents = Array.isArray(direct) ? direct : [direct];
131 expect(directEvents.map((event) => event.type)).toEqual([
132 "thread.message-sent",
133 "thread.turn-start-requested",
134 ]);
135 }),
136 );
137});
apps/server/src/orchestration/projector.test.ts+66
...@@ -1170,4 +1170,70 @@ describe("orchestration projector", () => {...@@ -1170,4 +1170,70 @@ describe("orchestration projector", () => {
1170 expect(thread?.checkpoints[0]?.turnId).toBe("turn-100");1170 expect(thread?.checkpoints[0]?.turnId).toBe("turn-100");
1171 expect(thread?.checkpoints.at(-1)?.turnId).toBe("turn-599");1171 expect(thread?.checkpoints.at(-1)?.turnId).toBe("turn-599");
1172 });1172 });
1173
1174 effectIt.effect("keeps the worktree setup record past the activity retention cap", () =>
1175 Effect.gen(function* () {
1176 const createdAt = "2026-03-01T10:00:00.000Z";
1177 const threadId = "thread-setup-retained";
1178 const afterCreate = yield* projectEvent(
1179 createEmptyReadModel(createdAt),
1180 makeEvent({
1181 sequence: 1,
1182 type: "thread.created",
1183 aggregateKind: "thread",
1184 aggregateId: threadId,
1185 occurredAt: createdAt,
1186 commandId: "cmd-create-setup-retained",
1187 payload: {
1188 threadId,
1189 projectId: "project-1",
1190 title: "setup retained",
1191 modelSelection: {
1192 provider: ProviderDriverKind.make("codex"),
1193 model: "gpt-5-codex",
1194 },
1195 runtimeMode: "full-access",
1196 branch: null,
1197 worktreePath: null,
1198 createdAt,
1199 updatedAt: createdAt,
1200 },
1201 }),
1202 );
1203 const activityEvent = (sequence: number, id: string, kind: string) =>
1204 makeEvent({
1205 sequence,
1206 type: "thread.activity-appended",
1207 aggregateKind: "thread",
1208 aggregateId: threadId,
1209 occurredAt: `2026-03-01T10:${String(Math.floor(sequence / 60) % 60).padStart(2, "0")}:${String(sequence % 60).padStart(2, "0")}.000Z`,
1210 commandId: `cmd-activity-${sequence}`,
1211 payload: {
1212 threadId,
1213 activity: {
1214 id,
1215 tone: "info",
1216 kind,
1217 summary: kind,
1218 payload: {},
1219 turnId: null,
1220 createdAt: `2026-03-01T10:${String(Math.floor(sequence / 60) % 60).padStart(2, "0")}:${String(sequence % 60).padStart(2, "0")}.000Z`,
1221 },
1222 },
1223 });
1224 let model = yield* projectEvent(
1225 afterCreate,
1226 activityEvent(2, `worktree-setup:${threadId}`, "worktree-setup"),
1227 );
1228 for (let index = 0; index < 600; index += 1) {
1229 model = yield* projectEvent(
1230 model,
1231 activityEvent(3 + index, `tool-${index}`, "tool.completed"),
1232 );
1233 }
1234 const thread = model.threads.find((entry) => entry.id === threadId);
1235 expect(thread?.activities).toHaveLength(501);
1236 expect(thread?.activities[0]?.id).toBe(`worktree-setup:${threadId}`);
1237 }),
1238 );
1173});1239});
apps/server/src/orchestration/projector.ts+8-1
...@@ -13,6 +13,7 @@ import {...@@ -13,6 +13,7 @@ import {
13 OrchestrationMessage,13 OrchestrationMessage,
14 OrchestrationSession,14 OrchestrationSession,
15 OrchestrationThread,15 OrchestrationThread,
16 WORKTREE_SETUP_ACTIVITY_KIND,
16} from "@t3tools/contracts";17} from "@t3tools/contracts";
17import {18import {
18 legacyLinkedPullRequestOf,19 legacyLinkedPullRequestOf,
...@@ -76,7 +77,13 @@ function retainThreadActivities(activities: OrchestrationThread["activities"]) {...@@ -76,7 +77,13 @@ function retainThreadActivities(activities: OrchestrationThread["activities"]) {
76 }77 }
77 const pendingActivities = new Set(pending.values());78 const pendingActivities = new Set(pending.values());
78 return activities.filter(79 return activities.filter(
79 (activity, index) => index >= recentStart || pendingActivities.has(activity),80 (activity, index) =>
81 index >= recentStart ||
82 pendingActivities.has(activity) ||
83 // The worktree setup record is upserted under one id for the thread's
84 // whole life and is the only durable copy of a running setup; an async
85 // setup script can outlast a chatty first turn.
86 activity.kind === WORKTREE_SETUP_ACTIVITY_KIND,
80 );87 );
81}88}
8289
apps/server/src/project/AgentSessionScanner.test.ts+1
...@@ -38,6 +38,7 @@ const makeProjectionSnapshotQueryLayer = (importedWorkspaceRoots: ReadonlyArray<...@@ -38,6 +38,7 @@ const makeProjectionSnapshotQueryLayer = (importedWorkspaceRoots: ReadonlyArray<
38 Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {38 Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
39 getCommandReadModel: () => Effect.die("unused"),39 getCommandReadModel: () => Effect.die("unused"),
40 getUserInputActivity: () => Effect.die("unused"),40 getUserInputActivity: () => Effect.die("unused"),
41 listActivitiesByKind: () => Effect.die("unused"),
41 getSnapshot: () => Effect.die("unused"),42 getSnapshot: () => Effect.die("unused"),
42 getShellSnapshot: () =>43 getShellSnapshot: () =>
43 Effect.succeed({44 Effect.succeed({
apps/server/src/project/ProjectSetupScriptRunner.test.ts+1
...@@ -29,6 +29,7 @@ const makeProject = (scripts: OrchestrationProject["scripts"]): OrchestrationPro...@@ -29,6 +29,7 @@ const makeProject = (scripts: OrchestrationProject["scripts"]): OrchestrationPro
29const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) =>29const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) =>
30 Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {30 Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
31 getUserInputActivity: () => Effect.die("unused"),31 getUserInputActivity: () => Effect.die("unused"),
32 listActivitiesByKind: () => Effect.die("unused"),
32 getCommandReadModel: () => Effect.die("unused"),33 getCommandReadModel: () => Effect.die("unused"),
33 getSnapshot: () => Effect.die("unused"),34 getSnapshot: () => Effect.die("unused"),
34 getShellSnapshot: () => Effect.die("unused"),35 getShellSnapshot: () => Effect.die("unused"),
apps/server/src/project/WorktreeSetupTracker.ts+4-2
...@@ -63,11 +63,12 @@ export class WorktreeSetupTracker extends Context.Service<...@@ -63,11 +63,12 @@ export class WorktreeSetupTracker extends Context.Service<
63 stageId: WorktreeSetupStageId,63 stageId: WorktreeSetupStageId,
64 line: string,64 line: string,
65 ) => Effect.Effect<void>;65 ) => Effect.Effect<void>;
66 /** Returns the settled snapshot, or null when nothing was tracked. */
66 readonly finish: (67 readonly finish: (
67 threadId: ThreadId,68 threadId: ThreadId,
68 phase: "done" | "failed" | "cancelled",69 phase: "done" | "failed" | "cancelled",
69 error?: string | null,70 error?: string | null,
70 ) => Effect.Effect<void>;71 ) => Effect.Effect<WorktreeSetupSnapshot | null>;
71 /**72 /**
72 * Drops the cancel handle. Called right before the turn is dispatched so a73 * Drops the cancel handle. Called right before the turn is dispatched so a
73 * late cancel cannot roll back a thread whose agent has already started.74 * late cancel cannot roll back a thread whose agent has already started.
...@@ -279,7 +280,7 @@ export const make = Effect.gen(function* () {...@@ -279,7 +280,7 @@ export const make = Effect.gen(function* () {
279 ),280 ),
280 },281 },
281 }));282 }));
282 if (!snapshot) return;283 if (!snapshot) return null;
283 yield* clearRetention(threadId);284 yield* clearRetention(threadId);
284 const fiber = yield* remove(threadId).pipe(285 const fiber = yield* remove(threadId).pipe(
285 Effect.delay(FINISHED_RETENTION),286 Effect.delay(FINISHED_RETENTION),
...@@ -292,6 +293,7 @@ export const make = Effect.gen(function* () {...@@ -292,6 +293,7 @@ export const make = Effect.gen(function* () {
292 Effect.forkDetach,293 Effect.forkDetach,
293 );294 );
294 retentionFibers.set(threadId, fiber);295 retentionFibers.set(threadId, fiber);
296 return snapshot;
295 });297 });
296298
297 const markUncancellable: WorktreeSetupTracker["Service"]["markUncancellable"] = (threadId) =>299 const markUncancellable: WorktreeSetupTracker["Service"]["markUncancellable"] = (threadId) =>
apps/server/src/provider/Layers/ProviderService.test.ts+1
...@@ -4966,6 +4966,7 @@ describe("agent browser access", () => {...@@ -4966,6 +4966,7 @@ describe("agent browser access", () => {
4966 getTurnStartMessage: () => Effect.die("unused"),4966 getTurnStartMessage: () => Effect.die("unused"),
4967 getImportedAgentSessionSources: () => Effect.die("unused"),4967 getImportedAgentSessionSources: () => Effect.die("unused"),
4968 getUserInputActivity: () => Effect.die("unused"),4968 getUserInputActivity: () => Effect.die("unused"),
4969 listActivitiesByKind: () => Effect.die("unused"),
4969 getCommandReadModel: () => Effect.die("unused"),4970 getCommandReadModel: () => Effect.die("unused"),
4970 getSnapshot: () => Effect.die("unused"),4971 getSnapshot: () => Effect.die("unused"),
4971 getShellSnapshot: () => Effect.die("unused"),4972 getShellSnapshot: () => Effect.die("unused"),
apps/server/src/provider/Layers/ProviderSessionReaper.test.ts+1
...@@ -234,6 +234,7 @@ describe("ProviderSessionReaper", () => {...@@ -234,6 +234,7 @@ describe("ProviderSessionReaper", () => {
234 Layer.provideMerge(234 Layer.provideMerge(
235 Layer.succeed(ProjectionSnapshotQuery, {235 Layer.succeed(ProjectionSnapshotQuery, {
236 getUserInputActivity: () => Effect.die("unused"),236 getUserInputActivity: () => Effect.die("unused"),
237 listActivitiesByKind: () => Effect.die("unused"),
237 getCommandReadModel: () => Effect.die("unused"),238 getCommandReadModel: () => Effect.die("unused"),
238 getSnapshot: () => Effect.die("unused"),239 getSnapshot: () => Effect.die("unused"),
239 getShellSnapshot: () => Effect.die("unused"),240 getShellSnapshot: () => Effect.die("unused"),
apps/server/src/server.test.ts+83-19
...@@ -44,7 +44,7 @@ import {...@@ -44,7 +44,7 @@ import {
44 WS_METHODS,44 WS_METHODS,
45 WsRpcGroup,45 WsRpcGroup,
46 EditorId,46 EditorId,
47 type WorktreeSetupSnapshot,47 WorktreeSetupSnapshot,
48 type WorktreeSetupStageId,48 type WorktreeSetupStageId,
49} from "@t3tools/contracts";49} from "@t3tools/contracts";
50import {50import {
...@@ -10836,21 +10836,24 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -10836,21 +10836,24 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
10836 ),10836 ),
10837 );10837 );
1083810838
10839 assert.equal(response.sequence, 6);10839 assert.equal(response.sequence, 8);
10840 assert.deepEqual(10840 assert.deepEqual(
10841 dispatchedCommands.map((command) => command.type),10841 dispatchedCommands.map((command) => command.type),
10842 [10842 [
10843 "thread.create",10843 "thread.create",
10844 "thread.message.user.append",
10845 "thread.activity.append",
10844 "thread.session.set",10846 "thread.session.set",
10845 "thread.meta.update",10847 "thread.meta.update",
10846 "thread.activity.append",10848 "thread.activity.append",
10847 "thread.activity.append",10849 "thread.activity.append",
10848 "thread.turn.start",10850 "thread.turn.start",
10851 "thread.activity.append",
10849 ],10852 ],
10850 );10853 );
10851 // The checkout can take minutes, so the thread reads as working from10854 // The checkout can take minutes, so the thread reads as working from
10852 // the moment setup starts rather than only once the turn is dispatched.10855 // the moment setup starts rather than only once the turn is dispatched.
10853 const preparingCommand = dispatchedCommands[1];10856 const preparingCommand = dispatchedCommands[3];
10854 assertTrue(preparingCommand?.type === "thread.session.set");10857 assertTrue(preparingCommand?.type === "thread.session.set");
10855 if (preparingCommand?.type === "thread.session.set") {10858 if (preparingCommand?.type === "thread.session.set") {
10856 assert.equal(preparingCommand.session.status, "starting");10859 assert.equal(preparingCommand.session.status, "starting");
...@@ -10913,9 +10916,25 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -10913,9 +10916,25 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
10913 );10916 );
10914 assert.deepEqual(10917 assert.deepEqual(
10915 setupActivities.map((command) => command.activity.kind),10918 setupActivities.map((command) => command.activity.kind),
10916 ["setup-script.requested", "setup-script.started"],10919 ["worktree-setup", "setup-script.requested", "setup-script.started", "worktree-setup"],
10917 );10920 );
10918 const finalCommand = dispatchedCommands[5];10921 // The setup record is upserted under one id: running once the thread
10922 // exists, settled at the end, so a late client renders the outcome
10923 // without the in-memory tracker.
10924 const runningActivity = setupActivities[0]?.activity;
10925 const settledActivity = setupActivities.at(-1)?.activity;
10926 assert.equal(runningActivity?.id, settledActivity?.id);
10927 assert.equal(settledActivity?.tone, "info");
10928 assertTrue(Schema.is(WorktreeSetupSnapshot)(runningActivity?.payload));
10929 if (Schema.is(WorktreeSetupSnapshot)(runningActivity?.payload)) {
10930 assert.equal(runningActivity.payload.phase, "running");
10931 }
10932 assertTrue(Schema.is(WorktreeSetupSnapshot)(settledActivity?.payload));
10933 if (Schema.is(WorktreeSetupSnapshot)(settledActivity?.payload)) {
10934 assert.equal(settledActivity.payload.phase, "done");
10935 assert.equal(settledActivity.payload.threadId, ThreadId.make("thread-bootstrap"));
10936 }
10937 const finalCommand = dispatchedCommands[7];
10919 assertTrue(finalCommand?.type === "thread.turn.start");10938 assertTrue(finalCommand?.type === "thread.turn.start");
10920 if (finalCommand?.type === "thread.turn.start") {10939 if (finalCommand?.type === "thread.turn.start") {
10921 assert.equal(finalCommand.bootstrap, undefined);10940 assert.equal(finalCommand.bootstrap, undefined);
...@@ -11110,13 +11129,19 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -11110,13 +11129,19 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
11110 ),11129 ),
11111 );11130 );
1111211131
11113 assert.equal(response.sequence, 2);11132 assert.equal(response.sequence, 4);
11114 assert.equal(createWorktree.mock.calls.length, 0);11133 assert.equal(createWorktree.mock.calls.length, 0);
11115 assert.deepEqual(11134 assert.deepEqual(
11116 dispatchedCommands.map((command) => command.type),11135 dispatchedCommands.map((command) => command.type),
11117 ["thread.create", "thread.turn.start"],11136 [
11137 "thread.create",
11138 "thread.message.user.append",
11139 "thread.activity.append",
11140 "thread.turn.start",
11141 "thread.activity.append",
11142 ],
11118 );11143 );
11119 const finalCommand = dispatchedCommands[1];11144 const finalCommand = dispatchedCommands[3];
11120 assertTrue(finalCommand?.type === "thread.turn.start");11145 assertTrue(finalCommand?.type === "thread.turn.start");
11121 if (finalCommand?.type === "thread.turn.start") {11146 if (finalCommand?.type === "thread.turn.start") {
11122 assert.equal(finalCommand.bootstrap, undefined);11147 assert.equal(finalCommand.bootstrap, undefined);
...@@ -11197,11 +11222,17 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -11197,11 +11222,17 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
11197 ),11222 ),
11198 );11223 );
1119911224
11200 assert.equal(response.sequence, 2);11225 assert.equal(response.sequence, 4);
11201 assert.equal(createWorktree.mock.calls.length, 0);11226 assert.equal(createWorktree.mock.calls.length, 0);
11202 assert.deepEqual(11227 assert.deepEqual(
11203 dispatchedCommands.map((command) => command.type),11228 dispatchedCommands.map((command) => command.type),
11204 ["thread.create", "thread.turn.start"],11229 [
11230 "thread.create",
11231 "thread.message.user.append",
11232 "thread.activity.append",
11233 "thread.turn.start",
11234 "thread.activity.append",
11235 ],
11205 );11236 );
11206 }).pipe(Effect.provide(NodeHttpServer.layerTest)),11237 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
11207 );11238 );
...@@ -11297,20 +11328,23 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -11297,20 +11328,23 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
11297 ),11328 ),
11298 );11329 );
1129911330
11300 assert.equal(response.sequence, 5);11331 assert.equal(response.sequence, 7);
11301 assert.deepEqual(11332 assert.deepEqual(
11302 dispatchedCommands.map((command) => command.type),11333 dispatchedCommands.map((command) => command.type),
11303 [11334 [
11304 "thread.create",11335 "thread.create",
11336 "thread.message.user.append",
11337 "thread.activity.append",
11305 "thread.session.set",11338 "thread.session.set",
11306 "thread.meta.update",11339 "thread.meta.update",
11307 "thread.activity.append",11340 "thread.activity.append",
11308 "thread.turn.start",11341 "thread.turn.start",
11342 "thread.activity.append",
11309 ],11343 ],
11310 );11344 );
11311 const setupFailureActivity = dispatchedCommands.find(11345 const setupFailureActivity = dispatchedCommands.find(
11312 (command): command is Extract<OrchestrationCommand, { type: "thread.activity.append" }> =>11346 (command): command is Extract<OrchestrationCommand, { type: "thread.activity.append" }> =>
11313 command.type === "thread.activity.append",11347 command.type === "thread.activity.append" && command.activity.kind !== "worktree-setup",
11314 );11348 );
11315 assert.equal(setupFailureActivity?.activity.kind, "setup-script.failed");11349 assert.equal(setupFailureActivity?.activity.kind, "setup-script.failed");
11316 assert.deepEqual(setupFailureActivity?.activity.payload, {11350 assert.deepEqual(setupFailureActivity?.activity.payload, {
...@@ -11430,15 +11464,18 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -11430,15 +11464,18 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
11430 ),11464 ),
11431 );11465 );
1143211466
11433 assert.equal(response.sequence, 5);11467 assert.equal(response.sequence, 7);
11434 assert.deepEqual(11468 assert.deepEqual(
11435 dispatchedCommands.map((command) => command.type),11469 dispatchedCommands.map((command) => command.type),
11436 [11470 [
11437 "thread.create",11471 "thread.create",
11472 "thread.message.user.append",
11473 "thread.activity.append",
11438 "thread.session.set",11474 "thread.session.set",
11439 "thread.meta.update",11475 "thread.meta.update",
11440 "thread.activity.append",11476 "thread.activity.append",
11441 "thread.turn.start",11477 "thread.turn.start",
11478 "thread.activity.append",
11442 ],11479 ],
11443 );11480 );
11444 const setupActivities = dispatchedCommands.filter(11481 const setupActivities = dispatchedCommands.filter(
...@@ -11447,7 +11484,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -11447,7 +11484,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
11447 );11484 );
11448 assert.deepEqual(11485 assert.deepEqual(
11449 setupActivities.map((command) => command.activity.kind),11486 setupActivities.map((command) => command.activity.kind),
11450 ["setup-script.requested"],11487 ["worktree-setup", "setup-script.requested", "worktree-setup"],
11451 );11488 );
11452 assertTrue(11489 assertTrue(
11453 setupActivities.every((command) => command.activity.kind !== "setup-script.failed"),11490 setupActivities.every((command) => command.activity.kind !== "setup-script.failed"),
...@@ -11593,8 +11630,15 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -11593,8 +11630,15 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
11593 assert.equal(stageStatus(running, "agent"), "pending");11630 assert.equal(stageStatus(running, "agent"), "pending");
11594 assert.isFalse(turnStarted());11631 assert.isFalse(turnStarted());
1159511632
11633 // The client that sent the message goes away mid-setup (a reload or a
11634 // dropped socket). The bootstrap belongs to the server, not the
11635 // connection: the thread already exists for every client, so it must
11636 // finish and start the turn regardless.
11637 yield* Fiber.interrupt(dispatchFiber);
11638 assert.isFalse(turnStarted());
11639
11596 yield* Deferred.succeed(scriptExit, undefined);11640 yield* Deferred.succeed(scriptExit, undefined);
11597 yield* Fiber.join(dispatchFiber);11641 yield* snapshotWhere((snapshot) => stageStatus(snapshot, "agent") === "done");
11598 assertTrue(turnStarted());11642 assertTrue(turnStarted());
11599 const settled = yield* snapshotWhere((snapshot) => snapshot.phase !== "running");11643 const settled = yield* snapshotWhere((snapshot) => snapshot.phase !== "running");
11600 assert.equal(settled.phase, "done");11644 assert.equal(settled.phase, "done");
...@@ -11701,7 +11745,14 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -11701,7 +11745,14 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
11701 assert.strictEqual(result.failure.bootstrapThreadDisposition, "deleted");11745 assert.strictEqual(result.failure.bootstrapThreadDisposition, "deleted");
11702 assert.deepEqual(11746 assert.deepEqual(
11703 dispatchedCommands.map((command) => command.type),11747 dispatchedCommands.map((command) => command.type),
11704 ["thread.create", "thread.session.set", "thread.delete"],11748 [
11749 "thread.create",
11750 "thread.message.user.append",
11751 "thread.activity.append",
11752 "thread.session.set",
11753 "thread.activity.append",
11754 "thread.delete",
11755 ],
11705 );11756 );
11706 assert.isDefined(pendingAttachmentId);11757 assert.isDefined(pendingAttachmentId);
11707 assert.isTrue(11758 assert.isTrue(
...@@ -11814,7 +11865,12 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -11814,7 +11865,12 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
11814 }),11865 }),
11815 ),11866 ),
11816 );11867 );
11817 assert.deepEqual(trace, ["thread.create", "drain:1", "thread.turn.start"]);11868 assert.deepEqual(trace, [
11869 "thread.create",
11870 "drain:1",
11871 "thread.message.user.append",
11872 "thread.turn.start",
11873 ]);
11818 }).pipe(Effect.provide(NodeHttpServer.layerTest)),11874 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
11819 );11875 );
1182011876
...@@ -11899,11 +11955,19 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -11899,11 +11955,19 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
11899 assert.strictEqual(result.failure.bootstrapThreadDisposition, undefined);11955 assert.strictEqual(result.failure.bootstrapThreadDisposition, undefined);
11900 assert.deepEqual(11956 assert.deepEqual(
11901 dispatchedCommands.map((command) => command.type),11957 dispatchedCommands.map((command) => command.type),
11902 ["thread.create", "thread.session.set", "thread.delete", "thread.session.set"],11958 [
11959 "thread.create",
11960 "thread.message.user.append",
11961 "thread.activity.append",
11962 "thread.session.set",
11963 "thread.activity.append",
11964 "thread.delete",
11965 "thread.session.set",
11966 ],
11903 );11967 );
11904 // The surviving thread must not keep its preparing session, or it would11968 // The surviving thread must not keep its preparing session, or it would
11905 // read as working forever.11969 // read as working forever.
11906 const failedSession = dispatchedCommands[3];11970 const failedSession = dispatchedCommands[6];
11907 assertTrue(failedSession?.type === "thread.session.set");11971 assertTrue(failedSession?.type === "thread.session.set");
11908 if (failedSession?.type === "thread.session.set") {11972 if (failedSession?.type === "thread.session.set") {
11909 assert.equal(failedSession.session.status, "error");11973 assert.equal(failedSession.session.status, "error");
apps/server/src/serverRuntimeStartup.test.ts+4
...@@ -164,6 +164,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa...@@ -164,6 +164,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa
164 } as never),164 } as never),
165 Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {165 Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
166 getUserInputActivity: () => Effect.die("unused"),166 getUserInputActivity: () => Effect.die("unused"),
167 listActivitiesByKind: () => Effect.succeed([]),
167 getCommandReadModel: () => Effect.die("unused"),168 getCommandReadModel: () => Effect.die("unused"),
168 getSnapshot: () => Effect.die("unused"),169 getSnapshot: () => Effect.die("unused"),
169 getShellSnapshot: () => Effect.die("unused"),170 getShellSnapshot: () => Effect.die("unused"),
...@@ -293,6 +294,7 @@ it.effect.each([...@@ -293,6 +294,7 @@ it.effect.each([
293 } as never),294 } as never),
294 Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {295 Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
295 getUserInputActivity: () => Effect.die("unused"),296 getUserInputActivity: () => Effect.die("unused"),
297 listActivitiesByKind: () => Effect.succeed([]),
296 getCommandReadModel: () => Effect.die("unused"),298 getCommandReadModel: () => Effect.die("unused"),
297 getSnapshot: () => Effect.die("unused"),299 getSnapshot: () => Effect.die("unused"),
298 getShellSnapshot: () => Effect.die("unused"),300 getShellSnapshot: () => Effect.die("unused"),
...@@ -378,6 +380,7 @@ it.effect(...@@ -378,6 +380,7 @@ it.effect(
378 } as never),380 } as never),
379 Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {381 Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
380 getUserInputActivity: () => Effect.die("unused"),382 getUserInputActivity: () => Effect.die("unused"),
383 listActivitiesByKind: () => Effect.succeed([]),
381 getCommandReadModel: () => Effect.die("unused"),384 getCommandReadModel: () => Effect.die("unused"),
382 getSnapshot: () => Effect.die("unused"),385 getSnapshot: () => Effect.die("unused"),
383 getShellSnapshot: () => Effect.die("unused"),386 getShellSnapshot: () => Effect.die("unused"),
...@@ -441,6 +444,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa...@@ -441,6 +444,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa
441 } as never),444 } as never),
442 Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {445 Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
443 getUserInputActivity: () => Effect.die("unused"),446 getUserInputActivity: () => Effect.die("unused"),
447 listActivitiesByKind: () => Effect.succeed([]),
444 getCommandReadModel: () => Effect.die("unused"),448 getCommandReadModel: () => Effect.die("unused"),
445 getSnapshot: () => Effect.die("unused"),449 getSnapshot: () => Effect.die("unused"),
446 getShellSnapshot: () => Effect.die("unused"),450 getShellSnapshot: () => Effect.die("unused"),
apps/server/src/serverRuntimeStartup.ts+90
...@@ -1,5 +1,6 @@...@@ -1,5 +1,6 @@
1import {1import {
2 CommandId,2 CommandId,
3 EventId,
3 DEFAULT_MODEL,4 DEFAULT_MODEL,
4 DEFAULT_PROVIDER_INTERACTION_MODE,5 DEFAULT_PROVIDER_INTERACTION_MODE,
5 DEFAULT_SERVER_SETTINGS,6 DEFAULT_SERVER_SETTINGS,
...@@ -10,6 +11,9 @@ import {...@@ -10,6 +11,9 @@ import {
10 ProviderInstanceId,11 ProviderInstanceId,
11 ThreadId,12 ThreadId,
12 TurnId,13 TurnId,
14 WORKTREE_SETUP_ACTIVITY_KIND,
15 WorktreeSetupSnapshot,
16 worktreeSetupActivityId,
13} from "@t3tools/contracts";17} from "@t3tools/contracts";
14import { resolveProjectSettings } from "@t3tools/shared/projectSettings";18import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
15import * as Cause from "effect/Cause";19import * as Cause from "effect/Cause";
...@@ -742,6 +746,91 @@ export const reconcileProviderSessions = Effect.gen(function* () {...@@ -742,6 +746,91 @@ export const reconcileProviderSessions = Effect.gen(function* () {
742 ),746 ),
743);747);
744748
749const decodeWorktreeSetupSnapshot = Schema.decodeUnknownOption(WorktreeSetupSnapshot);
750
751/**
752 * A worktree bootstrap records its setup snapshot on the thread while it runs
753 * and settles it when it finishes. The bootstrap itself lives only in memory,
754 * so a process exit mid-setup leaves a `running` record with nobody to finish
755 * it. Before the turn started that also strands the persisted user message, so
756 * the setup is marked failed and the user is told to send again. After the
757 * handoff only an async setup script was still running; its stage is marked
758 * failed and the setup settles as done, like any other script failure.
759 */
760export const reconcileWorktreeSetups = Effect.gen(function* () {
761 const crypto = yield* Crypto.Crypto;
762 const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService;
763 const query = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;
764 // The command read model carries no activity bodies; read the setup
765 // records directly, live threads only.
766 const recordedSetups = yield* query.listActivitiesByKind(WORKTREE_SETUP_ACTIVITY_KIND);
767 const interruptedAt = DateTime.formatIso(yield* DateTime.now);
768
769 for (const recorded of recordedSetups) {
770 const snapshot = decodeWorktreeSetupSnapshot(recorded.payload);
771 if (Option.isNone(snapshot) || snapshot.value.phase !== "running") continue;
772 if (recorded.id !== worktreeSetupActivityId(snapshot.value.threadId)) continue;
773 const threadId = snapshot.value.threadId;
774
775 const turnStarted = snapshot.value.stages.some(
776 (stage) => stage.id === "agent" && stage.status === "done",
777 );
778 const interrupted: WorktreeSetupSnapshot = {
779 ...snapshot.value,
780 phase: turnStarted ? "done" : "failed",
781 endedAt: interruptedAt,
782 error: turnStarted
783 ? null
784 : "The server restarted before the worktree setup finished. Send the message again.",
785 stages: snapshot.value.stages.map((stage) =>
786 stage.status === "running" || stage.status === "pending"
787 ? {
788 ...stage,
789 status: "failed",
790 endedAt: interruptedAt,
791 detail: "interrupted by a server restart",
792 }
793 : stage,
794 ),
795 sequence: snapshot.value.sequence + 1,
796 };
797 yield* orchestrationEngine
798 .dispatch({
799 type: "thread.activity.append",
800 commandId: CommandId.make(yield* crypto.randomUUIDv4),
801 threadId,
802 activity: {
803 id: EventId.make(worktreeSetupActivityId(threadId)),
804 tone: "error",
805 kind: WORKTREE_SETUP_ACTIVITY_KIND,
806 summary: turnStarted
807 ? "Setup script interrupted by a server restart"
808 : "Worktree setup interrupted by a server restart",
809 payload: interrupted,
810 turnId: null,
811 createdAt: snapshot.value.startedAt,
812 },
813 createdAt: interruptedAt,
814 })
815 .pipe(
816 Effect.catchCause((cause) =>
817 Cause.hasInterrupts(cause)
818 ? Effect.failCause(cause)
819 : Effect.logWarning("failed to settle interrupted worktree setup", {
820 threadId,
821 cause,
822 }),
823 ),
824 );
825 }
826}).pipe(
827 Effect.catchCause((cause) =>
828 Cause.hasInterrupts(cause)
829 ? Effect.failCause(cause)
830 : Effect.logWarning("worktree setup startup reconciliation failed", { cause }),
831 ),
832);
833
745interface StartupOptions {834interface StartupOptions {
746 readonly activate?: Effect.Effect<void>;835 readonly activate?: Effect.Effect<void>;
747 readonly awaitAuxiliaryParked?: Effect.Effect<void>;836 readonly awaitAuxiliaryParked?: Effect.Effect<void>;
...@@ -881,6 +970,7 @@ export const make = (options?: StartupOptions) =>...@@ -881,6 +970,7 @@ export const make = (options?: StartupOptions) =>
881 );970 );
882971
883 yield* runStartupPhase("provider-sessions.reconcile", reconcileProviderSessions);972 yield* runStartupPhase("provider-sessions.reconcile", reconcileProviderSessions);
973 yield* runStartupPhase("worktree-setups.reconcile", reconcileWorktreeSetups);
884974
885 yield* Effect.logDebug("startup phase: syncing clean projects");975 yield* Effect.logDebug("startup phase: syncing clean projects");
886 yield* runStartupPhase("projects.auto-pull", syncAutoPullProjects);976 yield* runStartupPhase("projects.auto-pull", syncAutoPullProjects);
apps/server/src/serverRuntimeStartup.worktreeSetup.test.ts created+156
...@@ -0,0 +1,156 @@
1import * as NodeServices from "@effect/platform-node/NodeServices";
2import {
3 EventId,
4 type OrchestrationCommand,
5 ThreadId,
6 WORKTREE_SETUP_ACTIVITY_KIND,
7 WorktreeSetupSnapshot,
8 worktreeSetupActivityId,
9 type WorktreeSetupPhase,
10} from "@t3tools/contracts";
11import { assert, it } from "@effect/vitest";
12import * as Effect from "effect/Effect";
13import * as Schema from "effect/Schema";
14import * as Stream from "effect/Stream";
15
16import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts";
17import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts";
18import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts";
19
20const startedAt = "2026-08-20T12:00:00.000Z";
21
22const snapshotFor = (
23 threadId: ThreadId,
24 phase: WorktreeSetupPhase,
25 agentStatus: "pending" | "done" = phase === "running" ? "pending" : "done",
26): WorktreeSetupSnapshot => ({
27 threadId,
28 phase,
29 startedAt,
30 endedAt: phase === "running" ? null : startedAt,
31 branch: "feature",
32 baseRef: "main",
33 worktreePath: null,
34 setupScript: null,
35 stages: [
36 {
37 id: "checkout",
38 status: "done",
39 startedAt,
40 endedAt: startedAt,
41 percent: null,
42 detail: null,
43 tail: [],
44 },
45 {
46 id: "setup-script",
47 status: phase === "running" ? "running" : "done",
48 startedAt,
49 endedAt: phase === "running" ? null : startedAt,
50 percent: null,
51 detail: null,
52 tail: [],
53 },
54 {
55 id: "agent",
56 status: agentStatus,
57 startedAt: null,
58 endedAt: null,
59 percent: null,
60 detail: null,
61 tail: [],
62 },
63 ],
64 error: null,
65 sequence: 4,
66});
67
68const recordedSetup = (id: string, phase: WorktreeSetupPhase, agentStatus?: "pending" | "done") => {
69 const threadId = ThreadId.make(id);
70 return {
71 id: EventId.make(worktreeSetupActivityId(threadId)),
72 tone: "info" as const,
73 kind: WORKTREE_SETUP_ACTIVITY_KIND,
74 summary: "Setting up worktree",
75 payload: snapshotFor(threadId, phase, agentStatus),
76 turnId: null,
77 createdAt: startedAt,
78 };
79};
80
81const run = (activities: ReadonlyArray<ReturnType<typeof recordedSetup>>) =>
82 Effect.gen(function* () {
83 const dispatched: Array<OrchestrationCommand> = [];
84 yield* ServerRuntimeStartup.reconcileWorktreeSetups.pipe(
85 Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
86 listActivitiesByKind: (kind: string) =>
87 Effect.succeed(kind === WORKTREE_SETUP_ACTIVITY_KIND ? activities : []),
88 } as unknown as ProjectionSnapshotQuery.ProjectionSnapshotQuery["Service"]),
89 Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {
90 readEvents: () => Stream.empty,
91 readThreadEvents: () => Stream.empty,
92 getThreadReplayStats: () => Effect.die("unused"),
93 dispatch: (command) =>
94 Effect.sync(() => {
95 dispatched.push(command);
96 return { sequence: dispatched.length };
97 }),
98 streamDomainEvents: Stream.empty,
99 subscribeDomainEvents: Effect.succeed(Stream.empty),
100 latestSequence: Effect.succeed(0),
101 }),
102 Effect.provide(NodeServices.layer),
103 );
104 return dispatched;
105 });
106
107it.effect("marks setups still recorded as running failed after a restart", () =>
108 Effect.gen(function* () {
109 const dispatched = yield* run([
110 recordedSetup("thread-running", "running"),
111 recordedSetup("thread-done", "done"),
112 recordedSetup("thread-failed", "failed"),
113 ]);
114
115 assert.equal(dispatched.length, 1);
116 const command = dispatched[0]!;
117 assert.equal(command.type, "thread.activity.append");
118 if (command.type !== "thread.activity.append") return;
119 assert.equal(command.threadId, ThreadId.make("thread-running"));
120 assert.equal(command.activity.id, worktreeSetupActivityId(ThreadId.make("thread-running")));
121 assert.equal(command.activity.tone, "error");
122 const payload = yield* Schema.decodeUnknownEffect(WorktreeSetupSnapshot)(
123 command.activity.payload,
124 );
125 assert.equal(payload.phase, "failed");
126 assert.isNotNull(payload.endedAt);
127 assert.equal(payload.sequence, 5);
128 assert.deepEqual(
129 payload.stages.map((stage) => stage.status),
130 ["done", "failed", "failed"],
131 );
132 }),
133);
134
135it.effect(
136 "settles an async setup script whose turn already started without failing the setup",
137 () =>
138 Effect.gen(function* () {
139 const dispatched = yield* run([recordedSetup("thread-async", "running", "done")]);
140
141 assert.equal(dispatched.length, 1);
142 const command = dispatched[0]!;
143 if (command.type !== "thread.activity.append") return assert.fail(command.type);
144 const payload = yield* Schema.decodeUnknownEffect(WorktreeSetupSnapshot)(
145 command.activity.payload,
146 );
147 // The turn is live; only the background script was lost. Nothing asks the
148 // user to resend, and the setup reads as done with a failed script stage.
149 assert.equal(payload.phase, "done");
150 assert.isNull(payload.error);
151 assert.deepEqual(
152 payload.stages.map((stage) => stage.status),
153 ["done", "failed", "done"],
154 );
155 }),
156);
apps/server/src/ws.ts+118-18
...@@ -76,6 +76,9 @@ import {...@@ -76,6 +76,9 @@ import {
76 type PullRequestRef,76 type PullRequestRef,
77 WS_METHODS,77 WS_METHODS,
78 WsRpcGroup,78 WsRpcGroup,
79 WORKTREE_SETUP_ACTIVITY_KIND,
80 worktreeSetupActivityId,
81 type WorktreeSetupSnapshot,
79} from "@t3tools/contracts";82} from "@t3tools/contracts";
80import { resolveServerBackgroundActivitySettings } from "@t3tools/shared/backgroundActivitySettings";83import { resolveServerBackgroundActivitySettings } from "@t3tools/shared/backgroundActivitySettings";
81import { HttpRouter, HttpServerRequest, HttpServerRespondable } from "effect/unstable/http";84import { HttpRouter, HttpServerRequest, HttpServerRespondable } from "effect/unstable/http";
...@@ -775,6 +778,44 @@ const makeWsRpcLayer = (...@@ -775,6 +778,44 @@ const makeWsRpcLayer = (
775 ),778 ),
776 );779 );
777780
781 // The worktree setup's durable record: one activity per thread, upserted
782 // by a fixed id when the setup starts and again when it settles. Live
783 // progress keeps streaming from the tracker; this is what a reload or
784 // another client reads. Best effort: the thread may already be gone
785 // after a failed bootstrap.
786 const recordWorktreeSetup = (snapshot: WorktreeSetupSnapshot) =>
787 serverCommandId("worktree-setup-activity").pipe(
788 Effect.flatMap((commandId) =>
789 dispatchFromClient({
790 type: "thread.activity.append",
791 commandId,
792 threadId: snapshot.threadId,
793 activity: {
794 id: EventId.make(worktreeSetupActivityId(snapshot.threadId)),
795 tone:
796 snapshot.phase === "failed" ||
797 snapshot.stages.some((stage) => stage.status === "failed")
798 ? "error"
799 : "info",
800 kind: WORKTREE_SETUP_ACTIVITY_KIND,
801 summary:
802 snapshot.phase === "running"
803 ? "Setting up worktree"
804 : snapshot.phase === "done"
805 ? "Worktree ready"
806 : snapshot.phase === "cancelled"
807 ? "Worktree setup cancelled"
808 : "Worktree setup failed",
809 payload: snapshot,
810 turnId: null,
811 createdAt: snapshot.startedAt,
812 },
813 createdAt: snapshot.endedAt ?? snapshot.startedAt,
814 }),
815 ),
816 Effect.ignoreCause({ log: true }),
817 );
818
778 const toBootstrapDispatchCommandCauseError = (cause: Cause.Cause<unknown>) => {819 const toBootstrapDispatchCommandCauseError = (cause: Cause.Cause<unknown>) => {
779 const error = Cause.squash(cause);820 const error = Cause.squash(cause);
780 return isOrchestrationDispatchCommandError(error)821 return isOrchestrationDispatchCommandError(error)
...@@ -1361,6 +1402,28 @@ const makeWsRpcLayer = (...@@ -1361,6 +1402,28 @@ const makeWsRpcLayer = (
1361 // terminals and provider sessions under the reused thread id.1402 // terminals and provider sessions under the reused thread id.
1362 yield* threadDeletionReactor.drainThrough(created.sequence);1403 yield* threadDeletionReactor.drainThrough(created.sequence);
1363 createdThread = true;1404 createdThread = true;
1405 // Persist the send now rather than with the turn: the thread is
1406 // real from here on, so any client (or a reload) sees the message
1407 // while the worktree is still being prepared. The turn start
1408 // later references this id instead of re-sending the text.
1409 yield* dispatchFromClient({
1410 type: "thread.message.user.append",
1411 commandId: yield* serverCommandId("bootstrap-thread-message"),
1412 threadId: command.threadId,
1413 message: {
1414 messageId: command.message.messageId,
1415 text: command.message.text,
1416 attachments: command.message.attachments,
1417 ...(command.message.context !== undefined
1418 ? { context: command.message.context }
1419 : {}),
1420 },
1421 createdAt: command.createdAt,
1422 });
1423 if (tracked) {
1424 const running = yield* worktreeSetupTracker.get(threadId);
1425 if (running) yield* recordWorktreeSetup(running);
1426 }
1364 }1427 }
13651428
1366 if (prepareWorktree && shouldPrepareWorktree && worktreeBaseRef) {1429 if (prepareWorktree && shouldPrepareWorktree && worktreeBaseRef) {
...@@ -1496,7 +1559,15 @@ const makeWsRpcLayer = (...@@ -1496,7 +1559,15 @@ const makeWsRpcLayer = (
1496 // running so the client keeps its row next to the agent's work,1559 // running so the client keeps its row next to the agent's work,
1497 // and settles when the script exits. The turn already started, so1560 // and settles when the script exits. The turn already started, so
1498 // the wait cannot fail the dispatch.1561 // the wait cannot fail the dispatch.
1499 const settle = track(worktreeSetupTracker.finish(threadId, "done"));1562 const settle = tracked
1563 ? worktreeSetupTracker
1564 .finish(threadId, "done")
1565 .pipe(
1566 Effect.flatMap((snapshot) =>
1567 snapshot ? recordWorktreeSetup(snapshot) : Effect.void,
1568 ),
1569 )
1570 : Effect.void;
1500 if (pendingSetupScript) {1571 if (pendingSetupScript) {
1501 yield* Fiber.join(pendingSetupScript).pipe(1572 yield* Fiber.join(pendingSetupScript).pipe(
1502 Effect.ignoreCause({ log: true }),1573 Effect.ignoreCause({ log: true }),
...@@ -1509,20 +1580,6 @@ const makeWsRpcLayer = (...@@ -1509,20 +1580,6 @@ const makeWsRpcLayer = (
1509 return started;1580 return started;
1510 });1581 });
15111582
1512 const runBootstrap = tracked
1513 ? Effect.gen(function* () {
1514 const fiber = yield* Effect.forkChild(bootstrapProgram);
1515 yield* worktreeSetupTracker.begin({
1516 threadId,
1517 branch: bootstrap?.prepareWorktree?.branch ?? null,
1518 baseRef: bootstrap?.prepareWorktree?.baseBranch ?? null,
1519 stages: ["fetch", "checkout", "submodules", "setup-script", "agent"],
1520 fiber,
1521 });
1522 return yield* Fiber.join(fiber);
1523 })
1524 : bootstrapProgram;
1525
1526 const cleanupAndFail = (1583 const cleanupAndFail = (
1527 cause: Cause.Cause<unknown>,1584 cause: Cause.Cause<unknown>,
1528 dispatchError: OrchestrationDispatchCommandError,1585 dispatchError: OrchestrationDispatchCommandError,
...@@ -1561,7 +1618,7 @@ const makeWsRpcLayer = (...@@ -1561,7 +1618,7 @@ const makeWsRpcLayer = (
1561 }),1618 }),
1562 );1619 );
15631620
1564 return yield* runBootstrap.pipe(1621 const settledBootstrapProgram = bootstrapProgram.pipe(
1565 Effect.catchCause((cause) => {1622 Effect.catchCause((cause) => {
1566 const dispatchError = toBootstrapDispatchCommandCauseError(cause);1623 const dispatchError = toBootstrapDispatchCommandCauseError(cause);
1567 if (Cause.hasInterruptsOnly(cause)) {1624 if (Cause.hasInterruptsOnly(cause)) {
...@@ -1597,7 +1654,15 @@ const makeWsRpcLayer = (...@@ -1597,7 +1654,15 @@ const makeWsRpcLayer = (
1597 Effect.uninterruptible,1654 Effect.uninterruptible,
1598 )1655 )
1599 : Effect.void;1656 : Effect.void;
1600 return track(worktreeSetupTracker.finish(threadId, "cancelled")).pipe(1657 return track(
1658 worktreeSetupTracker
1659 .finish(threadId, "cancelled")
1660 .pipe(
1661 Effect.flatMap((snapshot) =>
1662 snapshot ? recordWorktreeSetup(snapshot) : Effect.void,
1663 ),
1664 ),
1665 ).pipe(
1601 Effect.andThen(removeCreatedWorktree),1666 Effect.andThen(removeCreatedWorktree),
1602 Effect.andThen(1667 Effect.andThen(
1603 tracked1668 tracked
...@@ -1612,10 +1677,45 @@ const makeWsRpcLayer = (...@@ -1612,10 +1677,45 @@ const makeWsRpcLayer = (
1612 );1677 );
1613 }1678 }
1614 return track(1679 return track(
1615 worktreeSetupTracker.finish(threadId, "failed", dispatchError.message),1680 worktreeSetupTracker
1681 .finish(threadId, "failed", dispatchError.message)
1682 .pipe(
1683 Effect.flatMap((snapshot) =>
1684 snapshot ? recordWorktreeSetup(snapshot) : Effect.void,
1685 ),
1686 ),
1616 ).pipe(Effect.andThen(cleanupAndFail(cause, dispatchError)));1687 ).pipe(Effect.andThen(cleanupAndFail(cause, dispatchError)));
1617 }),1688 }),
1618 );1689 );
1690
1691 // The bootstrap outlives the connection that asked for it: a reload
1692 // or a dropped socket must not abandon a half-made worktree, and
1693 // the thread it created is already visible to every client. The
1694 // RPC only waits on the detached fiber; a user cancel interrupts it
1695 // through the tracker.
1696 const runBootstrap = tracked
1697 ? Effect.gen(function* () {
1698 // Fork and register as one step: a detached fiber keeps going
1699 // if the caller is interrupted, so it must never exist without
1700 // the tracker entry that cancel and the stage updates key on.
1701 const fiber = yield* Effect.uninterruptible(
1702 Effect.gen(function* () {
1703 const fiber = yield* Effect.forkDetach(settledBootstrapProgram);
1704 yield* worktreeSetupTracker.begin({
1705 threadId,
1706 branch: bootstrap?.prepareWorktree?.branch ?? null,
1707 baseRef: bootstrap?.prepareWorktree?.baseBranch ?? null,
1708 stages: ["fetch", "checkout", "submodules", "setup-script", "agent"],
1709 fiber,
1710 });
1711 return fiber;
1712 }),
1713 );
1714 return yield* Fiber.join(fiber);
1715 })
1716 : settledBootstrapProgram;
1717
1718 return yield* runBootstrap;
1619 });1719 });
16201720
1621 const dispatchNormalizedCommand = (1721 const dispatchNormalizedCommand = (
apps/web/src/components/ChatView.logic.test.ts+159-12
...@@ -59,6 +59,8 @@ import {...@@ -59,6 +59,8 @@ import {
59 restorePlanFollowUpComposer,59 restorePlanFollowUpComposer,
60 resolveComposerProviderSelection,60 resolveComposerProviderSelection,
61 resolveDraftPromotionNavigationTarget,61 resolveDraftPromotionNavigationTarget,
62 findRecordedWorktreeSetup,
63 resolveVisibleWorktreeSetup,
62 observeProactivePanelUserChoice,64 observeProactivePanelUserChoice,
63 resolveProactiveTurnDiffAction,65 resolveProactiveTurnDiffAction,
64 resolveThreadMetadataUpdateForNextTurn,66 resolveThreadMetadataUpdateForNextTurn,
...@@ -1016,20 +1018,10 @@ describe("draft promotion during worktree setup", () => {...@@ -1016,20 +1018,10 @@ describe("draft promotion during worktree setup", () => {
1016 const serverThreadRef = { environmentId, threadId };1018 const serverThreadRef = { environmentId, threadId };
10171019
1018 it.each([null, "idle", "starting", "ready"] as const)(1020 it.each([null, "idle", "starting", "ready"] as const)(
1019 "keeps the draft mounted while the first turn waits with session %s",1021 "keeps the draft mounted until the server owns the send, with session %s",
1020 (status) => {1022 (status) => {
1021 const serverThread = makeThread({1023 const serverThread = makeThread({
1022 messages: [1024 messages: [],
1023 {
1024 id: MessageId.make("submitted-message"),
1025 role: "user",
1026 text: "Start in a new worktree",
1027 turnId: null,
1028 createdAt: now,
1029 updatedAt: now,
1030 streaming: false,
1031 },
1032 ],
1033 session: status ? { ...readySession, status } : null,1025 session: status ? { ...readySession, status } : null,
1034 });1026 });
10351027
...@@ -1043,6 +1035,31 @@ describe("draft promotion during worktree setup", () => {...@@ -1043,6 +1035,31 @@ describe("draft promotion during worktree setup", () => {
1043 },1035 },
1044 );1036 );
10451037
1038 it("promotes once the bootstrap persisted the user message, before any turn", () => {
1039 const serverThread = makeThread({
1040 messages: [
1041 {
1042 id: MessageId.make("submitted-message"),
1043 role: "user",
1044 text: "Start in a new worktree",
1045 turnId: null,
1046 createdAt: now,
1047 updatedAt: now,
1048 streaming: false,
1049 },
1050 ],
1051 session: null,
1052 });
1053
1054 expect(
1055 resolveDraftPromotionNavigationTarget({
1056 serverThreadRef,
1057 serverThread,
1058 backgroundSubmissionPending: false,
1059 }),
1060 ).toEqual(serverThreadRef);
1061 });
1062
1046 it("promotes when the provider starts the first turn", () => {1063 it("promotes when the provider starts the first turn", () => {
1047 const latestTurn = { ...completedTurn, state: "running" as const, completedAt: null };1064 const latestTurn = { ...completedTurn, state: "running" as const, completedAt: null };
10481065
...@@ -2472,3 +2489,133 @@ describe("restorePlanFollowUpComposer", () => {...@@ -2472,3 +2489,133 @@ describe("restorePlanFollowUpComposer", () => {
2472 });2489 });
2473 });2490 });
2474});2491});
2492
2493describe("worktree setup visibility", () => {
2494 const stage = (
2495 id: "fetch" | "checkout" | "submodules" | "setup-script" | "agent",
2496 status: "done" | "running" | "failed" | "pending",
2497 ) => ({
2498 id,
2499 status,
2500 startedAt: now,
2501 endedAt: status === "running" || status === "pending" ? null : now,
2502 percent: null,
2503 detail: null,
2504 tail: [],
2505 });
2506 const base = {
2507 threadId,
2508 phase: "running" as const,
2509 startedAt: now,
2510 endedAt: null,
2511 branch: "feature",
2512 baseRef: "main",
2513 worktreePath: null,
2514 setupScript: null,
2515 stages: [stage("checkout", "running"), stage("agent", "pending")],
2516 error: null,
2517 sequence: 1,
2518 };
2519 const settledDone = {
2520 ...base,
2521 phase: "done" as const,
2522 endedAt: now,
2523 stages: [stage("checkout", "done"), stage("setup-script", "done"), stage("agent", "done")],
2524 };
2525
2526 it("reads the settled snapshot back from the thread's activities", () => {
2527 const activities = [
2528 { kind: "setup-script.started", payload: {} },
2529 { kind: "worktree-setup", payload: settledDone },
2530 { kind: "worktree-setup", payload: { not: "a snapshot" } },
2531 ];
2532 expect(findRecordedWorktreeSetup(activities, threadId)).toEqual(settledDone);
2533 expect(findRecordedWorktreeSetup(activities, ThreadId.make("other"))).toBeNull();
2534 });
2535
2536 it("shows a running setup and hides a clean one once the turn started", () => {
2537 expect(
2538 resolveVisibleWorktreeSetup({
2539 live: base,
2540 recorded: null,
2541 turnStarted: false,
2542 isWorking: true,
2543 }),
2544 ).toEqual(base);
2545 expect(
2546 resolveVisibleWorktreeSetup({
2547 live: null,
2548 recorded: settledDone,
2549 turnStarted: false,
2550 isWorking: true,
2551 }),
2552 ).toEqual(settledDone);
2553 expect(
2554 resolveVisibleWorktreeSetup({
2555 live: null,
2556 recorded: settledDone,
2557 turnStarted: true,
2558 isWorking: true,
2559 }),
2560 ).toBeNull();
2561 });
2562
2563 it("keeps a failed script visible for the running turn and a failed setup always", () => {
2564 const scriptFailed = {
2565 ...settledDone,
2566 stages: [stage("checkout", "done"), stage("setup-script", "failed"), stage("agent", "done")],
2567 };
2568 expect(
2569 resolveVisibleWorktreeSetup({
2570 live: null,
2571 recorded: scriptFailed,
2572 turnStarted: true,
2573 isWorking: true,
2574 }),
2575 ).toEqual(scriptFailed);
2576 expect(
2577 resolveVisibleWorktreeSetup({
2578 live: null,
2579 recorded: scriptFailed,
2580 turnStarted: true,
2581 isWorking: false,
2582 }),
2583 ).toBeNull();
2584 const failed = { ...settledDone, phase: "failed" as const, error: "git exploded" };
2585 expect(
2586 resolveVisibleWorktreeSetup({
2587 live: null,
2588 recorded: failed,
2589 turnStarted: true,
2590 isWorking: false,
2591 }),
2592 ).toEqual(failed);
2593 });
2594
2595 it("prefers whichever snapshot is newer by sequence", () => {
2596 expect(
2597 resolveVisibleWorktreeSetup({
2598 live: { ...base, sequence: 3 },
2599 recorded: { ...settledDone, sequence: 7 },
2600 turnStarted: false,
2601 isWorking: false,
2602 }),
2603 ).toEqual({ ...settledDone, sequence: 7 });
2604 expect(
2605 resolveVisibleWorktreeSetup({
2606 live: { ...settledDone, sequence: 9 },
2607 recorded: { ...base, sequence: 1 },
2608 turnStarted: false,
2609 isWorking: false,
2610 }),
2611 ).toEqual({ ...settledDone, sequence: 9 });
2612 expect(
2613 resolveVisibleWorktreeSetup({
2614 live: base,
2615 recorded: null,
2616 turnStarted: false,
2617 isWorking: false,
2618 }),
2619 ).toEqual(base);
2620 });
2621});
apps/web/src/components/ChatView.logic.ts+57-4
...@@ -18,6 +18,8 @@ import {...@@ -18,6 +18,8 @@ import {
18 type ThreadId,18 type ThreadId,
19 type ThreadLinkedPullRequest,19 type ThreadLinkedPullRequest,
20 type TurnId,20 type TurnId,
21 WORKTREE_SETUP_ACTIVITY_KIND,
22 WorktreeSetupSnapshot,
21} from "@t3tools/contracts";23} from "@t3tools/contracts";
22import { parseScopedThreadKey } from "@t3tools/client-runtime/environment";24import { parseScopedThreadKey } from "@t3tools/client-runtime/environment";
23import { resolveAssetUrl } from "@t3tools/client-runtime/state/assets";25import { resolveAssetUrl } from "@t3tools/client-runtime/state/assets";
...@@ -40,6 +42,7 @@ import {...@@ -40,6 +42,7 @@ import {
40 type TurnDiffSummary,42 type TurnDiffSummary,
41} from "../types";43} from "../types";
42import { type ComposerImageAttachment, type DraftThreadState } from "../composerDraftStore";44import { type ComposerImageAttachment, type DraftThreadState } from "../composerDraftStore";
45import * as Option from "effect/Option";
43import * as Schema from "effect/Schema";46import * as Schema from "effect/Schema";
44import { appAtomRegistry } from "../rpc/atomRegistry";47import { appAtomRegistry } from "../rpc/atomRegistry";
45import { environmentThreadDetails } from "../state/threads";48import { environmentThreadDetails } from "../state/threads";
...@@ -251,6 +254,52 @@ export function toolGroupConsumesUpwardNavigation(target: EventTarget | null): b...@@ -251,6 +254,52 @@ export function toolGroupConsumesUpwardNavigation(target: EventTarget | null): b
251 return false;254 return false;
252}255}
253256
257const decodeWorktreeSetupSnapshot = Schema.decodeUnknownOption(WorktreeSetupSnapshot);
258
259/**
260 * The worktree setup the server recorded on the thread, if any: running once
261 * the bootstrap created the thread, then the settled outcome. It is what a
262 * reload or a second client renders, and what tells them to attach the live
263 * stream while it still says running.
264 */
265export function findRecordedWorktreeSetup(
266 activities: ReadonlyArray<{ readonly kind: string; readonly payload: unknown }>,
267 threadId: ThreadId,
268): WorktreeSetupSnapshot | null {
269 for (let index = activities.length - 1; index >= 0; index -= 1) {
270 const activity = activities[index]!;
271 if (activity.kind !== WORKTREE_SETUP_ACTIVITY_KIND) continue;
272 const decoded = decodeWorktreeSetupSnapshot(activity.payload);
273 if (Option.isSome(decoded) && decoded.value.threadId === threadId) return decoded.value;
274 }
275 return null;
276}
277
278/**
279 * Which setup snapshot the timeline shows, if any. The live stream wins while
280 * it has a newer sequence; the recorded activity covers everything else. A
281 * running setup always shows. Once settled, the card stays only while it
282 * still says something the turn does not: the turn has not started yet, or a
283 * stage failed and the turn is still running so the exit code stays reachable.
284 */
285export function resolveVisibleWorktreeSetup(input: {
286 live: WorktreeSetupSnapshot | null;
287 recorded: WorktreeSetupSnapshot | null;
288 turnStarted: boolean;
289 isWorking: boolean;
290}): WorktreeSetupSnapshot | null {
291 const snapshot =
292 input.live && (!input.recorded || input.live.sequence >= input.recorded.sequence)
293 ? input.live
294 : input.recorded;
295 if (!snapshot) return null;
296 if (snapshot.phase === "running") return snapshot;
297 if (snapshot.phase !== "done") return snapshot;
298 if (!input.turnStarted) return snapshot;
299 const stageFailed = snapshot.stages.some((stage) => stage.status === "failed");
300 return stageFailed && input.isWorking ? snapshot : null;
301}
302
254export function resolveDraftHeroState(input: {303export function resolveDraftHeroState(input: {
255 isLocalDraftThread: boolean;304 isLocalDraftThread: boolean;
256 hasTimelineEntries: boolean;305 hasTimelineEntries: boolean;
...@@ -405,7 +454,7 @@ export function resolveThreadSwitchTimeline<T extends readonly unknown[]>(input:...@@ -405,7 +454,7 @@ export function resolveThreadSwitchTimeline<T extends readonly unknown[]>(input:
405454
406export function resolveDraftPromotionNavigationTarget(input: {455export function resolveDraftPromotionNavigationTarget(input: {
407 serverThreadRef: ScopedThreadRef | null;456 serverThreadRef: ScopedThreadRef | null;
408 serverThread: Pick<Thread, "latestTurn" | "session"> | null | undefined;457 serverThread: Pick<Thread, "latestTurn" | "session" | "messages"> | null | undefined;
409 backgroundSubmissionPending: boolean;458 backgroundSubmissionPending: boolean;
410}): ScopedThreadRef | null {459}): ScopedThreadRef | null {
411 if (input.backgroundSubmissionPending) {460 if (input.backgroundSubmissionPending) {
...@@ -415,9 +464,13 @@ export function resolveDraftPromotionNavigationTarget(input: {...@@ -415,9 +464,13 @@ export function resolveDraftPromotionNavigationTarget(input: {
415 const turnStarted = input.serverThread?.latestTurn?.startedAt != null;464 const turnStarted = input.serverThread?.latestTurn?.startedAt != null;
416 const startupStopped =465 const startupStopped =
417 sessionStatus === "error" || sessionStatus === "stopped" || sessionStatus === "interrupted";466 sessionStatus === "error" || sessionStatus === "stopped" || sessionStatus === "interrupted";
418 // Keep local preparation feedback mounted until the server can render the467 // A worktree bootstrap persists the user message before the turn, so the
419 // running turn or its startup error on the canonical thread route.468 // thread route can render the send and the live setup by itself. Otherwise
420 return turnStarted || startupStopped ? input.serverThreadRef : null;469 // keep the draft mounted until the server can render the running turn or
470 // its startup error.
471 const messagePersisted =
472 input.serverThread?.messages.some((message) => message.role === "user") ?? false;
473 return turnStarted || startupStopped || messagePersisted ? input.serverThreadRef : null;
421}474}
422475
423export function scheduleEnvironmentReconnectWarning(showWarning: () => void): () => void {476export function scheduleEnvironmentReconnectWarning(showWarning: () => void): () => void {
apps/web/src/components/ChatView.tsx+60-125
...@@ -435,6 +435,8 @@ import {...@@ -435,6 +435,8 @@ import {
435 resolveComposerInteractionMode,435 resolveComposerInteractionMode,
436 resolveComposerProviderSelection,436 resolveComposerProviderSelection,
437 resolveDraftHeroState,437 resolveDraftHeroState,
438 findRecordedWorktreeSetup,
439 resolveVisibleWorktreeSetup,
438 restorePlanFollowUpComposer,440 restorePlanFollowUpComposer,
439 isPaintOnlyThreadTimeline,441 isPaintOnlyThreadTimeline,
440 peekHeldThreadTimeline,442 peekHeldThreadTimeline,
...@@ -1430,16 +1432,6 @@ function releaseChatTimelineAnchor<T extends { readonly messageId: MessageId | n...@@ -1430,16 +1432,6 @@ function releaseChatTimelineAnchor<T extends { readonly messageId: MessageId | n
1430 return current.messageId === null ? current : { ...current, messageId: null };1432 return current.messageId === null ? current : { ...current, messageId: null };
1431}1433}
14321434
1433/**
1434 * Worktree setups whose async setup script may still be running after the
1435 * draft route hands off to the created thread. Keyed by scoped thread key;
1436 * entries are removed once the setup settles.
1437 */
1438const pendingWorktreeSetupByThreadKey = new Map<
1439 string,
1440 { environmentId: EnvironmentId; threadId: ThreadId }
1441>();
1442
1443export default function ChatView(props: ChatViewProps) {1435export default function ChatView(props: ChatViewProps) {
1444 const {1436 const {
1445 environmentId,1437 environmentId,
...@@ -1662,19 +1654,9 @@ export default function ChatView(props: ChatViewProps) {...@@ -1662,19 +1654,9 @@ export default function ChatView(props: ChatViewProps) {
1662 return () => revokeBlobPreviewUrl(src);1654 return () => revokeBlobPreviewUrl(src);
1663 }, [expandedImage]);1655 }, [expandedImage]);
1664 const [optimisticUserMessages, setOptimisticUserMessages] = useState<ChatMessage[]>([]);1656 const [optimisticUserMessages, setOptimisticUserMessages] = useState<ChatMessage[]>([]);
1665 // The bootstrap worktree setup this composer last dispatched. Set when a1657 // Last live snapshot from the setup stream. The server drops a finished
1666 // worktree send starts and cleared once the turn starts or the next send1658 // snapshot after a grace period and emits null; holding it here bridges the
1667 // begins, so a failed or cancelled card stays until the user acts.1659 // gap until the settled activity arrives on the thread projection.
1668 const [worktreeSetupRef, setWorktreeSetupRef] = useState<{
1669 environmentId: EnvironmentId;
1670 threadId: ThreadId;
1671 ownerKey: string;
1672 } | null>(() => {
1673 // The draft route unmounts when it promotes to the created thread, while an
1674 // async setup script may still be running. Adopt the ref the draft left.
1675 const handed = pendingWorktreeSetupByThreadKey.get(routeThreadKey);
1676 return handed ? { ...handed, ownerKey: routeThreadKey } : null;
1677 });
1678 const [heldWorktreeSetup, setHeldWorktreeSetup] = useState<WorktreeSetupSnapshot | null>(null);1660 const [heldWorktreeSetup, setHeldWorktreeSetup] = useState<WorktreeSetupSnapshot | null>(null);
1679 // Set by "Work locally": the draft whose restored message should be resent1661 // Set by "Work locally": the draft whose restored message should be resent
1680 // once the cancelled dispatch has settled and the draft is in local mode.1662 // once the cancelled dispatch has settled and the draft is in local mode.
...@@ -3157,7 +3139,7 @@ export default function ChatView(props: ChatViewProps) {...@@ -3157,7 +3139,7 @@ export default function ChatView(props: ChatViewProps) {
3157 resetLocalDispatch,3139 resetLocalDispatch,
3158 localDispatchStartedAt,3140 localDispatchStartedAt,
3159 latestUserMessageAt,3141 latestUserMessageAt,
3160 isPreparingWorktree,3142 isPreparingWorktree: isLocallyPreparingWorktree,
3161 isSendBusy,3143 isSendBusy,
3162 backgroundSubmissionPending,3144 backgroundSubmissionPending,
3163 } = useLocalDispatchState({3145 } = useLocalDispatchState({
...@@ -3193,8 +3175,30 @@ export default function ChatView(props: ChatViewProps) {...@@ -3193,8 +3175,30 @@ export default function ChatView(props: ChatViewProps) {
3193 (isSendBusy || phase === "connecting" || phase === "running") &&3175 (isSendBusy || phase === "connecting" || phase === "running") &&
3194 compactRequestIsActive &&3176 compactRequestIsActive &&
3195 !compactionSettled;3177 !compactionSettled;
3178 // The server records a running worktree setup on the thread for the whole
3179 // bootstrap window. That record, with no turn yet, is how a reload or another
3180 // client sees a worktree still being prepared, so it counts as working like
3181 // the local dispatch that started it. It settles on every failure path and
3182 // on restart, so this cannot outlive the setup. The placeholder "starting"
3183 // session is not used here: an ordinary first turn projects one too, and it
3184 // already drives the connecting state on its own.
3185 const recordedWorktreeSetup = useMemo(
3186 () => findRecordedWorktreeSetup(activeThread?.activities ?? [], routeThreadRef.threadId),
3187 [activeThread?.activities, routeThreadRef.threadId],
3188 );
3189 const awaitingBootstrapTurn =
3190 activeServerThread !== null &&
3191 activeServerThread.id === routeThreadRef.threadId &&
3192 activeServerThread.latestTurn === null &&
3193 recordedWorktreeSetup?.phase === "running";
3196 const isWorking =3194 const isWorking =
3197 phase === "running" || isSendBusy || isConnecting || isRevertingCheckpoint || isCompacting;3195 phase === "running" ||
3196 isSendBusy ||
3197 isConnecting ||
3198 isRevertingCheckpoint ||
3199 isCompacting ||
3200 awaitingBootstrapTurn;
3201 const isPreparingWorktree = isLocallyPreparingWorktree || awaitingBootstrapTurn;
3198 const activeWorkStartedAt = deriveActiveWorkStartedAt(3202 const activeWorkStartedAt = deriveActiveWorkStartedAt(
3199 activeLatestTurn,3203 activeLatestTurn,
3200 activeThread?.session ?? null,3204 activeThread?.session ?? null,
...@@ -3487,115 +3491,57 @@ export default function ChatView(props: ChatViewProps) {...@@ -3487,115 +3491,57 @@ export default function ChatView(props: ChatViewProps) {
3487 activeThreadKey,3491 activeThreadKey,
3488 );3492 );
3489 const displayedThreadRef = parseScopedThreadKey(displayedTimelineKey);3493 const displayedThreadRef = parseScopedThreadKey(displayedTimelineKey);
3490 // Live stages of a bootstrap worktree setup. The subscription follows the3494 // Live stages of a bootstrap worktree setup. A worktree send creates the
3491 // thread that was set up, not the route: a deleted bootstrap thread rotates3495 // server thread under the route's thread id before anything else, so the
3492 // the draft's thread id, and the failed card must survive that.3496 // stream is keyed by that id alone: no owner bookkeeping, and a remount,
3493 const worktreeSetupOwnerKey = draftId ?? routeThreadKey;3497 // reload, or second client picks it up the same way. The subscription is
3494 const worktreeSetupActive =3498 // held only while a snapshot can still change.
3495 worktreeSetupRef !== null && worktreeSetupRef.ownerKey === worktreeSetupOwnerKey;3499 const routeThreadPreparesWorktree =
3496 // A thread reopened mid-setup has no dispatch ref: this view remounted.3500 (isPreparingWorktree && activeThread?.id === routeThreadRef.threadId) ||
3497 // The server projects a starting session before any message or turn exists3501 heldWorktreeSetup?.phase === "running";
3498 // for exactly that window, so follow the setup stream from the thread's own
3499 // state. The ref is adopted below so the card then follows the same
3500 // lifecycle as the original send, including an async setup script that
3501 // keeps running after the agent's turn lands.
3502 const resumedWorktreeSetupRef =
3503 !worktreeSetupActive &&
3504 isServerThread &&
3505 activeThreadRef !== null &&
3506 activeThreadShell?.session?.status === "starting" &&
3507 activeThreadShell.latestTurn === null &&
3508 activeThreadShell.latestUserMessageAt === null
3509 ? activeThreadRef
3510 : null;
3511 useEffect(() => {
3512 if (!resumedWorktreeSetupRef) return;
3513 // Only a ref owned by this route survives adoption. This component is
3514 // reused across routes, so a ref left by another thread's send is stale
3515 // here and would leave the resumed setup with no target after handoff.
3516 setWorktreeSetupRef((current) =>
3517 current?.ownerKey === worktreeSetupOwnerKey
3518 ? current
3519 : { ...resumedWorktreeSetupRef, ownerKey: worktreeSetupOwnerKey },
3520 );
3521 }, [resumedWorktreeSetupRef, worktreeSetupOwnerKey]);
3522 // The setup runs on the environment that received the dispatch, so both
3523 // the subscription and cancel target that one even if the draft's machine
3524 // picker changes underneath.
3525 const worktreeSetupTarget = useMemo(
3526 () =>
3527 worktreeSetupActive
3528 ? { environmentId: worktreeSetupRef.environmentId, threadId: worktreeSetupRef.threadId }
3529 : resumedWorktreeSetupRef,
3530 [resumedWorktreeSetupRef, worktreeSetupActive, worktreeSetupRef],
3531 );
3532 const worktreeSetupQuery = useEnvironmentQuery(3502 const worktreeSetupQuery = useEnvironmentQuery(
3533 worktreeSetupTarget3503 routeThreadPreparesWorktree
3534 ? vcsEnvironment.worktreeSetup({3504 ? vcsEnvironment.worktreeSetup({
3535 environmentId: worktreeSetupTarget.environmentId,3505 environmentId: routeThreadRef.environmentId,
3536 input: { threadId: worktreeSetupTarget.threadId },3506 input: { threadId: routeThreadRef.threadId },
3537 })3507 })
3538 : null,3508 : null,
3539 );3509 );
3540 const latestWorktreeSetup = worktreeSetupQuery.data;3510 const latestWorktreeSetup = worktreeSetupQuery.data;
3541 useEffect(() => {3511 useEffect(() => {
3542 // The server drops finished snapshots after a grace period and emits null.
3543 // Hold the last real snapshot so a settled card does not vanish.
3544 if (latestWorktreeSetup) setHeldWorktreeSetup(latestWorktreeSetup);3512 if (latestWorktreeSetup) setHeldWorktreeSetup(latestWorktreeSetup);
3545 }, [latestWorktreeSetup]);3513 }, [latestWorktreeSetup]);
3546 const worktreeSetup =
3547 worktreeSetupTarget !== null && heldWorktreeSetup?.threadId === worktreeSetupTarget.threadId
3548 ? heldWorktreeSetup
3549 : null;
3550 // A finished card is dropped once the agent's turn shows in the timeline:
3551 // the card belongs to the send, and the agent takes over from there. An
3552 // async setup script keeps the snapshot running past the handoff and its
3553 // row leaves the moment the script exits cleanly; a failed script stays
3554 // for the rest of the turn so the exit code and terminal remain reachable.
3555 const worktreeSetupDoneAndTurnVisible =
3556 worktreeSetup?.phase === "done" &&
3557 activeThread?.latestTurn?.startedAt != null &&
3558 (!isWorking || !worktreeSetup.stages.some((stage) => stage.status === "failed"));
3559 useEffect(() => {3514 useEffect(() => {
3560 if (!worktreeSetupDoneAndTurnVisible) return;
3561 setWorktreeSetupRef(null);
3562 setHeldWorktreeSetup(null);3515 setHeldWorktreeSetup(null);
3563 }, [worktreeSetupDoneAndTurnVisible]);3516 }, [routeThreadKey]);
3564 // The handoff entry only matters while the setup is still running: once it3517 const liveWorktreeSetup =
3565 // settles in any phase, a later mount of the thread must not adopt it.3518 heldWorktreeSetup?.threadId === routeThreadRef.threadId ? heldWorktreeSetup : null;
3566 const worktreeSetupSettledKey =3519 const worktreeSetup = resolveVisibleWorktreeSetup({
3567 worktreeSetup && worktreeSetup.phase !== "running" && worktreeSetupRef3520 live: liveWorktreeSetup,
3568 ? scopedThreadKey(scopeThreadRef(worktreeSetupRef.environmentId, worktreeSetupRef.threadId))3521 recorded: recordedWorktreeSetup,
3569 : null;3522 turnStarted: activeThread?.latestTurn?.startedAt != null,
3570 useEffect(() => {3523 isWorking,
3571 if (worktreeSetupSettledKey) pendingWorktreeSetupByThreadKey.delete(worktreeSetupSettledKey);3524 });
3572 }, [worktreeSetupSettledKey]);
3573 // Sends wait for the agent handoff, not for the setup script: an async3525 // Sends wait for the agent handoff, not for the setup script: an async
3574 // script keeps the snapshot running while the agent already works, and a3526 // script keeps the snapshot running while the agent already works, and a
3575 // follow-up must not be held behind a slow install. A resumed setup has no3527 // follow-up must not be held behind a slow install. Before the first
3576 // snapshot until its first stream event, so that gap blocks as well. The3528 // snapshot arrives the starting session stands in for it.
3577 // gap is judged by the shell, not by the resumed ref, since adopting the
3578 // ref clears it before the query has delivered anything.
3579 const worktreeSetupAwaitingFirstSnapshot =
3580 worktreeSetup === null &&
3581 worktreeSetupTarget !== null &&
3582 isServerThread &&
3583 activeThreadShell?.session?.status === "starting" &&
3584 activeThreadShell.latestTurn === null;
3585 const worktreeSetupBlocksSend =3529 const worktreeSetupBlocksSend =
3586 worktreeSetup !== null3530 worktreeSetup !== null
3587 ? worktreeSetup.phase === "running" && !worktreeSetupAgentStarted(worktreeSetup)3531 ? worktreeSetup.phase === "running" && !worktreeSetupAgentStarted(worktreeSetup)
3588 : worktreeSetupAwaitingFirstSnapshot;3532 : isServerThread &&
3533 activeThreadShell?.session?.status === "starting" &&
3534 activeThreadShell.latestTurn === null;
3589 const cancelWorktreeSetup = useAtomCommand(vcsEnvironment.cancelWorktreeSetup, {3535 const cancelWorktreeSetup = useAtomCommand(vcsEnvironment.cancelWorktreeSetup, {
3590 reportFailure: false,3536 reportFailure: false,
3591 });3537 });
3592 const onCancelWorktreeSetup = useCallback(() => {3538 const onCancelWorktreeSetup = useCallback(() => {
3593 if (!worktreeSetup || !worktreeSetupTarget || worktreeSetup.phase !== "running") return;3539 if (!worktreeSetup || worktreeSetup.phase !== "running") return;
3594 void cancelWorktreeSetup({3540 void cancelWorktreeSetup({
3595 environmentId: worktreeSetupTarget.environmentId,3541 environmentId: routeThreadRef.environmentId,
3596 input: { threadId: worktreeSetup.threadId },3542 input: { threadId: worktreeSetup.threadId },
3597 });3543 });
3598 }, [cancelWorktreeSetup, worktreeSetup, worktreeSetupTarget]);3544 }, [cancelWorktreeSetup, routeThreadRef.environmentId, worktreeSetup]);
3599 // The setup terminal belongs to the thread that was set up. A failed3545 // The setup terminal belongs to the thread that was set up. A failed
3600 // bootstrap deletes that thread and closes its terminals, so only offer the3546 // bootstrap deletes that thread and closes its terminals, so only offer the
3601 // terminal while the setup thread is still the active one.3547 // terminal while the setup thread is still the active one.
...@@ -7588,17 +7534,6 @@ export default function ChatView(props: ChatViewProps) {...@@ -7588,17 +7534,6 @@ export default function ChatView(props: ChatViewProps) {
7588 preparingWorktree: Boolean(baseBranchForWorktree),7534 preparingWorktree: Boolean(baseBranchForWorktree),
7589 submissionIntent: resolvedSubmissionIntent,7535 submissionIntent: resolvedSubmissionIntent,
7590 });7536 });
7591 setWorktreeSetupRef(
7592 baseBranchForWorktree
7593 ? { environmentId, threadId: threadIdForSend, ownerKey: worktreeSetupOwnerKey }
7594 : null,
7595 );
7596 if (baseBranchForWorktree) {
7597 pendingWorktreeSetupByThreadKey.set(
7598 scopedThreadKey(scopeThreadRef(environmentId, threadIdForSend)),
7599 { environmentId, threadId: threadIdForSend },
7600 );
7601 }
76027537
7603 const messageIdForSend = newMessageId();7538 const messageIdForSend = newMessageId();
7604 const messageCreatedAt = new Date().toISOString();7539 const messageCreatedAt = new Date().toISOString();
...@@ -8692,11 +8627,11 @@ export default function ChatView(props: ChatViewProps) {...@@ -8692,11 +8627,11 @@ export default function ChatView(props: ChatViewProps) {
8692 // setup (the bootstrap created it), so this keys off the route, not8627 // setup (the bootstrap created it), so this keys off the route, not
8693 // `isLocalDraftThread`.8628 // `isLocalDraftThread`.
8694 const onWorktreeSetupWorkLocally = useCallback(() => {8629 const onWorktreeSetupWorkLocally = useCallback(() => {
8695 if (!worktreeSetup || !worktreeSetupRef || worktreeSetup.phase !== "running" || !draftId) {8630 if (!worktreeSetup || worktreeSetup.phase !== "running" || !draftId) {
8696 return;8631 return;
8697 }8632 }
8698 const target = {8633 const target = {
8699 environmentId: worktreeSetupRef.environmentId,8634 environmentId: routeThreadRef.environmentId,
8700 input: { threadId: worktreeSetup.threadId },8635 input: { threadId: worktreeSetup.threadId },
8701 };8636 };
8702 void (async () => {8637 void (async () => {
...@@ -8704,7 +8639,7 @@ export default function ChatView(props: ChatViewProps) {...@@ -8704,7 +8639,7 @@ export default function ChatView(props: ChatViewProps) {
8704 if (result._tag !== "Success" || !result.value.cancelled) return;8639 if (result._tag !== "Success" || !result.value.cancelled) return;
8705 setWorkLocallyResendDraftId(draftId);8640 setWorkLocallyResendDraftId(draftId);
8706 })();8641 })();
8707 }, [cancelWorktreeSetup, draftId, worktreeSetup, worktreeSetupRef]);8642 }, [cancelWorktreeSetup, draftId, routeThreadRef.environmentId, worktreeSetup]);
8708 const onSendRef = useRef(onSend);8643 const onSendRef = useRef(onSend);
8709 onSendRef.current = onSend;8644 onSendRef.current = onSend;
8710 // Resend once the cancelled dispatch has settled and the composer is free.8645 // Resend once the cancelled dispatch has settled and the composer is free.
apps/web/src/session-logic.ts+6-1
...@@ -468,7 +468,12 @@ export function deriveWorkLogEntries(...@@ -468,7 +468,12 @@ export function deriveWorkLogEntries(
468 }468 }
469 const entries: DerivedWorkLogEntry[] = [];469 const entries: DerivedWorkLogEntry[] = [];
470 for (const activity of foldUserInputActivities(ordered)) {470 for (const activity of foldUserInputActivities(ordered)) {
471 if (activity.tone !== "error" && isWorktreeSetupActivity(activity.kind)) continue;471 if (
472 isWorktreeSetupActivity(activity.kind) &&
473 (activity.tone !== "error" || activity.kind === "worktree-setup")
474 ) {
475 continue;
476 }
472 if (activity.kind === "tool.started") continue;477 if (activity.kind === "tool.started") continue;
473 // Agent task.started rows are CTA seeds: they carry the true spawn turn,478 // Agent task.started rows are CTA seeds: they carry the true spawn turn,
474 // which is the batch key (completions of background subagents arrive479 // which is the batch key (completions of background subagents arrive
packages/client-runtime/src/work-log/presentation.ts+10-1
...@@ -11,8 +11,17 @@ import { resolveMediaSource } from "@t3tools/client-runtime/media-source";...@@ -11,8 +11,17 @@ import { resolveMediaSource } from "@t3tools/client-runtime/media-source";
11import { parseChangeRequestUrl } from "@t3tools/shared/changeRequestUrl";11import { parseChangeRequestUrl } from "@t3tools/shared/changeRequestUrl";
12import { isWorkspaceImagePreviewPath } from "@t3tools/shared/filePreview";12import { isWorkspaceImagePreviewPath } from "@t3tools/shared/filePreview";
1313
14/**
15 * Activities the worktree setup card already represents. The settled record
16 * is rendered by the card on web (and mobile's status row), never as a
17 * worklog entry, so it is hidden from the activity feed even when it failed.
18 */
14export function isWorktreeSetupActivity(kind: string): boolean {19export function isWorktreeSetupActivity(kind: string): boolean {
15 return kind === "setup-script.requested" || kind === "setup-script.started";20 return (
21 kind === "setup-script.requested" ||
22 kind === "setup-script.started" ||
23 kind === "worktree-setup"
24 );
16}25}
1726
18export type WorkLogToolLifecycleStatus = RuntimeItemStatus | "stopped";27export type WorkLogToolLifecycleStatus = RuntimeItemStatus | "stopped";
packages/contracts/src/orchestration.ts+25
...@@ -1440,6 +1440,24 @@ const ThreadHistoryImportCommand = Schema.Struct({...@@ -1440,6 +1440,24 @@ const ThreadHistoryImportCommand = Schema.Struct({
1440 ).check(Schema.isNonEmpty()),1440 ).check(Schema.isNonEmpty()),
1441});1441});
14421442
1443/**
1444 * Persists a user message without starting a turn. Used by worktree bootstraps
1445 * so the send is durable while the worktree is still being prepared; the
1446 * turn that follows references the same message id.
1447 */
1448const ThreadMessageUserAppendCommand = Schema.Struct({
1449 type: Schema.Literal("thread.message.user.append"),
1450 commandId: CommandId,
1451 threadId: ThreadId,
1452 message: Schema.Struct({
1453 messageId: MessageId,
1454 text: Schema.String,
1455 attachments: Schema.Array(ChatAttachment),
1456 context: Schema.optional(OrchestrationMessageContext),
1457 }),
1458 createdAt: IsoDateTime,
1459});
1460
1443const ThreadProposedPlanUpsertCommand = Schema.Struct({1461const ThreadProposedPlanUpsertCommand = Schema.Struct({
1444 type: Schema.Literal("thread.proposed-plan.upsert"),1462 type: Schema.Literal("thread.proposed-plan.upsert"),
1445 commandId: CommandId,1463 commandId: CommandId,
...@@ -1537,6 +1555,7 @@ const InternalOrchestrationCommand = Schema.Union([...@@ -1537,6 +1555,7 @@ const InternalOrchestrationCommand = Schema.Union([
1537 ThreadMessageAssistantDeltaCommand,1555 ThreadMessageAssistantDeltaCommand,
1538 ThreadMessageAssistantCompleteCommand,1556 ThreadMessageAssistantCompleteCommand,
1539 ThreadHistoryImportCommand,1557 ThreadHistoryImportCommand,
1558 ThreadMessageUserAppendCommand,
1540 ThreadProposedPlanUpsertCommand,1559 ThreadProposedPlanUpsertCommand,
1541 ThreadTurnDiffCompleteCommand,1560 ThreadTurnDiffCompleteCommand,
1542 ThreadActivityAppendCommand,1561 ThreadActivityAppendCommand,
...@@ -1878,6 +1897,12 @@ export const OrchestrationEventMetadata = Schema.Struct({...@@ -1878,6 +1897,12 @@ export const OrchestrationEventMetadata = Schema.Struct({
1878 requestId: Schema.optional(ApprovalRequestId),1897 requestId: Schema.optional(ApprovalRequestId),
1879 ingestedAt: Schema.optional(IsoDateTime),1898 ingestedAt: Schema.optional(IsoDateTime),
1880 historyImport: Schema.optional(Schema.Boolean),1899 historyImport: Schema.optional(Schema.Boolean),
1900 /**
1901 * The user message was persisted ahead of its turn (worktree bootstrap).
1902 * Reactors that key off a user message as "turn is starting" wait for the
1903 * turn-start event instead.
1904 */
1905 deferredTurn: Schema.optional(Schema.Boolean),
1881 origin: Schema.optional(OrchestrationClientOrigin),1906 origin: Schema.optional(OrchestrationClientOrigin),
1882});1907});
1883export type OrchestrationEventMetadata = typeof OrchestrationEventMetadata.Type;1908export type OrchestrationEventMetadata = typeof OrchestrationEventMetadata.Type;
packages/contracts/src/worktreeSetup.ts+11
...@@ -71,6 +71,17 @@ export const WorktreeSetupSnapshot = Schema.Struct({...@@ -71,6 +71,17 @@ export const WorktreeSetupSnapshot = Schema.Struct({
71});71});
72export type WorktreeSetupSnapshot = typeof WorktreeSetupSnapshot.Type;72export type WorktreeSetupSnapshot = typeof WorktreeSetupSnapshot.Type;
7373
74/**
75 * Thread activity that carries a `WorktreeSetupSnapshot` as its payload. The
76 * bootstrap writes it under a fixed id once the thread exists (phase running)
77 * and again when the setup settles, so the projection always holds the
78 * latest known state: a client attaches the live stream while it says
79 * running and renders the outcome from it afterwards, on any device or
80 * after a reload.
81 */
82export const WORKTREE_SETUP_ACTIVITY_KIND = "worktree-setup";
83export const worktreeSetupActivityId = (threadId: ThreadId) => `worktree-setup:${threadId}`;
84
74export const WorktreeSetupSubscribeInput = Schema.Struct({85export const WorktreeSetupSubscribeInput = Schema.Struct({
75 threadId: ThreadId,86 threadId: ThreadId,
76});87});