Open History to see the last commit
1import { assert, describe, it } from "@effect/vitest";
2import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
3import {
4 EnvironmentId,
5 EventId,
6 MessageId,
7 NodeId,
8 type OrchestrationV2DomainEvent,
9 type OrchestrationV2ThreadShell,
10 type OrchestrationV2TurnItem,
11 ProjectId,
12 ProviderInstanceId,
13 RunId,
14 RuntimeRequestId,
15 ThreadId,
16 TurnItemId,
17} from "@t3tools/contracts";
18import { RelayAgentActivityState } from "@t3tools/contracts/relay";
19import * as DateTime from "effect/DateTime";
20import * as Deferred from "effect/Deferred";
21import * as Effect from "effect/Effect";
22import * as Fiber from "effect/Fiber";
23import * as Layer from "effect/Layer";
24import * as Option from "effect/Option";
25import * as Queue from "effect/Queue";
26import * as Ref from "effect/Ref";
27import * as Schema from "effect/Schema";
28import * as Stream from "effect/Stream";
29import * as TestClock from "effect/testing/TestClock";
30import * as FetchHttpClient from "effect/http/FetchHttpClient";
31
32import * as EnvironmentKeys from "../cloud/environmentKeys.ts";
33import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
34import {
35 PUBLISH_AGENT_ACTIVITY_SECRET,
36 RELAY_ENVIRONMENT_CREDENTIAL_SECRET,
37 RELAY_ISSUER_SECRET,
38 RELAY_URL_SECRET,
39} from "../cloud/config.ts";
40import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
41import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
42import * as SqlitePersistence from "../persistence/Sqlite.ts";
43import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
44import * as ProjectService from "../project/ProjectService.ts";
45import * as AgentAwarenessRelay from "./AgentAwarenessRelay.ts";
46
47const THREAD_ID = ThreadId.make("relay-thread");
48const SECOND_THREAD_ID = ThreadId.make("relay-thread-2");
49const PROJECT_ID = ProjectId.make("relay-project");
50const NOW = "2026-09-04T12:00:00.000Z";
51
52function shell(overrides: Partial<OrchestrationV2ThreadShell> = {}): OrchestrationV2ThreadShell {
53 return {
54 id: THREAD_ID,
55 projectId: PROJECT_ID,
56 title: "Thread",
57 providerInstanceId: ProviderInstanceId.make("codex"),
58 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "test-model" },
59 runtimeMode: "full-access",
60 interactionMode: "default",
61 worktreePath: null,
62 activeProviderThreadId: null,
63 lineage: { rootThreadId: THREAD_ID, parentThreadId: null, relationshipToParent: null },
64 forkedFrom: null,
65 createdBy: "user",
66 creationSource: "web",
67 activeRunId: null,
68 latestVisibleMessage: null,
69 hasActionableProposedPlan: false,
70 itemCount: 0,
71 visibleItemCount: 0,
72 lastVisitedAt: null,
73 deletedAt: null,
74 branch: null,
75 linkedPullRequest: null,
76 status: "running",
77 activityRunStatus: null,
78 pendingRuntimeRequest: null,
79 pendingBackgroundTasks: [],
80 latestRunId: null,
81 latestRunRequestedAt: null,
82 latestRunStartedAt: null,
83 latestRunCompletedAt: null,
84 latestUserMessageAt: null,
85 createdAt: DateTime.makeUnsafe(NOW),
86 updatedAt: DateTime.makeUnsafe(NOW),
87 archivedAt: null,
88 settledOverride: null,
89 settledAt: null,
90 snoozedUntil: null,
91 snoozedAt: null,
92 pinnedAt: null,
93 ...overrides,
94 };
95}
96
97const CHILD_THREAD_ID = ThreadId.make("relay-child-thread");
98
99function waitingChild(
100 overrides: Partial<OrchestrationV2ThreadShell> = {},
101): OrchestrationV2ThreadShell {
102 return shell({
103 id: CHILD_THREAD_ID,
104 title: "Child",
105 lineage: {
106 rootThreadId: THREAD_ID,
107 parentThreadId: THREAD_ID,
108 relationshipToParent: "subagent",
109 },
110 createdBy: "agent",
111 creationSource: "mcp",
112 activityRunStatus: "running",
113 pendingRuntimeRequest: {
114 id: RuntimeRequestId.make("child-request"),
115 kind: "command",
116 createdAt: DateTime.makeUnsafe(NOW),
117 },
118 ...overrides,
119 });
120}
121
122const PublishPayload = Schema.Struct({ state: Schema.NullOr(RelayAgentActivityState) });
123const decodePublishPayload = Schema.decodeUnknownSync(Schema.fromJsonString(PublishPayload));
124const unused = () => Effect.die("Unexpected test dependency call");
125
126describe("startup agent activity", () => {
127 it("publishes active work and only terminal runs completed after startup", () => {
128 const startedAt = DateTime.toEpochMillis(DateTime.makeUnsafe(NOW));
129 const oldCompleted = ThreadId.make("old-completed");
130 const newCompleted = ThreadId.make("new-completed");
131 const oldFailed = ThreadId.make("old-failed");
132 const newFailed = ThreadId.make("new-failed");
133 const ids = AgentAwarenessRelay.resolveAgentAwarenessRelayActiveThreadIds({
134 environmentId: EnvironmentId.make("relay-env"),
135 startedAt,
136 projects: [{ id: PROJECT_ID, title: "Project" }],
137 threads: [
138 shell(),
139 shell({
140 id: oldCompleted,
141 status: "completed",
142 latestRunCompletedAt: DateTime.makeUnsafe("2026-09-04T11:59:00.000Z"),
143 }),
144 shell({
145 id: newCompleted,
146 status: "completed",
147 latestRunCompletedAt: DateTime.makeUnsafe("2026-09-04T12:00:01.000Z"),
148 }),
149 shell({
150 id: oldFailed,
151 status: "failed",
152 latestRunCompletedAt: DateTime.makeUnsafe("2026-09-04T11:59:00.000Z"),
153 }),
154 shell({
155 id: newFailed,
156 status: "failed",
157 latestRunCompletedAt: DateTime.makeUnsafe("2026-09-04T12:00:01.000Z"),
158 }),
159 shell({ id: ThreadId.make("idle"), status: "idle" }),
160 shell({ id: ThreadId.make("missing-project"), projectId: ProjectId.make("missing") }),
161 ],
162 });
163 assert.deepStrictEqual(ids, [THREAD_ID, newCompleted, newFailed]);
164 });
165});
166
167const makeTestRelay = Effect.fnUntraced(function* (
168 options: {
169 readonly respond?: (attempt: number) => Response;
170 readonly failSecretRead?: (name: string) => boolean;
171 readonly linked?: boolean;
172 readonly readShell?: (threadId: ThreadId) => Effect.Effect<OrchestrationV2ThreadShell | null>;
173 readonly domainEvents?: Stream.Stream<OrchestrationV2DomainEvent>;
174 readonly catchUpThreads?: ReadonlyArray<OrchestrationV2ThreadShell>;
175 } = {},
176) {
177 const values = new Map<string, Uint8Array>(
178 options.linked === false
179 ? []
180 : [
181 [PUBLISH_AGENT_ACTIVITY_SECRET, new TextEncoder().encode("true")],
182 [RELAY_URL_SECRET, new TextEncoder().encode("https://relay.example.test")],
183 [RELAY_ISSUER_SECRET, new TextEncoder().encode("https://relay.example.test")],
184 [RELAY_ENVIRONMENT_CREDENTIAL_SECRET, new TextEncoder().encode("credential-1")],
185 ],
186 );
187 const secretReads: string[] = [];
188 const secrets = ServerSecretStore.ServerSecretStore.of({
189 get: (name) =>
190 Effect.suspend(() => {
191 secretReads.push(name);
192 if (options.failSecretRead?.(name)) {
193 return Effect.fail(
194 new ServerSecretStore.SecretStoreReadError({
195 resource: name,
196 cause: "temporary read failure",
197 }),
198 );
199 }
200 return Effect.succeed(Option.fromUndefinedOr(values.get(name)));
201 }),
202 create: (name, value) => Effect.sync(() => void values.set(name, value)),
203 set: (name, value) => Effect.sync(() => void values.set(name, value)),
204 remove: (name) => Effect.sync(() => void values.delete(name)),
205 getOrCreateRandom: unused,
206 });
207 const currentShell = yield* Ref.make<OrchestrationV2ThreadShell | null>(shell());
208 const shellReads: ThreadId[] = [];
209 const catchUp = { activitySnapshotReads: 0 };
210 const threads = ThreadManagementService.ThreadManagementService.of({
211 getThreadShell: unused,
212 getShellSnapshot: unused,
213 getAgentAwarenessSnapshot: (threadId) =>
214 threadId === undefined
215 ? Effect.sync(() => {
216 catchUp.activitySnapshotReads += 1;
217 return { snapshotSequence: 1, threads: options.catchUpThreads ?? [] };
218 })
219 : Effect.sync(() => shellReads.push(threadId)).pipe(
220 Effect.andThen(options.readShell?.(threadId) ?? Ref.get(currentShell)),
221 Effect.map((thread) => ({
222 snapshotSequence: 1,
223 threads: thread === null ? [] : [thread],
224 })),
225 ),
226 listActiveSubagentThreadIds: unused,
227 listAppOwnedChildThreadIds: unused,
228 countThreads: unused,
229 listActiveShellIndex: () => Effect.succeed([]),
230 listOlderSettledThreads: unused,
231 readShellSnapshot: unused,
232 getThreadTitles: unused,
233 getThreadLinkMetadata: unused,
234 ensureLegacyTranscript: unused,
235 dispatch: unused,
236 searchThread: () => Effect.die("unused"),
237 searchThreadStream: () => Stream.empty,
238 getThreadHistoryPage: () => Effect.die("unused"),
239 getTimelinePage: () => Effect.die("Unused timeline read"),
240 getMessageCount: () => Effect.die("unused message count"),
241 getMessagesById: () => Effect.die("unused messages by id"),
242 getTurnItem: () => Effect.die("unused turn item read"),
243 getThreadRecords: () => Effect.die("unused record read"),
244 getThreadProjection: unused,
245 getCheckpointContext: unused,
246 getThreadSnapshot: unused,
247 getThreadTranscript: unused,
248 getThreadSnapshotWindow: unused,
249 getProjectThreadRecords: () => Effect.die("unused project record read"),
250 listProjectThreads: unused,
251 sendToThread: unused,
252 assertProviderInstanceEnabled: unused,
253 isProviderInstanceEnabledForProject: unused,
254 hasReceipt: unused,
255 waitForThread: unused,
256 settleAfterRun: unused,
257 settleThread: unused,
258 interruptThread: unused,
259 stopWorkflow: () => Effect.die("unused stopWorkflow"),
260 forceStopThread: () => Effect.die("unused forceStopThread"),
261 awaitBackgroundTasks: unused,
262 stopDelegatedTasks: unused,
263 getThreadEventSequence: unused,
264 recoverDelegatedTask: unused,
265 cancelDelegatedTask: unused,
266 delegatedTaskResultPending: unused,
267 streamStoredEvents: Stream.empty,
268 streamStoredEventsFrom: () => Stream.empty,
269 streamDomainEvents: options.domainEvents ?? Stream.empty,
270 });
271 const publications: Array<{
272 readonly url: string;
273 readonly authorization: string | null;
274 readonly state: RelayAgentActivityState | null;
275 }> = [];
276 const fetch: typeof globalThis.fetch = Object.assign(
277 (
278 input: Parameters<typeof globalThis.fetch>[0],
279 init?: Parameters<typeof globalThis.fetch>[1],
280 ) => {
281 const body = init?.body;
282 if (typeof body !== "string" && !(body instanceof Uint8Array)) {
283 return Promise.reject(new Error("Expected a serialized activity publish payload"));
284 }
285 const payload = decodePublishPayload(
286 typeof body === "string" ? body : new TextDecoder().decode(body),
287 );
288 publications.push({
289 url: String(input),
290 authorization: new Headers(init?.headers).get("authorization"),
291 state: payload.state,
292 });
293 return Promise.resolve(
294 options.respond?.(publications.length) ?? Response.json({ ok: true, deliveries: [] }),
295 );
296 },
297 { preconnect: () => {} },
298 );
299 const project = {
300 id: PROJECT_ID,
301 title: "Project",
302 workspaceRoot: "/workspace",
303 defaultModelSelection: null,
304 scripts: [],
305 createdAt: NOW,
306 updatedAt: NOW,
307 deletedAt: null,
308 };
309 const relay = yield* AgentAwarenessRelay.make.pipe(
310 Effect.provide(EnvironmentKeys.layer),
311 Effect.provideService(ServerSecretStore.ServerSecretStore, secrets),
312 Effect.provideService(ThreadManagementService.ThreadManagementService, threads),
313 Effect.provideService(ServerEnvironment.ServerEnvironment, {
314 getEnvironmentId: Effect.succeed(EnvironmentId.make("relay-environment")),
315 getDescriptor: unused(),
316 }),
317 Effect.provideService(ProjectService.ProjectService, {
318 create: unused,
319 bootstrap: unused,
320 update: unused,
321 delete: unused,
322 getByWorkspaceRoot: unused,
323 workspaceScanDirectory: unused,
324 snapshot: Effect.succeed({ projects: [project], updatedAt: NOW }),
325 listPage: unused,
326 getShell: unused,
327 listShells: unused,
328 getById: () => Effect.succeedSome(project),
329 }),
330 Effect.provideService(FetchHttpClient.Fetch, fetch),
331 Effect.provide(NodeCrypto.layer),
332 );
333 return { relay, secrets, secretReads, currentShell, shellReads, publications, catchUp };
334});
335
336describe("AgentAwarenessRelay", () => {
337 it.effect("publishes the parent as needing approval while its subagent waits", () =>
338 Effect.gen(function* () {
339 const shells = yield* Ref.make(
340 new Map<ThreadId, OrchestrationV2ThreadShell>([
341 [THREAD_ID, shell({ status: "running" })],
342 [CHILD_THREAD_ID, waitingChild({ pendingRuntimeRequest: null })],
343 ]),
344 );
345 const setShell = (next: OrchestrationV2ThreadShell) =>
346 Ref.update(shells, (current) => new Map(current).set(next.id, next));
347 const { relay, publications } = yield* makeTestRelay({
348 readShell: (threadId) =>
349 Ref.get(shells).pipe(Effect.map((current) => current.get(threadId) ?? null)),
350 });
351 yield* relay.publishThread(THREAD_ID);
352 yield* setShell(shell({ status: "completed" }));
353 yield* setShell(waitingChild());
354 yield* relay.publishThread(CHILD_THREAD_ID);
355 assert.equal(publications.length, 2);
356 assert.isTrue(publications[1]?.url.includes(`/threads/${THREAD_ID}/`) ?? false);
357 assert.equal(publications[1]?.state?.phase, "waiting_for_approval");
358
359 // A rename keeps the same request's alert time, so the phone doesn't re-alert.
360 yield* setShell(
361 shell({
362 status: "completed",
363 title: "Renamed thread",
364 updatedAt: DateTime.makeUnsafe("2026-09-04T12:01:00.000Z"),
365 }),
366 );
367 yield* relay.publishThread(THREAD_ID);
368 assert.equal(publications.length, 3);
369 assert.equal(publications[2]?.state?.threadTitle, "Renamed thread");
370 assert.equal(publications[2]?.state?.updatedAt, publications[1]?.state?.updatedAt);
371
372 yield* setShell(waitingChild({ pendingRuntimeRequest: null }));
373 yield* relay.publishThread(CHILD_THREAD_ID);
374 yield* TestClock.adjust("5 seconds");
375 yield* relay.drain;
376 assert.equal(publications.length, 4);
377 assert.equal(publications[3]?.state?.phase, "completed");
378 }),
379 );
380
381 it.effect("publishes the parent as needing input when its subagent raises a secret card", () =>
382 Effect.gen(function* () {
383 const now = DateTime.makeUnsafe(NOW);
384 const child = waitingChild({
385 pendingRuntimeRequest: {
386 id: RuntimeRequestId.make("child-secret-card"),
387 kind: "user_input",
388 createdAt: now,
389 blocking: true,
390 userOnly: true,
391 },
392 });
393 const parent = shell({ status: "completed" });
394 const events = yield* Queue.unbounded<OrchestrationV2DomainEvent>();
395 const pulls = yield* Queue.unbounded<void>();
396 const { relay, publications } = yield* makeTestRelay({
397 readShell: (threadId) => Effect.succeed(threadId === child.id ? child : parent),
398 domainEvents: Stream.fromEffectRepeat(
399 Queue.offer(pulls, undefined).pipe(Effect.andThen(Queue.take(events))),
400 ),
401 });
402 yield* relay.start();
403 yield* Queue.take(pulls);
404 const runId = RunId.make("run:child-secret-card");
405 // Recording a secret card emits only this event for the child.
406 yield* Queue.offer(events, {
407 id: EventId.make("event:child-secret-card"),
408 type: "turn-item.updated",
409 threadId: CHILD_THREAD_ID,
410 runId,
411 occurredAt: now,
412 payload: {
413 id: TurnItemId.make("child-secret-card"),
414 threadId: CHILD_THREAD_ID,
415 runId,
416 nodeId: NodeId.make("node:child-secret-card"),
417 providerThreadId: null,
418 providerTurnId: null,
419 nativeItemRef: null,
420 parentItemId: null,
421 ordinal: 1,
422 status: "waiting",
423 title: "SMS code",
424 startedAt: now,
425 completedAt: null,
426 updatedAt: now,
427 type: "secret_request",
428 label: "SMS code",
429 reason: "The agent needs the code.",
430 secretStatus: "pending",
431 },
432 });
433 yield* Queue.take(pulls);
434 yield* relay.drain;
435 assert.equal(publications.length, 1);
436 assert.isTrue(publications[0]?.url.includes(`/threads/${THREAD_ID}/`) ?? false);
437 assert.equal(publications[0]?.state?.phase, "waiting_for_input");
438 }),
439 );
440
441 it.effect(
442 "clears the parent's input wait when the user steers its subagent past a link card",
443 () =>
444 Effect.gen(function* () {
445 const now = DateTime.makeUnsafe(NOW);
446 let child = waitingChild({
447 pendingRuntimeRequest: {
448 id: RuntimeRequestId.make("child-link-card"),
449 kind: "user_input",
450 createdAt: now,
451 blocking: true,
452 userOnly: true,
453 },
454 });
455 const parent = shell({ status: "completed" });
456 const events = yield* Queue.unbounded<OrchestrationV2DomainEvent>();
457 const pulls = yield* Queue.unbounded<void>();
458 const { relay, publications } = yield* makeTestRelay({
459 readShell: (threadId) => Effect.succeed(threadId === child.id ? child : parent),
460 domainEvents: Stream.fromEffectRepeat(
461 Queue.offer(pulls, undefined).pipe(Effect.andThen(Queue.take(events))),
462 ),
463 });
464 yield* relay.start();
465 yield* Queue.take(pulls);
466 const runId = RunId.make("run:child-link-card");
467 const item = {
468 threadId: CHILD_THREAD_ID,
469 runId,
470 nodeId: NodeId.make("node:child-link-card"),
471 providerThreadId: null,
472 providerTurnId: null,
473 nativeItemRef: null,
474 parentItemId: null,
475 title: null,
476 startedAt: now,
477 updatedAt: now,
478 } as const;
479 yield* Queue.offer(events, {
480 id: EventId.make("event:child-link-card"),
481 type: "turn-item.updated",
482 threadId: CHILD_THREAD_ID,
483 runId,
484 occurredAt: now,
485 payload: {
486 ...item,
487 id: TurnItemId.make("child-link-card"),
488 ordinal: 1,
489 status: "waiting",
490 completedAt: null,
491 type: "link_request",
492 url: "https://example.com/login",
493 label: "Sign in",
494 reason: "The agent needs you to sign in.",
495 linkStatus: "pending",
496 },
497 });
498 yield* Queue.take(pulls);
499 yield* relay.drain;
500 assert.equal(publications[0]?.state?.phase, "waiting_for_input");
501
502 // Steering a running child records only these events for it; the
503 // projection then stops counting the link card as pending.
504 child = waitingChild({ pendingRuntimeRequest: null });
505 yield* Queue.offer(events, {
506 id: EventId.make("event:child-steer"),
507 type: "turn-item.updated",
508 threadId: CHILD_THREAD_ID,
509 runId,
510 occurredAt: now,
511 payload: {
512 ...item,
513 id: TurnItemId.make("child-steer"),
514 ordinal: 2,
515 status: "completed",
516 completedAt: now,
517 createdBy: "user",
518 creationSource: "web",
519 type: "user_message",
520 messageId: MessageId.make("message:child-steer"),
521 inputIntent: "steer",
522 text: "Skip the login.",
523 attachments: [],
524 },
525 });
526 yield* Queue.take(pulls);
527 // The relay confirms a move into completed before publishing it.
528 yield* TestClock.adjust("5 seconds");
529 yield* relay.drain;
530 assert.equal(publications.length, 2);
531 assert.equal(publications[1]?.state?.phase, "completed");
532 }),
533 );
534
535 it.effect("republishes a completed parent for its waiting child after a relink", () =>
536 Effect.gen(function* () {
537 const parent = shell({ status: "completed" });
538 const child = waitingChild();
539 const { relay, secrets, publications } = yield* makeTestRelay({
540 catchUpThreads: [parent, child],
541 readShell: (id) => Effect.succeed(id === child.id ? child : parent),
542 });
543 yield* relay.start();
544 yield* TestClock.adjust("1 second");
545 yield* relay.drain;
546 assert.equal(publications.length, 1);
547 assert.equal(publications[0]?.state?.phase, "waiting_for_approval");
548
549 yield* secrets.set(
550 RELAY_ENVIRONMENT_CREDENTIAL_SECRET,
551 new TextEncoder().encode("credential-2"),
552 );
553 yield* relay.requestCatchUp();
554 yield* TestClock.adjust("1 second");
555 yield* relay.drain;
556 assert.equal(publications.length, 2);
557 assert.equal(publications[1]?.state?.phase, "waiting_for_approval");
558 }),
559 );
560
561 it.effect("keeps a child's alert time when the parent's own request comes and goes", () =>
562 Effect.gen(function* () {
563 const shells = yield* Ref.make(
564 new Map<ThreadId, OrchestrationV2ThreadShell>([
565 [THREAD_ID, shell()],
566 [CHILD_THREAD_ID, waitingChild()],
567 ]),
568 );
569 const setParent = (next: OrchestrationV2ThreadShell) =>
570 Ref.update(shells, (current) => new Map(current).set(THREAD_ID, next));
571 const { relay, publications } = yield* makeTestRelay({
572 readShell: (threadId) =>
573 Ref.get(shells).pipe(Effect.map((current) => current.get(threadId) ?? null)),
574 });
575 yield* relay.publishThread(CHILD_THREAD_ID);
576 yield* setParent(
577 shell({
578 updatedAt: DateTime.makeUnsafe("2026-09-04T12:01:00.000Z"),
579 pendingRuntimeRequest: {
580 id: RuntimeRequestId.make("parent-request"),
581 kind: "command",
582 createdAt: DateTime.makeUnsafe("2026-09-04T12:01:00.000Z"),
583 },
584 }),
585 );
586 yield* relay.publishThread(THREAD_ID);
587 yield* setParent(shell({ updatedAt: DateTime.makeUnsafe("2026-09-04T12:02:00.000Z") }));
588 yield* relay.publishThread(THREAD_ID);
589 assert.deepStrictEqual(
590 publications.map(({ state }) => state?.phase),
591 ["waiting_for_approval", "waiting_for_approval", "waiting_for_approval"],
592 );
593 assert.equal(publications[2]?.state?.updatedAt, publications[0]?.state?.updatedAt);
594 }),
595 );
596
597 it.effect("coalesces queued updates and reruns a thread dirtied during publishing", () =>
598 Effect.gen(function* () {
599 const started = yield* Deferred.make<void>();
600 const release = yield* Deferred.make<void>();
601 const rerunStarted = yield* Deferred.make<void>();
602 const releaseRerun = yield* Deferred.make<void>();
603 const processed: Array<{ threadId: ThreadId; revision: number }> = [];
604 let revision = 1;
605 const worker = yield* AgentAwarenessRelay.makeAgentAwarenessPublishWorker((threadId) =>
606 Effect.gen(function* () {
607 const currentRevision = revision;
608 const index = processed.length;
609 processed.push({ threadId, revision: currentRevision });
610 if (index === 0) {
611 yield* Deferred.succeed(started, undefined);
612 yield* Deferred.await(release);
613 } else if (threadId === THREAD_ID) {
614 yield* Deferred.succeed(rerunStarted, undefined);
615 yield* Deferred.await(releaseRerun);
616 }
617 }),
618 );
619 yield* worker.enqueue(THREAD_ID);
620 yield* Deferred.await(started);
621 revision = 2;
622 for (let i = 0; i < 100; i++) {
623 yield* worker.enqueue(SECOND_THREAD_ID);
624 yield* worker.enqueue(THREAD_ID);
625 }
626 const drained = yield* Deferred.make<void>();
627 const draining = yield* worker.drain.pipe(
628 Effect.andThen(Deferred.succeed(drained, undefined)),
629 Effect.forkChild,
630 );
631 yield* Deferred.succeed(release, undefined);
632 yield* Deferred.await(rerunStarted);
633 assert.isFalse(yield* Deferred.isDone(drained));
634 yield* Deferred.succeed(releaseRerun, undefined);
635 yield* Fiber.join(draining);
636 assert.deepEqual(processed, [
637 { threadId: THREAD_ID, revision: 1 },
638 { threadId: SECOND_THREAD_ID, revision: 2 },
639 { threadId: THREAD_ID, revision: 2 },
640 ]);
641 }),
642 );
643
644 it.effect("deduplicates state and republishes title changes", () =>
645 Effect.gen(function* () {
646 const { relay, currentShell, publications } = yield* makeTestRelay();
647 yield* relay.publishThread(THREAD_ID);
648 yield* Ref.set(
649 currentShell,
650 shell({ updatedAt: DateTime.makeUnsafe("2026-09-04T13:00:00Z") }),
651 );
652 yield* relay.publishThread(THREAD_ID);
653 assert.equal(publications.length, 1);
654 yield* Ref.set(currentShell, shell({ title: "Renamed thread" }));
655 yield* relay.publishThread(THREAD_ID);
656 assert.equal(publications.length, 2);
657 assert.equal(publications[1]?.state?.threadTitle, "Renamed thread");
658 }),
659 );
660
661 it.effect("republishes unchanged state with fresh credentials after relinking", () =>
662 Effect.gen(function* () {
663 const { relay, secrets, publications } = yield* makeTestRelay();
664 yield* relay.publishThread(THREAD_ID);
665 yield* secrets.set(
666 RELAY_ENVIRONMENT_CREDENTIAL_SECRET,
667 new TextEncoder().encode("credential-2"),
668 );
669 yield* relay.publishThread(THREAD_ID);
670 assert.equal(publications.length, 2);
671 assert.equal(publications[0]?.authorization, "Bearer credential-1");
672 assert.equal(publications[1]?.authorization, "Bearer credential-2");
673 yield* secrets.set(
674 RELAY_URL_SECRET,
675 new TextEncoder().encode("https://new-relay.example.test"),
676 );
677 yield* relay.publishThread(THREAD_ID);
678 assert.equal(publications.length, 3);
679 assert.isTrue(publications[2]?.url.startsWith("https://new-relay.example.test/") ?? false);
680 }),
681 );
682
683 it.effect.each(["completed", "failed"] as const)(
684 "publishes a tombstone when a %s thread is settled through a domain event",
685 (status) =>
686 Effect.gen(function* () {
687 const events = yield* Queue.unbounded<OrchestrationV2DomainEvent>();
688 const pulls = yield* Queue.unbounded<void>();
689 const { relay, currentShell, publications } = yield* makeTestRelay({
690 domainEvents: Stream.fromEffectRepeat(
691 Queue.offer(pulls, undefined).pipe(Effect.andThen(Queue.take(events))),
692 ),
693 });
694 yield* relay.start();
695 yield* Queue.take(pulls);
696 yield* relay.publishThread(THREAD_ID);
697 const terminal = shell({ status });
698 yield* Ref.set(currentShell, terminal);
699 yield* relay.publishThread(THREAD_ID);
700 yield* TestClock.adjust("5 seconds");
701 yield* relay.drain;
702 assert.equal(publications[1]?.state?.phase, status);
703
704 const now = yield* DateTime.now;
705 const settled = {
706 ...terminal,
707 settledOverride: "settled" as const,
708 settledAt: now,
709 lastVisitedAt: null,
710 };
711 yield* Ref.set(currentShell, settled);
712 yield* Queue.offer(events, {
713 id: EventId.make("event:settled"),
714 type: "thread.settled",
715 threadId: THREAD_ID,
716 occurredAt: now,
717 payload: settled,
718 });
719 yield* Queue.take(pulls);
720 yield* relay.drain;
721 assert.equal(publications.length, 2);
722 yield* TestClock.adjust("5 seconds");
723 yield* relay.drain;
724 assert.equal(publications.length, 3);
725 assert.equal(publications[2]?.state, null);
726 }),
727 );
728
729 it.effect("holds a completion through a continuation gap, then publishes it once", () =>
730 Effect.gen(function* () {
731 const { relay, currentShell, publications } = yield* makeTestRelay();
732 yield* relay.publishThread(THREAD_ID);
733 yield* Ref.set(currentShell, shell({ status: "completed" }));
734 yield* relay.publishThread(THREAD_ID);
735 yield* Ref.set(currentShell, shell());
736 yield* relay.publishThread(THREAD_ID);
737 yield* TestClock.adjust("5 seconds");
738 yield* relay.drain;
739 assert.deepEqual(
740 publications.map((publication) => publication.state?.phase),
741 ["running"],
742 );
743
744 yield* Ref.set(currentShell, shell({ status: "completed" }));
745 yield* relay.publishThread(THREAD_ID);
746 assert.equal(publications.length, 1);
747 yield* TestClock.adjust("5 seconds");
748 yield* relay.drain;
749 assert.deepEqual(
750 publications.map((publication) => publication.state?.phase),
751 ["running", "completed"],
752 );
753 }),
754 );
755
756 it.effect(
757 "clears a completion that landed while snoozed, even when processed after the wake",
758 () =>
759 Effect.gen(function* () {
760 yield* TestClock.setTime(DateTime.toEpochMillis(DateTime.makeUnsafe(NOW)));
761 const { relay, currentShell, publications } = yield* makeTestRelay();
762 const snooze = {
763 snoozedAt: DateTime.makeUnsafe(NOW),
764 snoozedUntil: DateTime.makeUnsafe("2026-09-04T13:00:00.000Z"),
765 };
766 yield* Ref.set(currentShell, shell(snooze));
767 yield* relay.publishThread(THREAD_ID);
768 yield* TestClock.adjust("2 hours");
769 yield* Ref.set(
770 currentShell,
771 shell({
772 ...snooze,
773 status: "completed",
774 latestRunCompletedAt: DateTime.makeUnsafe("2026-09-04T12:30:00.000Z"),
775 }),
776 );
777 yield* relay.publishThread(THREAD_ID);
778 yield* TestClock.adjust("5 seconds");
779 yield* relay.drain;
780 assert.deepEqual(
781 publications.map((publication) => publication.state?.phase ?? null),
782 ["running", null],
783 );
784 }),
785 );
786
787 it.effect("still publishes a completion when the thread settles during the hold", () =>
788 Effect.gen(function* () {
789 const { relay, currentShell, publications } = yield* makeTestRelay();
790 yield* relay.publishThread(THREAD_ID);
791 const completed = shell({ status: "completed" });
792 yield* Ref.set(currentShell, completed);
793 yield* relay.publishThread(THREAD_ID);
794 yield* Ref.set(currentShell, {
795 ...completed,
796 settledOverride: "settled",
797 settledAt: yield* DateTime.now,
798 lastVisitedAt: null,
799 });
800 yield* relay.publishThread(THREAD_ID);
801 yield* TestClock.adjust("5 seconds");
802 yield* relay.drain;
803 assert.deepEqual(
804 publications.map((publication) => publication.state?.phase ?? null),
805 ["running", "completed"],
806 );
807
808 yield* TestClock.adjust("5 seconds");
809 yield* relay.drain;
810 assert.deepEqual(
811 publications.map((publication) => publication.state?.phase ?? null),
812 ["running", "completed", null],
813 );
814 }),
815 );
816
817 it.effect("publishes a cleared thread's tombstone once while it stays cleared", () =>
818 Effect.gen(function* () {
819 const { relay, currentShell, publications } = yield* makeTestRelay();
820 yield* relay.publishThread(THREAD_ID);
821 const completed = shell({ status: "completed" });
822 yield* Ref.set(currentShell, completed);
823 yield* relay.publishThread(THREAD_ID);
824 yield* TestClock.adjust("5 seconds");
825 yield* relay.drain;
826 yield* Ref.set(currentShell, {
827 ...completed,
828 settledOverride: "settled",
829 settledAt: yield* DateTime.now,
830 lastVisitedAt: null,
831 });
832 yield* relay.publishThread(THREAD_ID);
833 yield* TestClock.adjust("5 seconds");
834 yield* relay.drain;
835 assert.deepEqual(
836 publications.map((publication) => publication.state?.phase ?? null),
837 ["running", "completed", null],
838 );
839
840 for (let i = 0; i < 3; i++) {
841 yield* relay.publishThread(THREAD_ID);
842 yield* TestClock.adjust("5 seconds");
843 yield* relay.drain;
844 }
845 assert.deepEqual(
846 publications.map((publication) => publication.state?.phase ?? null),
847 ["running", "completed", null],
848 );
849 }),
850 );
851
852 it.effect("does not alert an old completion after publishing its thread's tombstone", () =>
853 Effect.gen(function* () {
854 const oldCompleted = shell({
855 status: "completed",
856 latestRunCompletedAt: DateTime.makeUnsafe("2026-09-04T11:59:00.000Z"),
857 });
858 yield* TestClock.setTime(DateTime.toEpochMillis(DateTime.makeUnsafe(NOW)));
859 const { relay, currentShell, publications } = yield* makeTestRelay();
860 yield* Ref.set(currentShell, {
861 ...oldCompleted,
862 settledOverride: "settled",
863 settledAt: DateTime.makeUnsafe(NOW),
864 lastVisitedAt: null,
865 });
866 yield* relay.publishThread(THREAD_ID);
867 yield* TestClock.adjust("5 seconds");
868 yield* relay.drain;
869 assert.deepEqual(
870 publications.map((publication) => publication.state),
871 [null],
872 );
873
874 yield* Ref.set(currentShell, oldCompleted);
875 yield* relay.publishThread(THREAD_ID);
876 yield* TestClock.adjust("5 seconds");
877 yield* relay.drain;
878 assert.deepEqual(
879 publications.map((publication) => publication.state),
880 [null],
881 );
882 }),
883 );
884
885 it.effect("alerts a completion settled during its hold after an earlier tombstone", () =>
886 Effect.gen(function* () {
887 yield* TestClock.setTime(DateTime.toEpochMillis(DateTime.makeUnsafe(NOW)));
888 const { relay, currentShell, publications } = yield* makeTestRelay();
889 const settledAt = DateTime.makeUnsafe(NOW);
890 yield* Ref.set(currentShell, {
891 ...shell({ status: "completed" }),
892 settledOverride: "settled",
893 settledAt,
894 lastVisitedAt: null,
895 });
896 yield* relay.publishThread(THREAD_ID);
897 yield* TestClock.adjust("5 seconds");
898 yield* relay.drain;
899
900 const completed = shell({
901 status: "completed",
902 latestRunCompletedAt: DateTime.makeUnsafe("2026-09-04T12:00:06.000Z"),
903 });
904 yield* Ref.set(currentShell, completed);
905 yield* relay.publishThread(THREAD_ID);
906 yield* Ref.set(currentShell, {
907 ...completed,
908 settledOverride: "settled",
909 settledAt,
910 lastVisitedAt: null,
911 });
912 yield* relay.publishThread(THREAD_ID);
913 yield* TestClock.adjust("5 seconds");
914 yield* relay.drain;
915 yield* TestClock.adjust("5 seconds");
916 yield* relay.drain;
917 assert.deepEqual(
918 publications.map((publication) => publication.state?.phase ?? null),
919 [null, "completed", null],
920 );
921 }),
922 );
923
924 it.effect("drops a held completion when a queued continuation hides the thread", () =>
925 Effect.gen(function* () {
926 const { relay, currentShell, publications } = yield* makeTestRelay();
927 yield* relay.publishThread(THREAD_ID);
928 yield* Ref.set(currentShell, shell({ status: "completed" }));
929 yield* relay.publishThread(THREAD_ID);
930 yield* Ref.set(currentShell, shell({ status: "queued" }));
931 yield* relay.publishThread(THREAD_ID);
932 yield* TestClock.adjust("5 seconds");
933 yield* relay.drain;
934 assert.notInclude(
935 publications.map((publication) => publication.state?.phase ?? null),
936 "completed",
937 );
938 }),
939 );
940
941 it.effect("retries a failed completion even after the thread settles", () =>
942 Effect.gen(function* () {
943 const { relay, currentShell, publications } = yield* makeTestRelay({
944 respond: (attempt) =>
945 attempt === 2
946 ? new Response("unavailable", { status: 503 })
947 : Response.json({ ok: true, deliveries: [] }),
948 });
949 yield* relay.publishThread(THREAD_ID);
950 const completed = shell({ status: "completed" });
951 yield* Ref.set(currentShell, completed);
952 yield* relay.publishThread(THREAD_ID);
953 yield* TestClock.adjust("5 seconds");
954 yield* relay.drain;
955 yield* Ref.set(currentShell, {
956 ...completed,
957 settledOverride: "settled",
958 settledAt: yield* DateTime.now,
959 lastVisitedAt: null,
960 });
961 yield* TestClock.adjust("1 second");
962 yield* relay.drain;
963 assert.deepEqual(
964 publications.map((publication) => publication.state?.phase ?? null),
965 ["running", "completed", "completed"],
966 );
967
968 yield* TestClock.adjust("5 seconds");
969 yield* relay.drain;
970 assert.equal(publications.at(-1)?.state ?? null, null);
971 }),
972 );
973
974 it.effect("keeps settled live work until its awaited monitor finishes", () =>
975 Effect.gen(function* () {
976 const { relay, currentShell, publications } = yield* makeTestRelay();
977 yield* relay.publishThread(THREAD_ID);
978 const held = shell({
979 status: "completed",
980 settledOverride: "settled",
981 pendingBackgroundTasks: [{ taskId: "watch", kind: "monitor", awaited: true }],
982 });
983 yield* Ref.set(currentShell, held);
984 yield* relay.publishThread(THREAD_ID);
985 yield* TestClock.adjust("5 seconds");
986 yield* relay.drain;
987 assert.equal(publications.length, 1);
988 assert.equal(publications[0]?.state?.phase, "running");
989
990 yield* Ref.set(currentShell, { ...held, pendingBackgroundTasks: [] });
991 yield* relay.publishThread(THREAD_ID);
992 yield* TestClock.adjust("5 seconds");
993 yield* relay.drain;
994 assert.equal(publications.length, 2);
995 assert.equal(publications[1]?.state, null);
996 }),
997 );
998 it.effect("publishes a held completion when its background item ends, not on tool output", () =>
999 Effect.gen(function* () {
1000 const store = yield* ProjectionStore.ProjectionStoreV2;
1001 const events = yield* Queue.unbounded<OrchestrationV2DomainEvent>();
1002 const pulls = yield* Queue.unbounded<void>();
1003 const { relay, shellReads, publications } = yield* makeTestRelay({
1004 readShell: (threadId) => store.getThreadShell(threadId).pipe(Effect.orDie),
1005 domainEvents: Stream.fromEffectRepeat(
1006 Queue.offer(pulls, undefined).pipe(Effect.andThen(Queue.take(events))),
1007 ),
1008 });
1009 yield* relay.start();
1010 yield* Queue.take(pulls);
1011 const deliver = Effect.fnUntraced(function* (event: OrchestrationV2DomainEvent) {
1012 yield* store.apply(event);
1013 yield* Queue.offer(events, event);
1014 yield* Queue.take(pulls);
1015 yield* relay.drain;
1016 });
1017
1018 const now = yield* DateTime.now;
1019 const runId = RunId.make("run:held-item");
1020 const run = {
1021 id: runId,
1022 threadId: THREAD_ID,
1023 ordinal: 1,
1024 providerInstanceId: ProviderInstanceId.make("codex"),
1025 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "test-model" },
1026 providerThreadId: null,
1027 userMessageId: MessageId.make("message:held-item"),
1028 rootNodeId: null,
1029 activeAttemptId: null,
1030 status: "running" as const,
1031 requestedAt: now,
1032 startedAt: now,
1033 completedAt: null,
1034 checkpointId: null,
1035 contextHandoffId: null,
1036 };
1037 const item = (
1038 id: string,
1039 fields: Pick<OrchestrationV2TurnItem, "status"> &
1040 (
1041 | { readonly type: "dynamic_tool"; readonly toolName: string; readonly input: unknown }
1042 | {
1043 readonly type: "assistant_message";
1044 readonly messageId: MessageId;
1045 readonly text: string;
1046 readonly streaming: boolean;
1047 }
1048 ),
1049 ): OrchestrationV2TurnItem => ({
1050 id: TurnItemId.make(id),
1051 threadId: THREAD_ID,
1052 runId,
1053 nodeId: NodeId.make("node:held-item"),
1054 providerThreadId: null,
1055 providerTurnId: null,
1056 nativeItemRef: null,
1057 parentItemId: null,
1058 ordinal: 1,
1059 title: null,
1060 startedAt: now,
1061 completedAt: null,
1062 updatedAt: now,
1063 ...fields,
1064 });
1065 const itemEvent = (id: string, payload: OrchestrationV2TurnItem) =>
1066 ({
1067 id: EventId.make(id),
1068 type: "turn-item.updated",
1069 threadId: THREAD_ID,
1070 runId,
1071 occurredAt: now,
1072 payload,
1073 }) satisfies OrchestrationV2DomainEvent;
1074 const background = { type: "dynamic_tool" as const, toolName: "watch", input: {} };
1075
1076 yield* deliver({
1077 id: EventId.make("event:held-item:thread"),
1078 type: "thread.created",
1079 threadId: THREAD_ID,
1080 occurredAt: now,
1081 payload: {
1082 createdBy: "user",
1083 creationSource: "web",
1084 id: THREAD_ID,
1085 projectId: PROJECT_ID,
1086 title: "Thread",
1087 providerInstanceId: ProviderInstanceId.make("codex"),
1088 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "test-model" },
1089 runtimeMode: "full-access",
1090 interactionMode: "default",
1091 branch: null,
1092 worktreePath: null,
1093 activeProviderThreadId: null,
1094 lineage: { rootThreadId: THREAD_ID, parentThreadId: null, relationshipToParent: null },
1095 forkedFrom: null,
1096 createdAt: now,
1097 updatedAt: now,
1098 archivedAt: null,
1099 settledOverride: null,
1100 settledAt: null,
1101 lastVisitedAt: null,
1102 deletedAt: null,
1103 },
1104 });
1105 yield* deliver({
1106 id: EventId.make("event:held-item:run"),
1107 type: "run.created",
1108 threadId: THREAD_ID,
1109 runId,
1110 occurredAt: now,
1111 payload: run,
1112 });
1113 yield* deliver(
1114 itemEvent("event:held-item:tool", item("item:tool", { ...background, status: "running" })),
1115 );
1116 yield* deliver({
1117 id: EventId.make("event:held-item:run-completed"),
1118 type: "run.updated",
1119 threadId: THREAD_ID,
1120 runId,
1121 occurredAt: now,
1122 payload: { ...run, status: "completed", completedAt: now },
1123 });
1124 assert.deepEqual(
1125 publications.map((publication) => publication.state?.phase),
1126 ["running"],
1127 );
1128
1129 const readsBeforeOutput = shellReads.length;
1130 yield* deliver(
1131 itemEvent(
1132 "event:held-item:reply",
1133 item("item:reply", {
1134 type: "assistant_message",
1135 messageId: MessageId.make("message:held-item:reply"),
1136 text: "Still watching",
1137 streaming: true,
1138 status: "running",
1139 }),
1140 ),
1141 );
1142 yield* deliver(
1143 itemEvent("event:held-item:tick", item("item:tool", { ...background, status: "running" })),
1144 );
1145 assert.equal(shellReads.length, readsBeforeOutput);
1146
1147 yield* deliver(
1148 itemEvent("event:held-item:end", item("item:tool", { ...background, status: "completed" })),
1149 );
1150 yield* TestClock.adjust("5 seconds");
1151 yield* relay.drain;
1152 assert.deepEqual(
1153 publications.map((publication) => publication.state?.phase),
1154 ["running", "completed"],
1155 );
1156 }).pipe(
1157 Effect.scoped,
1158 Effect.provide(ProjectionStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory))),
1159 ),
1160 );
1161});
1162
1163it.effect("catches up after startup without polling after success", () =>
1164 Effect.gen(function* () {
1165 const { relay, secrets, secretReads, catchUp, publications } = yield* makeTestRelay({
1166 catchUpThreads: [shell()],
1167 });
1168 yield* relay.start();
1169 yield* TestClock.adjust("1 second");
1170 yield* relay.drain;
1171 assert.equal(catchUp.activitySnapshotReads, 1);
1172 assert.equal(publications.length, 1);
1173 assert.include(publications[0]?.state, {
1174 threadId: THREAD_ID,
1175 projectTitle: "Project",
1176 threadTitle: "Thread",
1177 phase: "running",
1178 });
1179
1180 yield* secrets.set(PUBLISH_AGENT_ACTIVITY_SECRET, new TextEncoder().encode("false"));
1181 yield* relay.requestCatchUp();
1182 yield* TestClock.adjust("1 second");
1183 assert.equal(catchUp.activitySnapshotReads, 1);
1184
1185 yield* secrets.set(PUBLISH_AGENT_ACTIVITY_SECRET, new TextEncoder().encode("true"));
1186 yield* relay.requestCatchUp();
1187 yield* TestClock.adjust("1 second");
1188 assert.equal(catchUp.activitySnapshotReads, 2);
1189
1190 const readsBeforeIdle = secretReads.length;
1191 yield* TestClock.adjust("10 minutes");
1192 assert.equal(catchUp.activitySnapshotReads, 2);
1193 assert.equal(secretReads.length, readsBeforeIdle);
1194
1195 yield* secrets.set(
1196 RELAY_ENVIRONMENT_CREDENTIAL_SECRET,
1197 new TextEncoder().encode("credential-2"),
1198 );
1199 yield* relay.requestCatchUp();
1200 yield* TestClock.adjust("1 second");
1201 yield* relay.drain;
1202 assert.equal(catchUp.activitySnapshotReads, 3);
1203 assert.equal(publications.length, 2);
1204 assert.equal(publications[1]?.state?.threadId, THREAD_ID);
1205 }),
1206);