From 5a0b8c6d75927c66e0bc27cd5220932ce0c80bc8 Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Tue, 3 Nov 2026 20:27:31 -0600 Subject: [PATCH 1/2] perf(server): narrow agent awareness startup reads - Add a focused projection snapshot for agent awareness. - Avoid loading full thread shell data during relay catch-up. --- .../src/orchestration-v2/Orchestrator.ts | 13 +++ .../orchestration-v2/ProjectionStore.test.ts | 44 ++++++++- .../src/orchestration-v2/ProjectionStore.ts | 97 ++++++++++++++++++- .../ProviderTurnControlService.test.ts | 1 + .../ThreadManagementService.ts | 2 + .../src/relay/AgentAwarenessRelay.test.ts | 36 ++++--- apps/server/src/relay/AgentAwarenessRelay.ts | 25 ++--- 7 files changed, 184 insertions(+), 34 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index f20ca8bdc92c1..a5fcbc46d36ed 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -444,6 +444,12 @@ export interface OrchestratorV2Shape { Effect.Success>, OrchestratorV2Error >; + readonly getAgentAwarenessSnapshot: ( + threadId?: ThreadId, + ) => Effect.Effect< + Effect.Success>, + OrchestratorV2Error + >; readonly getShellSnapshot: ( options?: ShellSnapshotOptions, ) => Effect.Effect; @@ -14188,6 +14194,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio .getPullRequestSnapshot(options) .pipe(Effect.mapError(shellProjectionError)); }), + getAgentAwarenessSnapshot: Effect.fn("OrchestratorV2.getAgentAwarenessSnapshot")( + function (threadId) { + return projectionStore + .getAgentAwarenessSnapshot(threadId) + .pipe(Effect.mapError(shellProjectionError)); + }, + ), getShellSnapshot: Effect.fn("OrchestratorV2.getShellSnapshot")(function (options) { return projectionStore.getShellSnapshot(options).pipe(Effect.mapError(shellProjectionError)); }), diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 14bb72e3c78fb..9d412e498c482 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -28,6 +28,7 @@ import { TurnItemId, type ThreadPullRequestLink, isLiveSubagentShell, + isListedThreadLineage, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -96,8 +97,40 @@ const traceSqlStatements = () => { /** The shell index flags every live shell and matches its shells on the fields subscriber trims read. */ const assertShellIndexCoversShells = Effect.fnUntraced(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; - const shells = (yield* store.getShellSnapshot({ location: "active" })).threads; + const snapshot = yield* store.getShellSnapshot(); + const shells = snapshot.threads; const index = yield* store.listActiveShellIndex(); + const { tracer, statements, queryTransactions } = traceSqlStatements(); + const activity = yield* store.getAgentAwarenessSnapshot().pipe(Effect.withTracer(tracer)); + assert.equal(activity.snapshotSequence, snapshot.snapshotSequence); + const awareness = (thread: Parameters[0]["thread"]) => + projectThreadAwarenessV2({ + environmentId: EnvironmentId.make("test"), + project: { title: "Test" }, + thread, + }); + assert.deepEqual( + activity.threads.map(awareness), + shells.filter((thread) => isListedThreadLineage(thread.lineage)).map(awareness), + ); + assert.deepEqual( + activity.threads.map((thread) => [thread.id, thread.projectId, thread.latestRunCompletedAt]), + shells + .filter((thread) => isListedThreadLineage(thread.lineage)) + .map((thread) => [thread.id, thread.projectId, thread.latestRunCompletedAt]), + ); + assert.isFalse(statements.some((statement) => /COUNT\(\*\)|GROUP BY/i.test(statement))); + assert.isTrue(queryTransactions.every((transaction) => transaction !== null)); + assert.equal(new Set(queryTransactions).size, 1); + for (const shell of [...shells, ...snapshot.archivedThreads]) { + const targeted = yield* store + .getAgentAwarenessSnapshot(shell.id) + .pipe(Effect.withTracer(tracer)); + assert.deepEqual(targeted.threads.map(awareness), [awareness(shell)]); + assert.deepEqual(targeted.threads[0]?.archivedAt, shell.archivedAt); + } + assert.isFalse(statements.some((statement) => /COUNT\(\*\)|GROUP BY/i.test(statement))); + const iso = (value: DateTime.Utc | null | undefined) => value == null ? null : DateTime.formatIso(value); const trimmed = ( @@ -250,6 +283,7 @@ it.effect("routes full projection reads to readers and retains uncommitted write }, }); const { queries, trace } = traceSqlRouting(); + yield* store.getAgentAwarenessSnapshot().pipe(trace); const projection = yield* store.getThreadProjection(threadId).pipe(trace); assert.equal(projection.thread.id, threadId); assert.deepEqual( @@ -309,6 +343,12 @@ it.effect("routes full projection reads to readers and retains uncommitted write SET payload_json = json_set(payload_json, '$.title', 'Uncommitted title') WHERE thread_id = ${threadId} `; + assert.equal( + (yield* store.getAgentAwarenessSnapshot()).threads.find( + (thread) => thread.id === threadId, + )?.title, + "Uncommitted title", + ); assert.equal( (yield* store.getThreadProjection(threadId)).thread.title, "Uncommitted title", @@ -8013,6 +8053,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }); const project = { title: "Project" }; const environmentId = EnvironmentId.make("environment:held-completion"); + yield* assertShellIndexCoversShells(); const held = yield* store.getThreadShell(threadId); assert.equal( held && projectThreadAwarenessV2({ environmentId, project, thread: held })?.phase, @@ -8028,6 +8069,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { occurredAt: releasedAt, payload: { ...providerThread, pendingBackgroundTasks: [], updatedAt: releasedAt }, }); + yield* assertShellIndexCoversShells(); const released = yield* store.getThreadShell(threadId); const state = released && projectThreadAwarenessV2({ environmentId, project, thread: released }); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index a46c8017731d5..5d03be7753d55 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1,3 +1,4 @@ +import type { ProjectThreadAwarenessV2Input } from "@t3tools/shared/agentAwareness"; import { pendingGuestMessageCount } from "@t3tools/contracts"; import { applyThreadPullRequestLinkUpdate } from "../sourceControl/threadPullRequests.ts"; import { projectTurnItemForWire } from "./WireProjection.ts"; @@ -491,6 +492,12 @@ export interface ProjectionStoreV2Shape { }, ProjectionStoreV2Error >; + readonly getAgentAwarenessSnapshot: ( + threadId?: ThreadId, + ) => Effect.Effect< + { readonly snapshotSequence: number; readonly threads: ReadonlyArray }, + ProjectionStoreV2Error + >; readonly getShellSnapshot: ( options?: ShellSnapshotOptions, ) => Effect.Effect; @@ -980,6 +987,17 @@ type ShellPendingTurnItemRow = PayloadRow & { readonly owning_provider_instance_id: string | null; }; +export type AgentAwarenessThread = ProjectThreadAwarenessV2Input["thread"] & + Pick< + OrchestrationV2ThreadShell, + | "projectId" + | "latestRunCompletedAt" + | "archivedAt" + | "activeRunId" + | "latestRunId" + | "hasActionableProposedPlan" + >; + type ShellThreadRow = { readonly thread_id: string; readonly payload_json: string; @@ -6077,7 +6095,8 @@ export const layer: Layer.Layer = familyOf, worktreePath, threadIds, - }: ShellSnapshotOptions = {}, + listedOnly = false, + }: ShellSnapshotOptions & { readonly listedOnly?: boolean } = {}, ) => sql` SELECT @@ -6259,7 +6278,12 @@ export const layer: Layer.Layer = candidate.ordinal DESC, candidate.run_id DESC LIMIT 1 ) AND blocked.status = 'failed' - WHERE t.deleted_at IS NULL${threadId === undefined ? sql`` : sql` AND t.thread_id = ${threadId}`}${ + WHERE t.deleted_at IS NULL${ + listedOnly + ? sql` AND json_extract(t.payload_json, '$.lineage.relationshipToParent') IS NOT 'subagent' + AND json_extract(t.payload_json, '$.lineage.sideChat') IS NOT 1` + : sql`` + }${threadId === undefined ? sql`` : sql` AND t.thread_id = ${threadId}`}${ projectId === undefined ? sql`` : sql` AND t.project_id = ${projectId}` }${ location === "active" @@ -6990,6 +7014,74 @@ export const layer: Layer.Layer = return { snapshotSequence: sequenceRows[0]?.snapshot_sequence ?? 0, threads }; }, Effect.mapError(shellSnapshotReadError)); + const getAgentAwarenessSnapshot = Effect.fn("ProjectionStore.getAgentAwarenessSnapshot")( + function* (threadId?: ThreadId) { + const { threadRows, providerThreadRows, pendingTurnItemRows, sequenceRows } = + yield* sql.withTransaction( + Effect.gen(function* () { + const threadRows = yield* selectShellThreadRows( + threadId, + threadId === undefined ? { location: "active", listedOnly: true } : {}, + ); + const ids = threadRows.map((row) => ThreadId.make(row.thread_id)); + const providerThreadRows = + ids.length === 0 ? [] : yield* selectShellProviderThreadRows(ids); + const pendingTurnItemRows = + ids.length === 0 + ? [] + : yield* selectShellPendingTurnItemRows( + ids, + threadRows.flatMap((row) => + row.active_run_id !== null && row.active_run_status === "running" + ? [row.active_run_id] + : [], + ), + ); + const sequenceRows = yield* sql<{ readonly snapshot_sequence: number | null }>` + SELECT MAX(sequence) AS snapshot_sequence FROM orchestration_events + WHERE application_event_version = 2 AND aggregate_kind = 'thread' + `; + return { threadRows, providerThreadRows, pendingTurnItemRows, sequenceRows }; + }), + ); + const { providerThreadsByThreadId, pendingTurnItemsByThreadId } = + yield* pendingBackgroundDataByThreadId({ providerThreadRows, pendingTurnItemRows }); + const threads = yield* Effect.forEach(threadRows, (row) => + shellThreadStateFromRow({ + row, + // Awareness never consumes timeline counts or inherited fork history. + runOrdinalsByThreadId: new Map(), + itemCountsByThreadId: new Map(), + shellItemCountsByThreadId: new Map(), + providerThreadsByThreadId, + pendingTurnItemsByThreadId, + }).pipe( + Effect.map((state): AgentAwarenessThread => ({ + id: state.thread.id, + projectId: state.thread.projectId, + archivedAt: state.thread.archivedAt, + activeRunId: state.activeRunId, + latestRunId: state.latestRunId, + hasActionableProposedPlan: state.hasActionableProposedPlan, + title: state.thread.title, + lineage: state.thread.lineage, + modelSelection: state.thread.modelSelection, + settledOverride: state.thread.settledOverride, + updatedAt: state.updatedAt, + status: state.latestRunStatus, + activityRunStatus: state.activityRunStatus, + latestRunCompletedAt: state.latestRunCompletedAt, + latestRunStopRequested: state.latestRunStopRequested, + pendingRuntimeRequest: state.pendingRuntimeRequest, + pendingBackgroundTasks: state.pendingBackgroundTasks, + })), + ), + ); + return { snapshotSequence: sequenceRows[0]?.snapshot_sequence ?? 0, threads }; + }, + Effect.mapError(shellSnapshotReadError), + ); + const getShellSnapshot: ProjectionStoreV2Shape["getShellSnapshot"] = (options) => sql .withTransaction(readShellSnapshotRows(options)) @@ -7257,6 +7349,7 @@ export const layer: Layer.Layer = return { apply, + getAgentAwarenessSnapshot: readOnReader(getAgentAwarenessSnapshot), getShellSnapshot: readOnReader(getShellSnapshot), getPullRequestSnapshot: readOnReader(getPullRequestSnapshot), listActiveSubagentThreadIds: readOnReader(listActiveSubagentThreadIds), diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index bc5656094445f..63cb35749b4d4 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -212,6 +212,7 @@ it.effect( getLimitRecoveryCandidates: () => Effect.die("unused getLimitRecoveryCandidates"), getLimitRecoveryWakeAt: () => Effect.die("unused getLimitRecoveryWakeAt"), getOpenRemoteDelegatedTasks: Effect.die("unused getOpenRemoteDelegatedTasks"), + getAgentAwarenessSnapshot: () => Effect.die("unused awareness snapshot"), getShellSnapshot: () => Effect.die("unused getShellSnapshot"), listActiveSubagentThreadIds: () => Effect.die("unused listActiveSubagentThreadIds"), countThreads: () => Effect.die("unused countThreads"), diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index f0346b2bec4ec..ed2935fedbf12 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -402,6 +402,7 @@ export interface ThreadManagementServiceShape { ProjectionRecords, ThreadManagementProjectionLoadError | ThreadManagementThreadNotFoundError >; + readonly getAgentAwarenessSnapshot: Orchestrator.OrchestratorV2["Service"]["getAgentAwarenessSnapshot"]; readonly getShellSnapshot: ( options?: Pick< ShellSnapshotOptions, @@ -1520,6 +1521,7 @@ const make = Effect.gen(function* () { getThreadTranscript, getThreadSnapshotWindow, getProjectThreadRecords, + getAgentAwarenessSnapshot: orchestrator.getAgentAwarenessSnapshot, getShellSnapshot: orchestrator.getShellSnapshot, listActiveSubagentThreadIds: orchestrator.listActiveSubagentThreadIds, countThreads: orchestrator.countThreads, diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 87838fd5a51ed..f9b67df87f683 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -178,17 +178,23 @@ const makeTestRelay = Effect.fnUntraced(function* ( }); const currentShell = yield* Ref.make(shell()); const shellReads: ThreadId[] = []; - const catchUp = { shellSnapshotReads: 0 }; + const catchUp = { activitySnapshotReads: 0 }; const threads = ThreadManagementService.ThreadManagementService.of({ - getThreadShell: (threadId) => - Effect.sync(() => shellReads.push(threadId)).pipe( - Effect.andThen(options.readShell?.(threadId) ?? Ref.get(currentShell)), - ), - getShellSnapshot: () => - Effect.sync(() => { - catchUp.shellSnapshotReads += 1; - return { schemaVersion: 2, snapshotSequence: 1, threads: [], archivedThreads: [] }; - }), + getThreadShell: unused, + getShellSnapshot: unused, + getAgentAwarenessSnapshot: (threadId) => + threadId === undefined + ? Effect.sync(() => { + catchUp.activitySnapshotReads += 1; + return { snapshotSequence: 1, threads: [] }; + }) + : Effect.sync(() => shellReads.push(threadId)).pipe( + Effect.andThen(options.readShell?.(threadId) ?? Ref.get(currentShell)), + Effect.map((thread) => ({ + snapshotSequence: 1, + threads: thread === null ? [] : [thread], + })), + ), listActiveSubagentThreadIds: unused, listAppOwnedChildThreadIds: unused, countThreads: unused, @@ -834,21 +840,21 @@ it.effect("catches up after startup without polling after success", () => const { relay, secrets, secretReads, catchUp } = yield* makeTestRelay(); yield* relay.start(); yield* TestClock.adjust("1 second"); - assert.equal(catchUp.shellSnapshotReads, 1); + assert.equal(catchUp.activitySnapshotReads, 1); yield* secrets.set(PUBLISH_AGENT_ACTIVITY_SECRET, new TextEncoder().encode("false")); yield* relay.requestCatchUp(); yield* TestClock.adjust("1 second"); - assert.equal(catchUp.shellSnapshotReads, 1); + assert.equal(catchUp.activitySnapshotReads, 1); yield* secrets.set(PUBLISH_AGENT_ACTIVITY_SECRET, new TextEncoder().encode("true")); yield* relay.requestCatchUp(); yield* TestClock.adjust("1 second"); - assert.equal(catchUp.shellSnapshotReads, 2); + assert.equal(catchUp.activitySnapshotReads, 2); const readsBeforeIdle = secretReads.length; yield* TestClock.adjust("10 minutes"); - assert.equal(catchUp.shellSnapshotReads, 2); + assert.equal(catchUp.activitySnapshotReads, 2); assert.equal(secretReads.length, readsBeforeIdle); yield* secrets.set( @@ -857,6 +863,6 @@ it.effect("catches up after startup without polling after success", () => ); yield* relay.requestCatchUp(); yield* TestClock.adjust("1 second"); - assert.equal(catchUp.shellSnapshotReads, 3); + assert.equal(catchUp.activitySnapshotReads, 3); }).pipe(Effect.scoped), ); diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index 6a235911d4f91..6e9175862e47d 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -1,8 +1,8 @@ +import type { AgentAwarenessThread } from "../orchestration-v2/ProjectionStore.ts"; import { isListedThreadLineage, type EnvironmentId, type OrchestrationV2DomainEvent, - type OrchestrationV2ThreadShell, type OrchestrationV2TurnItem, type Project, type ThreadId, @@ -283,7 +283,7 @@ const makePublishProof = Effect.fn("makePublishProof")(function* (input: { }); function describeThreadShellForAwareness( - thread: Option.Option, + thread: Option.Option, ): Record { if (Option.isNone(thread)) { return { found: false }; @@ -302,7 +302,7 @@ function describeThreadShellForAwareness( function resolveAgentAwarenessRelayPublishSnapshot(input: { readonly environmentId: EnvironmentId; - readonly thread: Option.Option; + readonly thread: Option.Option; readonly project: Option.Option; }): { readonly projectId: string | null; @@ -336,7 +336,7 @@ function resolveAgentAwarenessRelayPublishSnapshot(input: { }; } -function terminalWorkSinceStart(thread: OrchestrationV2ThreadShell, startedAt: number): boolean { +function terminalWorkSinceStart(thread: AgentAwarenessThread, startedAt: number): boolean { return ( thread.latestRunCompletedAt != null && DateTime.toEpochMillis(thread.latestRunCompletedAt) > startedAt @@ -347,7 +347,7 @@ export function resolveAgentAwarenessRelayActiveThreadIds(input: { readonly environmentId: EnvironmentId; readonly startedAt: number; readonly projects: ReadonlyArray>; - readonly threads: ReadonlyArray; + readonly threads: ReadonlyArray; }): ReadonlyArray { const projectById = new Map(input.projects.map((project) => [project.id, project])); return input.threads @@ -524,7 +524,7 @@ export const make = Effect.gen(function* () { }); // Read only this thread; full shell reads would make each activity event scale with all threads. - const threadShell = yield* threads.getThreadShell(threadId); + const threadShell = (yield* threads.getAgentAwarenessSnapshot(threadId)).threads[0] ?? null; if ( threadShell !== null && !isListedThreadLineage(threadShell.lineage) && @@ -535,7 +535,7 @@ export const make = Effect.gen(function* () { } const thread = threadShell === null || threadShell.archivedAt !== null - ? Option.none() + ? Option.none() : Option.some(threadShell); const project = Option.isSome(thread) ? yield* projects.getById(thread.value.projectId) @@ -735,19 +735,12 @@ export const make = Effect.gen(function* () { } const environmentId = yield* serverEnvironment.getEnvironmentId; const projectSnapshot = yield* projects.snapshot; - // Subagents and side chats never publish awareness, and subagents are most threads. - const topLevelThreadIds = (yield* threads.listActiveShellIndex()).flatMap((thread) => - isListedThreadLineage(thread.lineage) ? [thread.id] : [], - ); - const shellSnapshot = yield* threads.getShellSnapshot({ - location: "active", - threadIds: topLevelThreadIds, - }); + const activitySnapshot = yield* threads.getAgentAwarenessSnapshot(); const activeThreadIds = resolveAgentAwarenessRelayActiveThreadIds({ environmentId, startedAt, projects: projectSnapshot.projects, - threads: shellSnapshot.threads, + threads: activitySnapshot.threads, }); if (activeThreadIds.length === 0) { yield* Effect.logDebug("agent activity snapshot has no publishable threads"); From 98204d9c26ed5f8e75cc9cc38b9727f66cd6cac6 Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Tue, 3 Nov 2026 20:42:20 -0600 Subject: [PATCH 2/2] test(server): verify awareness catch-up publishes active threads --- .../src/relay/AgentAwarenessRelay.test.ts | 42 ++++++++++++------- 1 file changed, 28 insertions(+), 14 deletions(-) diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index f9b67df87f683..8747ad7e3d86e 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -144,6 +144,7 @@ const makeTestRelay = Effect.fnUntraced(function* ( readonly linked?: boolean; readonly readShell?: (threadId: ThreadId) => Effect.Effect; readonly domainEvents?: Stream.Stream; + readonly catchUpThreads?: ReadonlyArray; } = {}, ) { const values = new Map( @@ -186,7 +187,7 @@ const makeTestRelay = Effect.fnUntraced(function* ( threadId === undefined ? Effect.sync(() => { catchUp.activitySnapshotReads += 1; - return { snapshotSequence: 1, threads: [] }; + return { snapshotSequence: 1, threads: options.catchUpThreads ?? [] }; }) : Effect.sync(() => shellReads.push(threadId)).pipe( Effect.andThen(options.readShell?.(threadId) ?? Ref.get(currentShell)), @@ -264,6 +265,16 @@ const makeTestRelay = Effect.fnUntraced(function* ( }, { preconnect: () => {} }, ); + const project = { + id: PROJECT_ID, + title: "Project", + workspaceRoot: "/workspace", + defaultModelSelection: null, + scripts: [], + createdAt: NOW, + updatedAt: NOW, + deletedAt: null, + }; const relay = yield* AgentAwarenessRelay.make.pipe( Effect.provideService(ServerSecretStore.ServerSecretStore, secrets), Effect.provideService(ThreadManagementService.ThreadManagementService, threads), @@ -277,21 +288,11 @@ const makeTestRelay = Effect.fnUntraced(function* ( update: unused, delete: unused, getByWorkspaceRoot: unused, - snapshot: Effect.succeed({ projects: [] } as never), + snapshot: Effect.succeed({ projects: [project], updatedAt: NOW }), listPage: unused, getShell: unused, listShells: unused, - getById: () => - Effect.succeedSome({ - id: PROJECT_ID, - title: "Project", - workspaceRoot: "/workspace", - defaultModelSelection: null, - scripts: [], - createdAt: NOW, - updatedAt: NOW, - deletedAt: null, - }), + getById: () => Effect.succeedSome(project), }), Effect.provideService(FetchHttpClient.Fetch, fetch), Effect.provide(NodeCrypto.layer), @@ -837,10 +838,20 @@ describe("AgentAwarenessRelay", () => { it.effect("catches up after startup without polling after success", () => Effect.gen(function* () { - const { relay, secrets, secretReads, catchUp } = yield* makeTestRelay(); + const { relay, secrets, secretReads, catchUp, publications } = yield* makeTestRelay({ + catchUpThreads: [shell()], + }); yield* relay.start(); yield* TestClock.adjust("1 second"); + yield* relay.drain; assert.equal(catchUp.activitySnapshotReads, 1); + assert.equal(publications.length, 1); + assert.include(publications[0]?.state, { + threadId: THREAD_ID, + projectTitle: "Project", + threadTitle: "Thread", + phase: "running", + }); yield* secrets.set(PUBLISH_AGENT_ACTIVITY_SECRET, new TextEncoder().encode("false")); yield* relay.requestCatchUp(); @@ -863,6 +874,9 @@ it.effect("catches up after startup without polling after success", () => ); yield* relay.requestCatchUp(); yield* TestClock.adjust("1 second"); + yield* relay.drain; assert.equal(catchUp.activitySnapshotReads, 3); + assert.equal(publications.length, 2); + assert.equal(publications[1]?.state?.threadId, THREAD_ID); }).pipe(Effect.scoped), );