diff --git a/apps/server/src/orchestration/ActivityPayloadProjection.ts b/apps/server/src/orchestration/ActivityPayloadProjection.ts index 5fd0cc8984b..67896961b38 100644 --- a/apps/server/src/orchestration/ActivityPayloadProjection.ts +++ b/apps/server/src/orchestration/ActivityPayloadProjection.ts @@ -201,6 +201,52 @@ export function projectActivityPayload( }; } +/** + * Matches the validity rule in the web client's + * `deriveLatestContextWindowSnapshot`: rows without a finite, non-negative + * `usedTokens` are skipped during its backward walk, so they must not shadow + * an earlier resolvable row here. + */ +function isResolvableContextWindowActivity(activity: OrchestrationThreadActivity): boolean { + if (activity.kind !== "context-window.updated") { + return false; + } + const payload = asRecord(activity.payload); + const usedTokens = payload?.usedTokens; + return typeof usedTokens === "number" && Number.isFinite(usedTokens) && usedTokens >= 0; +} + +/** + * Drops all but the last resolvable context-window activity per turn from a + * snapshot. Clients only ever read the latest usage value (walking the array + * backwards), so shipping the full history — often thousands of rows on long + * threads — buys nothing. Retention is per turn rather than per thread because + * a live `thread.reverted` makes the client discard whole turns; keeping each + * turn's latest row means the meter can still resolve a value from the turns + * that survive. Malformed rows pass through untouched rather than shadowing a + * valid earlier row. Live `thread.activity-appended` events are untouched: + * newer updates still stream through and supersede the retained rows on the + * client. + */ +function dropStaleContextWindowActivities( + activities: ReadonlyArray, +): ReadonlyArray { + const latestIndexByTurn = new Map(); + for (let index = 0; index < activities.length; index += 1) { + if (isResolvableContextWindowActivity(activities[index]!)) { + latestIndexByTurn.set(activities[index]!.turnId, index); + } + } + if (latestIndexByTurn.size === 0) { + return activities; + } + return activities.filter( + (activity, index) => + !isResolvableContextWindowActivity(activity) || + latestIndexByTurn.get(activity.turnId) === index, + ); +} + export function projectThreadDetailSnapshot( snapshot: OrchestrationThreadDetailSnapshot, ): OrchestrationThreadDetailSnapshot { @@ -208,7 +254,9 @@ export function projectThreadDetailSnapshot( ...snapshot, thread: { ...snapshot.thread, - activities: snapshot.thread.activities.map(projectActivityPayload), + activities: dropStaleContextWindowActivities(snapshot.thread.activities).map( + projectActivityPayload, + ), }, }; } diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 30e14770de5..65eee28e59b 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -5690,15 +5690,6 @@ it.layer(NodeServices.layer)("server router seam", (it) => { ), ); assert.equal(fullDiffResult.diff, "full-diff"); - - const replayResult = yield* Effect.scoped( - withWsRpcClient(wsUrl, (client) => - client[ORCHESTRATION_WS_METHODS.replayEvents]({ - fromSequenceExclusive: 0, - }), - ), - ); - assert.deepEqual(replayResult, []); }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); @@ -6361,73 +6352,6 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); - it.effect("enriches replayed project events with repository identity metadata", () => - Effect.gen(function* () { - const repositoryIdentity = { - canonicalKey: "github.com/t3tools/t3code", - locator: { - source: "git-remote" as const, - remoteName: "origin", - remoteUrl: "git@github.com:T3Tools/t3code.git", - }, - displayName: "T3Tools/t3code", - provider: "github", - owner: "T3Tools", - name: "t3code", - }; - - yield* buildAppUnderTest({ - layers: { - orchestrationEngine: { - readEvents: (_fromSequenceExclusive) => - Stream.make({ - sequence: 1, - eventId: EventId.make("event-1"), - aggregateKind: "project", - aggregateId: defaultProjectId, - occurredAt: "2026-04-05T00:00:00.000Z", - commandId: null, - causationEventId: null, - correlationId: null, - metadata: {}, - type: "project.created", - payload: { - projectId: defaultProjectId, - title: "Default Project", - workspaceRoot: "/tmp/default-project", - defaultModelSelection, - scripts: [], - createdAt: "2026-04-05T00:00:00.000Z", - updatedAt: "2026-04-05T00:00:00.000Z", - }, - } satisfies Extract), - }, - repositoryIdentityResolver: { - resolve: () => Effect.succeed(repositoryIdentity), - }, - }, - }); - - const wsUrl = yield* getWsServerUrl("/ws"); - const replayResult = yield* Effect.scoped( - withWsRpcClient(wsUrl, (client) => - client[ORCHESTRATION_WS_METHODS.replayEvents]({ - fromSequenceExclusive: 0, - }), - ), - ); - - const replayedEvent = replayResult[0]; - assert.equal(replayedEvent?.type, "project.created"); - assert.deepEqual( - replayedEvent && replayedEvent.type === "project.created" - ? replayedEvent.payload.repositoryIdentity - : null, - repositoryIdentity, - ); - }).pipe(Effect.provide(NodeHttpServer.layerTest)), - ); - it.effect("stops the provider session and closes thread terminals after archive", () => Effect.gen(function* () { const threadId = ThreadId.make("thread-archive"); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 2a8be25a728..744e48661bf 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -46,7 +46,6 @@ import { ProjectWriteFileError, RelayClientInstallFailedError, type RelayClientInstallProgressEvent, - OrchestrationReplayEventsError, type FilesystemBrowseFailure, FilesystemBrowseError, AssetWorkspaceContextNotFoundError, @@ -60,7 +59,6 @@ import { WS_METHODS, WsRpcGroup, } from "@t3tools/contracts"; -import { clamp } from "effect/Number"; import { HttpRouter, HttpServerRequest, HttpServerRespondable } from "effect/unstable/http"; import { RpcSerialization, RpcServer } from "effect/unstable/rpc"; @@ -99,7 +97,6 @@ import * as VcsProvisioningService from "./vcs/VcsProvisioningService.ts"; import * as GitWorkflowService from "./git/GitWorkflowService.ts"; import * as ReviewService from "./review/ReviewService.ts"; import * as ProjectSetupScriptRunner from "./project/ProjectSetupScriptRunner.ts"; -import * as RepositoryIdentityResolver from "./project/RepositoryIdentityResolver.ts"; import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; import * as ProcessDiagnostics from "./diagnostics/ProcessDiagnostics.ts"; @@ -302,7 +299,6 @@ const RPC_REQUIRED_SCOPE = new Map([ [ORCHESTRATION_WS_METHODS.dispatchCommand, AuthOrchestrationOperateScope], [ORCHESTRATION_WS_METHODS.getTurnDiff, AuthOrchestrationReadScope], [ORCHESTRATION_WS_METHODS.getFullThreadDiff, AuthOrchestrationReadScope], - [ORCHESTRATION_WS_METHODS.replayEvents, AuthOrchestrationReadScope], [ORCHESTRATION_WS_METHODS.subscribeShell, AuthOrchestrationReadScope], [ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot, AuthOrchestrationReadScope], [ORCHESTRATION_WS_METHODS.subscribeThread, AuthOrchestrationReadScope], @@ -441,8 +437,6 @@ const makeWsRpcLayer = ( const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries; const workspaceFileSystem = yield* WorkspaceFileSystem.WorkspaceFileSystem; const projectSetupScriptRunner = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner; - const repositoryIdentityResolver = - yield* RepositoryIdentityResolver.RepositoryIdentityResolver; const serverEnvironment = yield* ServerEnvironment.ServerEnvironment; const serverAuth = yield* EnvironmentAuth.EnvironmentAuth; const sourceControlDiscovery = yield* SourceControlDiscovery.SourceControlDiscovery; @@ -592,53 +586,6 @@ const makeWsRpcLayer = ( }); }; - const enrichProjectEvent = ( - event: OrchestrationEvent, - ): Effect.Effect => { - switch (event.type) { - case "project.created": - return repositoryIdentityResolver.resolve(event.payload.workspaceRoot).pipe( - Effect.map((repositoryIdentity) => ({ - ...event, - payload: { - ...event.payload, - repositoryIdentity, - }, - })), - ); - case "project.meta-updated": - return Effect.gen(function* () { - const workspaceRoot = - event.payload.workspaceRoot ?? - Option.match( - yield* projectionSnapshotQuery.getProjectShellById(event.payload.projectId), - { - onNone: () => null, - onSome: (project) => project.workspaceRoot, - }, - ) ?? - null; - if (workspaceRoot === null) { - return event; - } - - const repositoryIdentity = yield* repositoryIdentityResolver.resolve(workspaceRoot); - return { - ...event, - payload: { - ...event.payload, - repositoryIdentity, - }, - } satisfies OrchestrationEvent; - }).pipe(Effect.orElseSucceed(() => event)); - default: - return Effect.succeed(event); - } - }; - - const enrichOrchestrationEvents = (events: ReadonlyArray) => - Effect.forEach(events, enrichProjectEvent, { concurrency: 4 }); - const toShellStreamEvent = ( event: OrchestrationEvent, ): Effect.Effect, never, never> => { @@ -1215,30 +1162,6 @@ const makeWsRpcLayer = ( ), { "rpc.aggregate": "orchestration" }, ), - [ORCHESTRATION_WS_METHODS.replayEvents]: (input) => - observeRpcEffect( - ORCHESTRATION_WS_METHODS.replayEvents, - Stream.runCollect( - orchestrationEngine.readEvents( - clamp(input.fromSequenceExclusive, { - maximum: Number.MAX_SAFE_INTEGER, - minimum: 0, - }), - ), - ).pipe( - Effect.map((events) => Array.from(events)), - Effect.flatMap(enrichOrchestrationEvents), - Effect.map((events) => events.map(projectActivityEvent)), - Effect.mapError( - (cause) => - new OrchestrationReplayEventsError({ - message: "Failed to replay orchestration events", - cause, - }), - ), - ), - { "rpc.aggregate": "orchestration" }, - ), [ORCHESTRATION_WS_METHODS.subscribeShell]: (input) => observeRpcStreamEffect( ORCHESTRATION_WS_METHODS.subscribeShell, diff --git a/apps/server/test/ActivityPayloadProjection.test.ts b/apps/server/test/ActivityPayloadProjection.test.ts index 6552dfb140c..d6098937e7f 100644 --- a/apps/server/test/ActivityPayloadProjection.test.ts +++ b/apps/server/test/ActivityPayloadProjection.test.ts @@ -11,6 +11,7 @@ import { import { describe, expect, it } from "vite-plus/test"; import { buildThreadFeed, type ThreadFeedActivity } from "../../mobile/src/lib/threadActivity.ts"; +import { deriveLatestContextWindowSnapshot } from "../../web/src/lib/contextWindow.ts"; import { deriveWorkLogEntries } from "../../web/src/session-logic.ts"; import { projectActivityEvent, @@ -231,3 +232,132 @@ describe("projectActivityPayload", () => { expect(event.payload.activity).toBe(activity); }); }); + +describe("context-window snapshot dedup", () => { + function makeContextWindowActivity( + id: string, + usedTokens: number, + turn = `turn-${id}`, + ): OrchestrationThreadActivity { + return { + id: EventId.make(id), + tone: "info", + kind: "context-window.updated", + summary: "Context window updated", + payload: { usedTokens, maxTokens: 200_000 }, + turnId: TurnId.make(turn), + createdAt: "2026-07-27T00:00:00.000Z", + }; + } + + it("keeps only the latest context-window activity per turn in snapshots", () => { + const stale1 = makeContextWindowActivity("ctx-1", 1_000, "turn-a"); + const latestA = makeContextWindowActivity("ctx-2", 2_000, "turn-a"); + const latestB = makeContextWindowActivity("ctx-3", 3_000, "turn-b"); + const tool = fixtures[0]!; + + const projected = projectThreadDetailSnapshot({ + snapshotSequence: 7, + thread: makeThread([stale1, tool, latestA, latestB]), + }); + + expect(projected.thread.activities.map((activity) => activity.id)).toEqual([ + tool.id, + latestA.id, + latestB.id, + ]); + // The retained rows keep their payloads untouched — the tool-data + // projection only rewrites payloads with a `data` record. + expect(projected.thread.activities[2]?.payload).toEqual(latestB.payload); + }); + + it("still resolves a meter value after the client reverts the newest turn", () => { + // A live thread.reverted makes the client drop all activities from + // discarded turns; each surviving turn must keep a usable row. + const olderTurn = makeContextWindowActivity("ctx-old", 1_500, "turn-kept"); + const revertedTurn = makeContextWindowActivity("ctx-new", 9_000, "turn-reverted"); + + const projected = projectThreadDetailSnapshot({ + snapshotSequence: 7, + thread: makeThread([olderTurn, revertedTurn]), + }); + const afterRevert = projected.thread.activities.filter( + (activity) => activity.turnId === TurnId.make("turn-kept"), + ); + + expect(deriveLatestContextWindowSnapshot(afterRevert)).toEqual( + deriveLatestContextWindowSnapshot([olderTurn]), + ); + }); + + it("matches what the web client derives from the full history", () => { + const activities = [ + makeContextWindowActivity("ctx-1", 1_000), + makeContextWindowActivity("ctx-2", 2_000), + ]; + const projected = projectThreadDetailSnapshot({ + snapshotSequence: 7, + thread: makeThread(activities), + }); + + expect(deriveLatestContextWindowSnapshot(projected.thread.activities)).toEqual( + deriveLatestContextWindowSnapshot(activities), + ); + }); + + it("does not let a malformed row shadow an earlier valid row in the same turn", () => { + const valid = makeContextWindowActivity("ctx-valid", 5_000, "turn-a"); + const malformed: OrchestrationThreadActivity = { + ...makeContextWindowActivity("ctx-broken", 0, "turn-a"), + payload: { usedTokens: null }, + }; + + const projected = projectThreadDetailSnapshot({ + snapshotSequence: 7, + thread: makeThread([valid, malformed]), + }); + + // The malformed row passes through, the valid row survives, and the + // client's backward walk resolves the same value as with full history. + expect(projected.thread.activities.map((activity) => activity.id)).toEqual([ + valid.id, + malformed.id, + ]); + expect(deriveLatestContextWindowSnapshot(projected.thread.activities)).toEqual( + deriveLatestContextWindowSnapshot([valid, malformed]), + ); + }); + + it("leaves snapshots without context-window activities untouched", () => { + const projected = projectThreadDetailSnapshot({ + snapshotSequence: 7, + thread: makeThread([fixtures[4]!]), + }); + expect(projected.thread.activities).toEqual([fixtures[4]]); + }); + + it("does not filter live activity-appended events", () => { + const activity = makeContextWindowActivity("ctx-live", 4_000); + const event = { + sequence: 9, + eventId: EventId.make("event-ctx"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-projection"), + occurredAt: "2026-07-27T00:00:02.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.activity-appended", + payload: { + threadId: ThreadId.make("thread-projection"), + activity, + }, + } satisfies Extract; + + const projected = projectActivityEvent(event); + expect( + projected.type === "thread.activity-appended" ? projected.payload.activity : undefined, + ).toEqual(activity); + }); +}); diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 84b7a8fa07f..b947bd63e4c 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -26,7 +26,6 @@ export const ORCHESTRATION_WS_METHODS = { dispatchCommand: "orchestration.dispatchCommand", getTurnDiff: "orchestration.getTurnDiff", getFullThreadDiff: "orchestration.getFullThreadDiff", - replayEvents: "orchestration.replayEvents", getArchivedShellSnapshot: "orchestration.getArchivedShellSnapshot", subscribeShell: "orchestration.subscribeShell", subscribeThread: "orchestration.subscribeThread", @@ -1363,14 +1362,6 @@ export type OrchestrationGetFullThreadDiffInput = typeof OrchestrationGetFullThr export const OrchestrationGetFullThreadDiffResult = ThreadTurnDiff; export type OrchestrationGetFullThreadDiffResult = typeof OrchestrationGetFullThreadDiffResult.Type; -export const OrchestrationReplayEventsInput = Schema.Struct({ - fromSequenceExclusive: NonNegativeInt, -}); -export type OrchestrationReplayEventsInput = typeof OrchestrationReplayEventsInput.Type; - -const OrchestrationReplayEventsResult = Schema.Array(OrchestrationEvent); -export type OrchestrationReplayEventsResult = typeof OrchestrationReplayEventsResult.Type; - export const OrchestrationRpcSchemas = { dispatchCommand: { input: ClientOrchestrationCommand, @@ -1384,10 +1375,6 @@ export const OrchestrationRpcSchemas = { input: OrchestrationGetFullThreadDiffInput, output: OrchestrationGetFullThreadDiffResult, }, - replayEvents: { - input: OrchestrationReplayEventsInput, - output: OrchestrationReplayEventsResult, - }, getArchivedShellSnapshot: { input: Schema.Struct({}), output: OrchestrationShellSnapshot, @@ -1433,11 +1420,3 @@ export class OrchestrationGetFullThreadDiffError extends Schema.TaggedErrorClass cause: Schema.optional(Schema.Defect()), }, ) {} - -export class OrchestrationReplayEventsError extends Schema.TaggedErrorClass()( - "OrchestrationReplayEventsError", - { - message: TrimmedNonEmptyString, - cause: Schema.optional(Schema.Defect()), - }, -) {} diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index fa2d23b8ef2..43fc2d1acb0 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -54,8 +54,6 @@ import { OrchestrationGetSnapshotError, OrchestrationGetTurnDiffError, OrchestrationGetTurnDiffInput, - OrchestrationReplayEventsError, - OrchestrationReplayEventsInput, OrchestrationRpcSchemas, } from "./orchestration.ts"; import { ProviderInstanceId } from "./providerInstance.ts"; @@ -631,12 +629,6 @@ export const WsOrchestrationGetFullThreadDiffRpc = Rpc.make( }, ); -export const WsOrchestrationReplayEventsRpc = Rpc.make(ORCHESTRATION_WS_METHODS.replayEvents, { - payload: OrchestrationReplayEventsInput, - success: OrchestrationRpcSchemas.replayEvents.output, - error: Schema.Union([OrchestrationReplayEventsError, EnvironmentAuthorizationError]), -}); - export const WsOrchestrationGetArchivedShellSnapshotRpc = Rpc.make( ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot, { @@ -765,7 +757,6 @@ export const WsRpcGroup = RpcGroup.make( WsOrchestrationDispatchCommandRpc, WsOrchestrationGetTurnDiffRpc, WsOrchestrationGetFullThreadDiffRpc, - WsOrchestrationReplayEventsRpc, WsOrchestrationGetArchivedShellSnapshotRpc, WsOrchestrationSubscribeShellRpc, WsOrchestrationSubscribeThreadRpc,