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 ThreadId,
15 TurnItemId,
16} from "@t3tools/contracts";
17import { RelayAgentActivityState } from "@t3tools/contracts/relay";
18import * as DateTime from "effect/DateTime";
19import * as Deferred from "effect/Deferred";
20import * as Effect from "effect/Effect";
21import * as Fiber from "effect/Fiber";
22import * as Option from "effect/Option";
23import * as Queue from "effect/Queue";
24import * as Ref from "effect/Ref";
25import * as Schema from "effect/Schema";
26import * as Stream from "effect/Stream";
27import * as TestClock from "effect/testing/TestClock";
28import * as FetchHttpClient from "effect/http/FetchHttpClient";
29
30import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
31import {
32 PUBLISH_AGENT_ACTIVITY_SECRET,
33 RELAY_ENVIRONMENT_CREDENTIAL_SECRET,
34 RELAY_ISSUER_SECRET,
35 RELAY_URL_SECRET,
36} from "../cloud/config.ts";
37import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
38import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
39import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
40import * as ProjectService from "../project/ProjectService.ts";
41import * as AgentAwarenessRelay from "./AgentAwarenessRelay.ts";
42
43const THREAD_ID = ThreadId.make("relay-thread");
44const SECOND_THREAD_ID = ThreadId.make("relay-thread-2");
45const PROJECT_ID = ProjectId.make("relay-project");
46const NOW = "2026-09-04T12:00:00.000Z";
47
48function shell(overrides: Partial<OrchestrationV2ThreadShell> = {}): OrchestrationV2ThreadShell {
49 return {
50 id: THREAD_ID,
51 projectId: PROJECT_ID,
52 title: "Thread",
53 providerInstanceId: ProviderInstanceId.make("codex"),
54 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "test-model" },
55 runtimeMode: "full-access",
56 interactionMode: "default",
57 worktreePath: null,
58 activeProviderThreadId: null,
59 lineage: { rootThreadId: THREAD_ID, parentThreadId: null, relationshipToParent: null },
60 forkedFrom: null,
61 createdBy: "user",
62 creationSource: "web",
63 activeRunId: null,
64 latestVisibleMessage: null,
65 hasActionableProposedPlan: false,
66 itemCount: 0,
67 visibleItemCount: 0,
68 lastVisitedAt: null,
69 deletedAt: null,
70 branch: null,
71 linkedPullRequest: null,
72 status: "running",
73 activityRunStatus: null,
74 pendingRuntimeRequest: null,
75 pendingBackgroundTasks: [],
76 latestRunId: null,
77 latestRunRequestedAt: null,
78 latestRunStartedAt: null,
79 latestRunCompletedAt: null,
80 latestUserMessageAt: null,
81 createdAt: DateTime.makeUnsafe(NOW),
82 updatedAt: DateTime.makeUnsafe(NOW),
83 archivedAt: null,
84 settledOverride: null,
85 settledAt: null,
86 snoozedUntil: null,
87 snoozedAt: null,
88 pinnedAt: null,
89 ...overrides,
90 };
91}
92
93const PublishPayload = Schema.Struct({ state: Schema.NullOr(RelayAgentActivityState) });
94const decodePublishPayload = Schema.decodeUnknownSync(Schema.fromJsonString(PublishPayload));
95const unused = () => Effect.die("Unexpected test dependency call");
96
97describe("startup agent activity", () => {
98 it("publishes active work and only terminal runs completed after startup", () => {
99 const startedAt = DateTime.toEpochMillis(DateTime.makeUnsafe(NOW));
100 const oldCompleted = ThreadId.make("old-completed");
101 const newCompleted = ThreadId.make("new-completed");
102 const oldFailed = ThreadId.make("old-failed");
103 const newFailed = ThreadId.make("new-failed");
104 const ids = AgentAwarenessRelay.resolveAgentAwarenessRelayActiveThreadIds({
105 environmentId: EnvironmentId.make("relay-env"),
106 startedAt,
107 projects: [{ id: PROJECT_ID, title: "Project" }],
108 threads: [
109 shell(),
110 shell({
111 id: oldCompleted,
112 status: "completed",
113 latestRunCompletedAt: DateTime.makeUnsafe("2026-09-04T11:59:00.000Z"),
114 }),
115 shell({
116 id: newCompleted,
117 status: "completed",
118 latestRunCompletedAt: DateTime.makeUnsafe("2026-09-04T12:00:01.000Z"),
119 }),
120 shell({
121 id: oldFailed,
122 status: "failed",
123 latestRunCompletedAt: DateTime.makeUnsafe("2026-09-04T11:59:00.000Z"),
124 }),
125 shell({
126 id: newFailed,
127 status: "failed",
128 latestRunCompletedAt: DateTime.makeUnsafe("2026-09-04T12:00:01.000Z"),
129 }),
130 shell({ id: ThreadId.make("idle"), status: "idle" }),
131 shell({ id: ThreadId.make("missing-project"), projectId: ProjectId.make("missing") }),
132 ],
133 });
134 assert.deepStrictEqual(ids, [THREAD_ID, newCompleted, newFailed]);
135 });
136});
137
138const makeTestRelay = Effect.fnUntraced(function* (
139 options: {
140 readonly respond?: (attempt: number) => Response;
141 readonly failSecretRead?: (name: string) => boolean;
142 /** Starts unlinked with publishing off when false. */
143 readonly linked?: boolean;
144 /** Serves shells from this source instead of `currentShell`. */
145 readonly readShell?: (threadId: ThreadId) => Effect.Effect<OrchestrationV2ThreadShell | null>;
146 readonly domainEvents?: Stream.Stream<OrchestrationV2DomainEvent>;
147 } = {},
148) {
149 const values = new Map<string, Uint8Array>(
150 options.linked === false
151 ? []
152 : [
153 [PUBLISH_AGENT_ACTIVITY_SECRET, new TextEncoder().encode("true")],
154 [RELAY_URL_SECRET, new TextEncoder().encode("https://relay.example.test")],
155 [RELAY_ISSUER_SECRET, new TextEncoder().encode("https://relay.example.test")],
156 [RELAY_ENVIRONMENT_CREDENTIAL_SECRET, new TextEncoder().encode("credential-1")],
157 ],
158 );
159 const secretReads: string[] = [];
160 const secrets = ServerSecretStore.ServerSecretStore.of({
161 get: (name) =>
162 Effect.suspend(() => {
163 secretReads.push(name);
164 if (options.failSecretRead?.(name)) {
165 return Effect.fail(
166 new ServerSecretStore.SecretStoreReadError({
167 resource: name,
168 cause: "temporary read failure",
169 }),
170 );
171 }
172 return Effect.succeed(Option.fromUndefinedOr(values.get(name)));
173 }),
174 create: (name, value) => Effect.sync(() => void values.set(name, value)),
175 set: (name, value) => Effect.sync(() => void values.set(name, value)),
176 remove: (name) => Effect.sync(() => void values.delete(name)),
177 getOrCreateRandom: unused,
178 });
179 const currentShell = yield* Ref.make<OrchestrationV2ThreadShell | null>(shell());
180 const shellReads: ThreadId[] = [];
181 // Catch-up publishes read the whole shell once each.
182 const catchUp = { shellSnapshotReads: 0 };
183 const threads = ThreadManagementService.ThreadManagementService.of({
184 getThreadShell: (threadId) =>
185 Effect.sync(() => shellReads.push(threadId)).pipe(
186 Effect.andThen(options.readShell?.(threadId) ?? Ref.get(currentShell)),
187 ),
188 getShellSnapshot: () =>
189 Effect.sync(() => {
190 catchUp.shellSnapshotReads += 1;
191 return { schemaVersion: 2, snapshotSequence: 1, threads: [], archivedThreads: [] };
192 }),
193 readShellSnapshot: unused,
194 ensureLegacyTranscript: unused,
195 dispatch: unused,
196 searchThread: () => Effect.die("unused"),
197 searchThreadStream: () => Stream.empty,
198 getThreadHistoryPage: () => Effect.die("unused"),
199 getTimelinePage: () => Effect.die("Unused timeline read"),
200 getMessageCount: () => Effect.die("unused message count"),
201 getTurnItem: () => Effect.die("unused turn item read"),
202 getThreadRecords: () => Effect.die("unused record read"),
203 getThreadProjection: unused,
204 getCheckpointContext: unused,
205 getThreadSnapshot: unused,
206 getThreadTranscript: unused,
207 getThreadSnapshotWindow: unused,
208 getProjectThreadRecords: () => Effect.die("unused project record read"),
209 getProjectThread: unused,
210 listProjectThreads: unused,
211 sendToThread: unused,
212 assertProviderInstanceEnabled: unused,
213 isProviderInstanceEnabledForProject: unused,
214 hasReceipt: unused,
215 waitForThread: unused,
216 settleAfterRun: unused,
217 settleThread: unused,
218 interruptThread: unused,
219 stopWorkflow: () => Effect.die("unused stopWorkflow"),
220 stopDelegatedTasks: unused,
221 getThreadEventSequence: unused,
222 recoverDelegatedTask: unused,
223 delegatedTaskResultPending: unused,
224 streamStoredEvents: Stream.empty,
225 streamStoredEventsFrom: () => Stream.empty,
226 streamDomainEvents: options.domainEvents ?? Stream.empty,
227 });
228 const publications: Array<{
229 readonly url: string;
230 readonly authorization: string | null;
231 readonly state: RelayAgentActivityState | null;
232 }> = [];
233 const fetch: typeof globalThis.fetch = Object.assign(
234 (
235 input: Parameters<typeof globalThis.fetch>[0],
236 init?: Parameters<typeof globalThis.fetch>[1],
237 ) => {
238 const body = init?.body;
239 if (typeof body !== "string" && !(body instanceof Uint8Array)) {
240 return Promise.reject(new Error("Expected a serialized activity publish payload"));
241 }
242 const payload = decodePublishPayload(
243 typeof body === "string" ? body : new TextDecoder().decode(body),
244 );
245 publications.push({
246 url: String(input),
247 authorization: new Headers(init?.headers).get("authorization"),
248 state: payload.state,
249 });
250 return Promise.resolve(
251 options.respond?.(publications.length) ?? Response.json({ ok: true, deliveries: [] }),
252 );
253 },
254 { preconnect: () => {} },
255 );
256 const relay = yield* AgentAwarenessRelay.make.pipe(
257 Effect.provideService(ServerSecretStore.ServerSecretStore, secrets),
258 Effect.provideService(ThreadManagementService.ThreadManagementService, threads),
259 Effect.provideService(ServerEnvironment.ServerEnvironment, {
260 getEnvironmentId: Effect.succeed(EnvironmentId.make("relay-environment")),
261 getDescriptor: unused(),
262 }),
263 Effect.provideService(ProjectService.ProjectService, {
264 create: unused,
265 bootstrap: unused,
266 update: unused,
267 delete: unused,
268 getByWorkspaceRoot: unused,
269 snapshot: Effect.succeed({ projects: [] } as never),
270 getShell: unused,
271 listShells: unused,
272 getById: () =>
273 Effect.succeed(
274 Option.some({
275 id: PROJECT_ID,
276 title: "Project",
277 workspaceRoot: "/workspace",
278 defaultModelSelection: null,
279 scripts: [],
280 createdAt: NOW,
281 updatedAt: NOW,
282 deletedAt: null,
283 }),
284 ),
285 }),
286 Effect.provideService(FetchHttpClient.Fetch, fetch),
287 Effect.provide(NodeCrypto.layer),
288 );
289 return { relay, secrets, secretReads, currentShell, shellReads, publications, catchUp };
290});
291
292describe("AgentAwarenessRelay", () => {
293 it("ignores transcript and tool updates but retains activity and metadata changes", () => {
294 for (const type of [
295 "message.updated",
296 "turn-item.updated",
297 "provider-turn.updated",
298 "thread.visited",
299 "thread.pinned",
300 ] as const) {
301 assert.isFalse(AgentAwarenessRelay.shouldPublishAgentAwarenessEvent({ type }));
302 }
303 for (const type of [
304 "run.created",
305 "run.updated",
306 "runtime-request.updated",
307 // Pending background work changes can release a held completion.
308 "subagent.updated",
309 "provider-thread.updated",
310 "thread.metadata-updated",
311 "thread.model-selection-updated",
312 "thread.provider-switched",
313 "thread.archived",
314 "thread.unarchived",
315 "thread.deleted",
316 ] as const) {
317 assert.isTrue(AgentAwarenessRelay.shouldPublishAgentAwarenessEvent({ type }));
318 }
319 });
320
321 it("does not publish imported thread creation as new agent activity", () => {
322 assert.isFalse(
323 AgentAwarenessRelay.shouldPublishAgentAwarenessEvent({
324 type: "thread.created",
325 payload: { historyOrigin: "v1_import" },
326 }),
327 );
328 assert.isTrue(
329 AgentAwarenessRelay.shouldPublishAgentAwarenessEvent({ type: "thread.created", payload: {} }),
330 );
331 });
332
333 it.effect("coalesces queued updates and reruns a thread dirtied during publishing", () =>
334 Effect.gen(function* () {
335 const started = yield* Deferred.make<void>();
336 const release = yield* Deferred.make<void>();
337 const rerunStarted = yield* Deferred.make<void>();
338 const releaseRerun = yield* Deferred.make<void>();
339 const processed: Array<{ threadId: ThreadId; revision: number }> = [];
340 let revision = 1;
341 const worker = yield* AgentAwarenessRelay.makeAgentAwarenessPublishWorker((threadId) =>
342 Effect.gen(function* () {
343 const currentRevision = revision;
344 const index = processed.length;
345 processed.push({ threadId, revision: currentRevision });
346 if (index === 0) {
347 yield* Deferred.succeed(started, undefined);
348 yield* Deferred.await(release);
349 } else if (threadId === THREAD_ID) {
350 yield* Deferred.succeed(rerunStarted, undefined);
351 yield* Deferred.await(releaseRerun);
352 }
353 }),
354 );
355 yield* worker.enqueue(THREAD_ID);
356 yield* Deferred.await(started);
357 revision = 2;
358 for (let i = 0; i < 100; i++) {
359 yield* worker.enqueue(SECOND_THREAD_ID);
360 yield* worker.enqueue(THREAD_ID);
361 }
362 const drained = yield* Deferred.make<void>();
363 const draining = yield* worker.drain.pipe(
364 Effect.andThen(Deferred.succeed(drained, undefined)),
365 Effect.forkChild,
366 );
367 yield* Deferred.succeed(release, undefined);
368 yield* Deferred.await(rerunStarted);
369 assert.isFalse(yield* Deferred.isDone(drained));
370 yield* Deferred.succeed(releaseRerun, undefined);
371 yield* Fiber.join(draining);
372 assert.deepEqual(processed, [
373 { threadId: THREAD_ID, revision: 1 },
374 { threadId: SECOND_THREAD_ID, revision: 2 },
375 { threadId: THREAD_ID, revision: 2 },
376 ]);
377 }),
378 );
379
380 it.effect("deduplicates state and republishes title changes", () =>
381 Effect.gen(function* () {
382 const { relay, currentShell, publications } = yield* makeTestRelay();
383 yield* relay.publishThread(THREAD_ID);
384 yield* Ref.set(
385 currentShell,
386 shell({ updatedAt: DateTime.makeUnsafe("2026-09-04T13:00:00Z") }),
387 );
388 yield* relay.publishThread(THREAD_ID);
389 assert.equal(publications.length, 1);
390 yield* Ref.set(currentShell, shell({ title: "Renamed thread" }));
391 yield* relay.publishThread(THREAD_ID);
392 assert.equal(publications.length, 2);
393 assert.equal(publications[1]?.state?.threadTitle, "Renamed thread");
394 }),
395 );
396
397 it.effect("stops before shell reads when disabled and republishes after re-enabling", () =>
398 Effect.gen(function* () {
399 const { relay, secrets, secretReads, shellReads, publications } = yield* makeTestRelay();
400 yield* relay.publishThread(THREAD_ID);
401 yield* secrets.set(PUBLISH_AGENT_ACTIVITY_SECRET, new TextEncoder().encode("false"));
402 const previousReads = secretReads.length;
403 yield* relay.publishThread(THREAD_ID);
404 assert.deepEqual(secretReads.slice(previousReads), [PUBLISH_AGENT_ACTIVITY_SECRET]);
405 assert.equal(shellReads.length, 1);
406 assert.equal(publications.length, 1);
407 yield* secrets.set(PUBLISH_AGENT_ACTIVITY_SECRET, new TextEncoder().encode("true"));
408 yield* relay.publishThread(THREAD_ID);
409 assert.equal(publications.length, 2);
410 }),
411 );
412
413 it.effect("republishes unchanged state with fresh credentials after relinking", () =>
414 Effect.gen(function* () {
415 const { relay, secrets, publications } = yield* makeTestRelay();
416 yield* relay.publishThread(THREAD_ID);
417 yield* secrets.set(
418 RELAY_ENVIRONMENT_CREDENTIAL_SECRET,
419 new TextEncoder().encode("credential-2"),
420 );
421 yield* relay.publishThread(THREAD_ID);
422 assert.equal(publications.length, 2);
423 assert.equal(publications[0]?.authorization, "Bearer credential-1");
424 assert.equal(publications[1]?.authorization, "Bearer credential-2");
425 yield* secrets.set(
426 RELAY_URL_SECRET,
427 new TextEncoder().encode("https://new-relay.example.test"),
428 );
429 yield* relay.publishThread(THREAD_ID);
430 assert.equal(publications.length, 3);
431 assert.isTrue(publications[2]?.url.startsWith("https://new-relay.example.test/") ?? false);
432 }),
433 );
434
435 it.effect("a stalled response body releases later threads and uses bounded retries", () =>
436 Effect.gen(function* () {
437 const bodyStarted = Promise.withResolvers<void>();
438 const secondRead = yield* Deferred.make<void>();
439 const currentShell = yield* Ref.make(shell());
440 const { relay, publications } = yield* makeTestRelay({
441 readShell: (threadId) =>
442 threadId === SECOND_THREAD_ID
443 ? Deferred.succeed(secondRead, undefined).pipe(
444 Effect.as(shell({ id: SECOND_THREAD_ID })),
445 )
446 : Ref.get(currentShell),
447 respond: (attempt) => {
448 if (attempt !== 1) return new Response("relay unavailable", { status: 503 });
449 const response = new Response(null, {
450 headers: { "content-type": "application/json" },
451 });
452 return Object.assign(response, {
453 arrayBuffer: () => {
454 bodyStarted.resolve();
455 return new Promise<ArrayBuffer>(() => {});
456 },
457 });
458 },
459 });
460 const first = yield* relay.publishThread(THREAD_ID).pipe(Effect.forkChild);
461 yield* Effect.promise(() => bodyStarted.promise);
462 const second = yield* relay.publishThread(SECOND_THREAD_ID).pipe(Effect.forkChild);
463 yield* TestClock.adjust("9999 millis");
464 assert.equal(publications.length, 1);
465 yield* Ref.set(currentShell, shell({ title: "Latest title" }));
466 yield* TestClock.adjust("1 millis");
467 // Assert progress before draining so the base fails instead of hanging.
468 assert.isTrue(yield* Deferred.isDone(secondRead));
469 yield* Fiber.join(first);
470 yield* Fiber.join(second);
471 assert.equal(publications.length, 2);
472 assert.isTrue(publications[1]?.url.includes(SECOND_THREAD_ID) ?? false);
473 for (const delay of [1_000, 2_000, 4_000, 8_000, 16_000]) {
474 const attempts = publications.length;
475 yield* TestClock.adjust(delay - 1);
476 yield* relay.drain;
477 assert.equal(publications.length, attempts);
478 yield* TestClock.adjust(1);
479 yield* relay.drain;
480 assert.equal(publications.length, attempts + 2);
481 assert.equal(publications[attempts]?.state?.threadTitle, "Latest title");
482 }
483 yield* TestClock.adjust("1 minute");
484 yield* relay.drain;
485 assert.equal(publications.length, 12);
486 }),
487 );
488
489 it.effect("retries a failed final state without another thread event", () =>
490 Effect.gen(function* () {
491 const { relay, currentShell, shellReads, publications } = yield* makeTestRelay({
492 respond: (attempt) =>
493 attempt === 2
494 ? new Response("relay unavailable", { status: 503 })
495 : Response.json({ ok: true, deliveries: [] }),
496 });
497 yield* relay.publishThread(THREAD_ID);
498 yield* Ref.set(
499 currentShell,
500 shell({
501 status: "completed",
502 latestRunCompletedAt: DateTime.add(yield* DateTime.now, { seconds: 1 }),
503 }),
504 );
505 yield* relay.publishThread(THREAD_ID);
506 assert.equal(publications.length, 2);
507 yield* TestClock.adjust("1 second");
508 yield* relay.drain;
509 assert.equal(publications.length, 3);
510 assert.equal(publications[2]?.state?.phase, "completed");
511 yield* TestClock.adjust("1 minute");
512 yield* relay.drain;
513 assert.equal(shellReads.length, 3);
514 }),
515 );
516
517 it.effect("retries the latest shell and credentials with first-completion confirmation", () =>
518 Effect.gen(function* () {
519 const { relay, secrets, currentShell, publications } = yield* makeTestRelay({
520 respond: (attempt) =>
521 attempt === 1
522 ? new Response("relay unavailable", { status: 503 })
523 : Response.json({ ok: true, deliveries: [] }),
524 });
525 yield* relay.publishThread(THREAD_ID);
526 yield* Ref.set(
527 currentShell,
528 shell({
529 status: "completed",
530 title: "Final title",
531 latestRunCompletedAt: DateTime.add(yield* DateTime.now, { seconds: 1 }),
532 }),
533 );
534 yield* secrets.set(
535 RELAY_ENVIRONMENT_CREDENTIAL_SECRET,
536 new TextEncoder().encode("credential-2"),
537 );
538 yield* TestClock.adjust("1 second");
539 yield* relay.drain;
540 assert.equal(publications.length, 1);
541 yield* TestClock.adjust("5 seconds");
542 yield* relay.drain;
543 assert.equal(publications.length, 2);
544 assert.equal(publications[1]?.authorization, "Bearer credential-2");
545 assert.equal(publications[1]?.state?.phase, "completed");
546 assert.equal(publications[1]?.state?.threadTitle, "Final title");
547 }),
548 );
549
550 it.effect("cancels a stale retry when a newer update publishes successfully", () =>
551 Effect.gen(function* () {
552 const { relay, currentShell, shellReads, publications } = yield* makeTestRelay({
553 respond: (attempt) =>
554 attempt === 1
555 ? new Response("relay unavailable", { status: 503 })
556 : Response.json({ ok: true, deliveries: [] }),
557 });
558 yield* relay.publishThread(THREAD_ID);
559 yield* Ref.set(currentShell, shell({ title: "New title" }));
560 yield* relay.publishThread(THREAD_ID);
561 assert.equal(publications.length, 2);
562 assert.equal(publications[1]?.state?.threadTitle, "New title");
563 yield* TestClock.adjust("1 minute");
564 yield* relay.drain;
565 assert.equal(shellReads.length, 2);
566 }),
567 );
568
569 it.effect.each(["disable", "unlink"] as const)(
570 "stops retries before shell reads after %s",
571 (change) =>
572 Effect.gen(function* () {
573 const { relay, secrets, shellReads, publications } = yield* makeTestRelay({
574 respond: () => new Response("relay unavailable", { status: 503 }),
575 });
576 yield* relay.publishThread(THREAD_ID);
577 if (change === "disable") {
578 yield* secrets.set(PUBLISH_AGENT_ACTIVITY_SECRET, new TextEncoder().encode("false"));
579 } else {
580 yield* secrets.remove(RELAY_ENVIRONMENT_CREDENTIAL_SECRET);
581 }
582 yield* TestClock.adjust("1 second");
583 yield* relay.drain;
584 yield* TestClock.adjust("1 minute");
585 yield* relay.drain;
586 assert.equal(publications.length, 1);
587 assert.equal(shellReads.length, 1);
588 }),
589 );
590
591 it.effect("recovers from a transient configuration read failure", () =>
592 Effect.gen(function* () {
593 let failRead = false;
594 const { relay, shellReads, publications } = yield* makeTestRelay({
595 failSecretRead: (name) => failRead && name === PUBLISH_AGENT_ACTIVITY_SECRET,
596 });
597 failRead = true;
598 yield* relay.publishThread(THREAD_ID);
599 assert.equal(shellReads.length, 0);
600 failRead = false;
601 yield* TestClock.adjust("1 second");
602 yield* relay.drain;
603 assert.equal(publications.length, 1);
604 assert.equal(publications[0]?.state?.phase, "running");
605 }),
606 );
607
608 it.effect("bounds retry attempts and resets the budget for a newer update", () =>
609 Effect.gen(function* () {
610 const { relay, currentShell, shellReads, publications } = yield* makeTestRelay({
611 respond: () => new Response("relay unavailable", { status: 503 }),
612 });
613 yield* relay.publishThread(THREAD_ID);
614 let attempts = 1;
615 for (const delay of [1_000, 2_000, 4_000, 8_000, 16_000]) {
616 yield* TestClock.adjust(delay - 1);
617 yield* relay.drain;
618 assert.equal(publications.length, attempts);
619 yield* TestClock.adjust(1);
620 yield* relay.drain;
621 attempts += 1;
622 assert.equal(publications.length, attempts);
623 }
624 yield* TestClock.adjust("1 minute");
625 yield* relay.drain;
626 assert.equal(publications.length, 6);
627 assert.equal(shellReads.length, 6);
628 yield* Ref.set(currentShell, shell({ title: "New update" }));
629 yield* relay.publishThread(THREAD_ID);
630 yield* TestClock.adjust("1 second");
631 yield* relay.drain;
632 assert.equal(publications.length, 8);
633 assert.equal(publications[7]?.state?.threadTitle, "New update");
634 }),
635 );
636
637 it.effect("retries a confirmed completion without restarting its confirmation window", () =>
638 Effect.gen(function* () {
639 const { relay, currentShell, publications } = yield* makeTestRelay({
640 respond: (attempt) =>
641 attempt <= 2
642 ? new Response("relay unavailable", { status: 503 })
643 : Response.json({ ok: true, deliveries: [] }),
644 });
645 yield* Ref.set(
646 currentShell,
647 shell({
648 status: "completed",
649 latestRunCompletedAt: DateTime.add(yield* DateTime.now, { seconds: 1 }),
650 }),
651 );
652 yield* relay.publishThread(THREAD_ID);
653 yield* TestClock.adjust("5 seconds");
654 yield* relay.drain;
655 assert.equal(publications.length, 1);
656 yield* TestClock.adjust("1 second");
657 yield* relay.drain;
658 assert.equal(publications.length, 2);
659 yield* TestClock.adjust("2 seconds");
660 yield* relay.drain;
661 assert.equal(publications.length, 3);
662 assert.isTrue(publications.every((publication) => publication.state?.phase === "completed"));
663 yield* TestClock.adjust("1 minute");
664 yield* relay.drain;
665 assert.equal(publications.length, 3);
666 }),
667 );
668
669 it.effect("does not reset an exhausted retry budget for old confirmation timers", () =>
670 Effect.gen(function* () {
671 let unavailable = false;
672 const { relay, currentShell, shellReads, publications } = yield* makeTestRelay({
673 respond: () =>
674 unavailable
675 ? new Response("relay unavailable", { status: 503 })
676 : Response.json({ ok: true, deliveries: [] }),
677 });
678 yield* relay.publishThread(THREAD_ID);
679 // Leave six obsolete confirmations at 5,000 through 5,005 milliseconds.
680 for (let i = 0; i < 6; i++) {
681 yield* Ref.set(currentShell, null);
682 yield* relay.publishThread(THREAD_ID);
683 yield* Ref.set(currentShell, shell());
684 yield* relay.publishThread(THREAD_ID);
685 yield* TestClock.adjust(1);
686 }
687 unavailable = true;
688 yield* Ref.set(currentShell, shell({ status: "failed" }));
689 yield* relay.publishThread(THREAD_ID);
690 yield* TestClock.adjust(1_000);
691 yield* relay.drain;
692 yield* TestClock.adjust(2_000);
693 yield* relay.drain;
694 yield* TestClock.adjust(1_994);
695 yield* relay.drain;
696 for (let i = 0; i < 5; i++) {
697 yield* TestClock.adjust(1);
698 yield* relay.drain;
699 }
700 const attempts = publications.length;
701 const reads = shellReads.length;
702 yield* TestClock.adjust("1 minute");
703 yield* relay.drain;
704 assert.equal(publications.length, attempts);
705 assert.equal(shellReads.length, reads);
706 }),
707 );
708
709 it.effect("confirms tombstones and cancels stale confirmations when activity recovers", () =>
710 Effect.gen(function* () {
711 const { relay, currentShell, publications } = yield* makeTestRelay();
712 yield* relay.publishThread(THREAD_ID);
713 yield* Ref.set(currentShell, null);
714 yield* relay.publishThread(THREAD_ID);
715 assert.equal(publications.length, 1);
716 yield* Ref.set(currentShell, shell());
717 yield* relay.publishThread(THREAD_ID);
718 yield* TestClock.adjust("5 seconds");
719 yield* relay.drain;
720 assert.equal(publications.length, 1);
721 yield* Ref.set(currentShell, null);
722 yield* relay.publishThread(THREAD_ID);
723 assert.equal(publications.length, 1);
724 yield* TestClock.adjust("5 seconds");
725 yield* relay.drain;
726 assert.equal(publications.length, 2);
727 assert.equal(publications[1]?.state, null);
728 }),
729 );
730
731 it.effect.each([
732 { label: "live", archived: false },
733 { label: "archived", archived: true },
734 ])("never publishes tombstones for $label subagent threads", ({ archived }) =>
735 Effect.gen(function* () {
736 const { relay, currentShell, publications } = yield* makeTestRelay();
737 yield* Ref.set(
738 currentShell,
739 shell({
740 lineage: {
741 rootThreadId: THREAD_ID,
742 parentThreadId: THREAD_ID,
743 relationshipToParent: "subagent",
744 },
745 ...(archived ? { archivedAt: yield* DateTime.now } : {}),
746 }),
747 );
748 yield* relay.publishThread(THREAD_ID);
749 yield* TestClock.adjust("5 seconds");
750 yield* relay.drain;
751 assert.equal(publications.length, 0);
752 }),
753 );
754
755 it.effect("confirms a first completed state and respects disabling during confirmation", () =>
756 Effect.gen(function* () {
757 const { relay, secrets, currentShell, publications } = yield* makeTestRelay();
758 yield* Ref.set(
759 currentShell,
760 shell({
761 status: "completed",
762 latestRunCompletedAt: DateTime.add(yield* DateTime.now, { seconds: 1 }),
763 }),
764 );
765 yield* relay.publishThread(THREAD_ID);
766 assert.equal(publications.length, 0);
767 yield* TestClock.adjust("5 seconds");
768 yield* relay.drain;
769 assert.equal(publications[0]?.state?.phase, "completed");
770 yield* Ref.set(currentShell, null);
771 yield* relay.publishThread(THREAD_ID);
772 yield* secrets.set(PUBLISH_AGENT_ACTIVITY_SECRET, new TextEncoder().encode("false"));
773 yield* TestClock.adjust("5 seconds");
774 yield* relay.drain;
775 assert.equal(publications.length, 1);
776 }),
777 );
778 it.effect("publishes a held completion when its background item ends, not on tool output", () =>
779 Effect.gen(function* () {
780 const store = yield* ProjectionStore.ProjectionStoreV2;
781 const events = yield* Queue.unbounded<OrchestrationV2DomainEvent>();
782 // The relay pulls the next event only after it has handled the previous one.
783 const pulls = yield* Queue.unbounded<void>();
784 const { relay, shellReads, publications } = yield* makeTestRelay({
785 readShell: (threadId) => store.getThreadShell(threadId).pipe(Effect.orDie),
786 domainEvents: Stream.fromEffectRepeat(
787 Queue.offer(pulls, undefined).pipe(Effect.andThen(Queue.take(events))),
788 ),
789 });
790 yield* relay.start();
791 yield* Queue.take(pulls);
792 const deliver = Effect.fnUntraced(function* (event: OrchestrationV2DomainEvent) {
793 yield* store.apply(event);
794 yield* Queue.offer(events, event);
795 yield* Queue.take(pulls);
796 yield* relay.drain;
797 });
798
799 const now = yield* DateTime.now;
800 const runId = RunId.make("run:held-item");
801 const run = {
802 id: runId,
803 threadId: THREAD_ID,
804 ordinal: 1,
805 providerInstanceId: ProviderInstanceId.make("codex"),
806 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "test-model" },
807 providerThreadId: null,
808 userMessageId: MessageId.make("message:held-item"),
809 rootNodeId: null,
810 activeAttemptId: null,
811 status: "running" as const,
812 requestedAt: now,
813 startedAt: now,
814 completedAt: null,
815 checkpointId: null,
816 contextHandoffId: null,
817 };
818 const item = (
819 id: string,
820 fields: Pick<OrchestrationV2TurnItem, "status"> &
821 (
822 | { readonly type: "dynamic_tool"; readonly toolName: string; readonly input: unknown }
823 | {
824 readonly type: "assistant_message";
825 readonly messageId: MessageId;
826 readonly text: string;
827 readonly streaming: boolean;
828 }
829 ),
830 ): OrchestrationV2TurnItem => ({
831 id: TurnItemId.make(id),
832 threadId: THREAD_ID,
833 runId,
834 nodeId: NodeId.make("node:held-item"),
835 providerThreadId: null,
836 providerTurnId: null,
837 nativeItemRef: null,
838 parentItemId: null,
839 ordinal: 1,
840 title: null,
841 startedAt: now,
842 completedAt: null,
843 updatedAt: now,
844 ...fields,
845 });
846 const itemEvent = (id: string, payload: OrchestrationV2TurnItem) =>
847 ({
848 id: EventId.make(id),
849 type: "turn-item.updated",
850 threadId: THREAD_ID,
851 runId,
852 occurredAt: now,
853 payload,
854 }) satisfies OrchestrationV2DomainEvent;
855 const background = { type: "dynamic_tool" as const, toolName: "watch", input: {} };
856
857 yield* deliver({
858 id: EventId.make("event:held-item:thread"),
859 type: "thread.created",
860 threadId: THREAD_ID,
861 occurredAt: now,
862 payload: {
863 createdBy: "user",
864 creationSource: "web",
865 id: THREAD_ID,
866 projectId: PROJECT_ID,
867 title: "Thread",
868 providerInstanceId: ProviderInstanceId.make("codex"),
869 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "test-model" },
870 runtimeMode: "full-access",
871 interactionMode: "default",
872 branch: null,
873 worktreePath: null,
874 activeProviderThreadId: null,
875 lineage: { rootThreadId: THREAD_ID, parentThreadId: null, relationshipToParent: null },
876 forkedFrom: null,
877 createdAt: now,
878 updatedAt: now,
879 archivedAt: null,
880 settledOverride: null,
881 settledAt: null,
882 lastVisitedAt: null,
883 deletedAt: null,
884 },
885 });
886 yield* deliver({
887 id: EventId.make("event:held-item:run"),
888 type: "run.created",
889 threadId: THREAD_ID,
890 runId,
891 occurredAt: now,
892 payload: run,
893 });
894 yield* deliver(
895 itemEvent("event:held-item:tool", item("item:tool", { ...background, status: "running" })),
896 );
897 yield* deliver({
898 id: EventId.make("event:held-item:run-completed"),
899 type: "run.updated",
900 threadId: THREAD_ID,
901 runId,
902 occurredAt: now,
903 payload: { ...run, status: "completed", completedAt: now },
904 });
905 assert.deepEqual(
906 publications.map((publication) => publication.state?.phase),
907 ["running"],
908 );
909
910 // Streaming output never reaches the shell read.
911 const readsBeforeOutput = shellReads.length;
912 yield* deliver(
913 itemEvent(
914 "event:held-item:reply",
915 item("item:reply", {
916 type: "assistant_message",
917 messageId: MessageId.make("message:held-item:reply"),
918 text: "Still watching",
919 streaming: true,
920 status: "running",
921 }),
922 ),
923 );
924 yield* deliver(
925 itemEvent("event:held-item:tick", item("item:tool", { ...background, status: "running" })),
926 );
927 assert.equal(shellReads.length, readsBeforeOutput);
928
929 yield* deliver(
930 itemEvent("event:held-item:end", item("item:tool", { ...background, status: "completed" })),
931 );
932 assert.deepEqual(
933 publications.map((publication) => publication.state?.phase),
934 ["running", "completed"],
935 );
936 }).pipe(Effect.scoped, Effect.provide(ProjectionStore.layerMemory)),
937 );
938});
939
940describe("startup catch-up", { concurrent: false }, () => {
941 const link = (secrets: ServerSecretStore.ServerSecretStore["Service"]) =>
942 Effect.all(
943 [
944 secrets.set(RELAY_URL_SECRET, new TextEncoder().encode("https://relay.example.test")),
945 secrets.set(RELAY_ISSUER_SECRET, new TextEncoder().encode("https://relay.example.test")),
946 secrets.set(RELAY_ENVIRONMENT_CREDENTIAL_SECRET, new TextEncoder().encode("credential-1")),
947 ],
948 { discard: true },
949 );
950 const enablePublishing = (secrets: ServerSecretStore.ServerSecretStore["Service"]) =>
951 secrets.set(PUBLISH_AGENT_ACTIVITY_SECRET, new TextEncoder().encode("true"));
952 const linkChecks = (secretReads: ReadonlyArray<string>) =>
953 secretReads.filter((name) => name === RELAY_URL_SECRET).length;
954
955 it.effect("checks an unlinked environment once a minute and still catches up once linked", () =>
956 Effect.gen(function* () {
957 const { relay, secrets, secretReads, catchUp } = yield* makeTestRelay({ linked: false });
958 yield* enablePublishing(secrets);
959 yield* relay.start();
960
961 // Get past the backoff ramp, then count checks in a steady window.
962 yield* TestClock.adjust("10 minutes");
963 const checksBeforeWindow = linkChecks(secretReads);
964 yield* TestClock.adjust("10 minutes");
965 assert.equal(linkChecks(secretReads) - checksBeforeWindow, 10);
966 assert.equal(catchUp.shellSnapshotReads, 0);
967
968 yield* link(secrets);
969 yield* TestClock.adjust("1 minute");
970 assert.equal(catchUp.shellSnapshotReads, 1);
971 }).pipe(Effect.scoped),
972 );
973
974 it.effect("publishes at once when this process links while the check is backed off", () =>
975 Effect.gen(function* () {
976 const { relay, secrets, catchUp } = yield* makeTestRelay({ linked: false });
977 yield* enablePublishing(secrets);
978 yield* relay.start();
979
980 // Backed off to 60 s: the next check is still seconds away.
981 yield* TestClock.adjust("10 minutes");
982 yield* link(secrets);
983 yield* TestClock.adjust("1 second");
984 assert.equal(catchUp.shellSnapshotReads, 0);
985
986 yield* relay.requestCatchUp();
987 yield* TestClock.adjust("1 second");
988 assert.equal(catchUp.shellSnapshotReads, 1);
989 }).pipe(Effect.scoped),
990 );
991
992 it.effect("catches up within 5 s when another process enables publishing on a link", () =>
993 Effect.gen(function* () {
994 const { relay, secrets, catchUp } = yield* makeTestRelay({ linked: false });
995 yield* link(secrets);
996 yield* relay.start();
997
998 yield* TestClock.adjust("10 minutes");
999 assert.equal(catchUp.shellSnapshotReads, 0);
1000
1001 // `t3 connect publish` writes the opt-in without waking this process.
1002 yield* enablePublishing(secrets);
1003 yield* TestClock.adjust("5 seconds");
1004 assert.equal(catchUp.shellSnapshotReads, 1);
1005 }).pipe(Effect.scoped),
1006 );
1007});