From 11af2de8056815f60888f77c3355494f0f9d0bb7 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 16:34:12 +0000 Subject: [PATCH 1/2] test(server): a thread visit stores only read state Fails on main: thread.visited stores the whole thread snapshot, and a stale visit that changes nothing still writes an event (#24003). Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/runtimeLayer.test.ts | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index e62cd8725de92..e6b7ec2e944c8 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -9902,6 +9902,19 @@ it.layer(layerSharedApplicationDataPlaneTest)("visited projection", (it) => { assert.isNotNull(visited.thread.lastVisitedAt); assert.equal(DateTime.formatIso(visited.thread.lastVisitedAt!), visitedAt); assert.deepEqual(visited.thread.updatedAt, createdUpdatedAt); + const sql = yield* SqlClient.SqlClient; + const visitPayloadKeys = (commandId: string) => + sql<{ readonly payload_json: string }>` + SELECT payload_json FROM orchestration_events WHERE command_id = ${commandId} + `.pipe( + Effect.map((rows) => + rows.map((row) => Object.keys(JSON.parse(row.payload_json)).toSorted()), + ), + ); + // A visit stores read state, not a copy of the thread. + for (const keys of yield* visitPayloadKeys("runtime-layer-visited-thread-visit")) { + assert.includeMembers(["lastVisitedAt", "opens", "threadId"], keys); + } yield* orchestrator.dispatch({ type: "thread.visit", @@ -9911,6 +9924,8 @@ it.layer(layerSharedApplicationDataPlaneTest)("visited projection", (it) => { }); const afterStaleVisit = yield* orchestrator.getThreadProjection(threadId); assert.equal(DateTime.formatIso(afterStaleVisit.thread.lastVisitedAt!), visitedAt); + // A stale visit with no open changes nothing, so it stores nothing. + assert.deepEqual(yield* visitPayloadKeys("runtime-layer-visited-thread-visit-stale"), []); const shell = yield* orchestrator.getShellSnapshot(); const thread = shell.threads.find((candidate) => candidate.id === threadId); @@ -9927,7 +9942,6 @@ it.layer(layerSharedApplicationDataPlaneTest)("visited projection", (it) => { .pipe(Effect.flip); assert.instanceOf(markUnread, Orchestrator.OrchestratorDispatchError); - const sql = yield* SqlClient.SqlClient; yield* sql`INSERT INTO orchestration_v2_projection_turn_items ( turn_item_id, thread_id, ordinal, type, status, updated_at, payload_json ) VALUES ( From 39566288667701799202d9da79ea1703d9643253 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 16:44:20 +0000 Subject: [PATCH 2/2] fix(server): a thread visit stores only its read state Visiting a thread wrote a thread.visited event holding a full copy of the thread, linked pull requests included, even when the visit changed nothing. The orchestrator now emits thread.visit-recorded carrying only lastVisitedAt and opens, and emits nothing for a stale visit that is not an open. The projection store and client reducer fold the slim payload into the thread; older clients skip the unknown event type. Fixes pingdotgg/t3code#24003. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 12 ++++--- .../src/orchestration-v2/ProjectionStore.ts | 17 +++++++++- apps/server/src/relay/AgentAwarenessRelay.ts | 1 + .../src/state/orchestrationV2Projection.ts | 6 ++++ packages/contracts/src/orchestrationV2.ts | 33 +++++++++++++++++++ 5 files changed, 63 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 26482feb129fd..e78b473e5c4f2 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -3726,19 +3726,19 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio command.searchQuery, ) : null; - // Viewing a thread changes read state only. Loading its transcript (or - // bumping updatedAt) makes a routine read receipt scale with its history. + if (!movesForward && opens === null) return; + // Viewing a thread changes read state only. Loading its transcript, bumping + // updatedAt, or copying the thread into the event makes a read receipt scale with it. yield* emit( events, command, )({ - type: "thread.visited", + type: "thread.visit-recorded", threadId: command.threadId, providerInstanceId: thread.providerInstanceId, occurredAt, payload: { - ...thread, - ...(movesForward ? { lastVisitedAt: visitedAt.value } : {}), + lastVisitedAt: movesForward ? visitedAt.value : thread.lastVisitedAt, ...(opens ? { opens } : {}), }, }); @@ -14991,6 +14991,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio planned.events.length > 0 || isUnobservedAcknowledgement(command) || command.type === "thread.background-work.settle" || + // A stale visit changes no read state. + command.type === "thread.visit" || // A read can answer a watch again before a push's reopen lands; nothing is left to clear. command.type === "thread.pull-request-watch.checks-reopen" || command.type === "thread.interrupt.unresponsive" || diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 721006481e83e..e90a82786953f 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1,5 +1,5 @@ import type { ProjectThreadAwarenessV2Input } from "@t3tools/shared/agentAwareness"; -import { pendingGuestMessageCount } from "@t3tools/contracts"; +import { applyOrchestrationV2ThreadVisit, pendingGuestMessageCount } from "@t3tools/contracts"; import { applyThreadPullRequestLinkUpdate } from "../sourceControl/threadPullRequests.ts"; import { projectTurnItemForWire } from "./WireProjection.ts"; import * as Stream from "effect/Stream"; @@ -927,6 +927,11 @@ export function applyToProjection( ...projection, thread: event.payload, }; + case "thread.visit-recorded": + return { + ...projection, + thread: applyOrchestrationV2ThreadVisit(projection.thread, event.payload), + }; case "run.created": case "run.updated": return withLocalVisibleTurnItems({ @@ -2242,6 +2247,15 @@ export const layer: Layer.Layer = }); break; } + case "thread.visit-recorded": { + const thread = yield* getThread(event.threadId); + yield* apply({ + ...event, + type: "thread.visited", + payload: applyOrchestrationV2ThreadVisit(thread, event.payload), + }); + break; + } case "thread.created": case "thread.archived": case "thread.unarchived": @@ -3074,6 +3088,7 @@ export const layer: Layer.Layer = event.type !== "thread.active-reordered" && event.type !== "thread.tags-updated" && event.type !== "thread.visited" && + event.type !== "thread.visit-recorded" && event.type !== "thread.marked-unread" && event.type !== "thread.metadata-updated" && event.type !== "thread.pull-request-synced" && diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index 73b52571e7416..76f854548deb6 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -117,6 +117,7 @@ function shouldPublishAgentAwarenessEvent( case "thread.active-reordered": case "thread.tags-updated": case "thread.visited": + case "thread.visit-recorded": case "thread.marked-unread": case "thread.runtime-mode-updated": case "thread.interaction-mode-updated": diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.ts b/packages/client-runtime/src/state/orchestrationV2Projection.ts index 1bc8397fc2bd7..76d1c9d6b9c15 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.ts @@ -1,4 +1,5 @@ import { applyThreadPullRequestLinkUpdate } from "../threadPullRequests.ts"; +import { applyOrchestrationV2ThreadVisit } from "@t3tools/contracts"; import type { OrchestrationV2DomainEvent, OrchestrationV2Run, @@ -321,6 +322,11 @@ export function applyOrchestrationV2ProjectionEvent( case "thread.tags-updated": case "thread.marked-unread": return { ...projection, thread: event.payload }; + case "thread.visit-recorded": + return { + ...projection, + thread: applyOrchestrationV2ThreadVisit(projection.thread, event.payload), + }; case "run.created": case "run.updated": { const next = { ...base, runs: upsertEntity(base.runs, event.payload) }; diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 35ac16fc063a4..42868af7da957 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -540,6 +540,24 @@ export const OrchestrationV2AppThread = Schema.Struct({ }); export type OrchestrationV2AppThread = typeof OrchestrationV2AppThread.Type; +/** A visit changes read state only, so its event leaves the rest of the thread out. */ +export const OrchestrationV2ThreadVisit = Schema.Struct({ + lastVisitedAt: OrchestrationV2AppThread.fields.lastVisitedAt, + opens: OrchestrationV2AppThread.fields.opens, +}); +export type OrchestrationV2ThreadVisit = typeof OrchestrationV2ThreadVisit.Type; + +export function applyOrchestrationV2ThreadVisit( + thread: Thread, + visit: OrchestrationV2ThreadVisit, +): Thread { + return { + ...thread, + lastVisitedAt: visit.lastVisitedAt, + ...(visit.opens === undefined ? {} : { opens: visit.opens }), + }; +} + /** A thread another agent drives: a local subagent or a task delegated from a linked environment. */ export const isDelegatedAppThread = ( thread: Pick, @@ -1973,6 +1991,11 @@ export const OrchestrationV2DomainEvent = Schema.Union([ type: Schema.Literal("thread.pull-request-link-updated"), payload: ThreadPullRequestLink, }), + Schema.Struct({ + ...OrchestrationV2EventBase.fields, + type: Schema.Literal("thread.visit-recorded"), + payload: OrchestrationV2ThreadVisit, + }), Schema.Struct({ ...OrchestrationV2EventBase.fields, type: Schema.Literal("run.created"), @@ -2406,6 +2429,11 @@ export const OrchestrationV2AppThreadJson = OrchestrationV2AppThread.mapFields(( })); export type OrchestrationV2AppThreadJson = typeof OrchestrationV2AppThreadJson.Type; +export const OrchestrationV2ThreadVisitJson = Schema.Struct({ + lastVisitedAt: OrchestrationV2AppThreadJson.fields.lastVisitedAt, + opens: OrchestrationV2AppThreadJson.fields.opens, +}); + export const OrchestrationV2RunJson = OrchestrationV2Run.mapFields((fields) => ({ ...fields, requestedAt: Schema.DateTimeUtcFromString, @@ -2906,6 +2934,11 @@ export const OrchestrationV2DomainEventJson = Schema.Union([ type: Schema.Literal("thread.pull-request-link-updated"), payload: ThreadPullRequestLink, }), + Schema.Struct({ + ...OrchestrationV2JsonEventBaseFields, + type: Schema.Literal("thread.visit-recorded"), + payload: OrchestrationV2ThreadVisitJson, + }), Schema.Struct({ ...OrchestrationV2JsonEventBaseFields, type: Schema.Literal("run.created"),