diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 0756fdd59cdafe8767f2b9c123455f73bfb84e62..6853c6457d6d021335ca655157b53fa9f23c0d38 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -3251,6 +3251,27 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }; } })(); + if ( + command.type === "thread.pull-request-link.sync" || + command.type === "thread.pull-request-watch.sync" + ) { + if (updatedThread === thread) return; + const link = threadPullRequestsOf(updatedThread).find((candidate) => + threadPullRequestKeysEqual(candidate, normalizeThreadPullRequestKey(command)), + ); + if (link === undefined) return; + yield* emit( + events, + command, + )({ + type: "thread.pull-request-link-updated", + threadId: command.threadId, + providerInstanceId: updatedThread.providerInstanceId, + occurredAt: now, + payload: link, + }); + return; + } const eventType = (() => { switch (command.type) { case "thread.archive": @@ -3282,9 +3303,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return "thread.metadata-updated" as const; case "thread.pull-request.link": case "thread.pull-request.unlink": - case "thread.pull-request-link.sync": case "thread.pull-request.watch": - case "thread.pull-request-watch.sync": case "thread.pull-request.sync": return "thread.pull-request-synced" as const; case "thread.runtime-mode.set": diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index f3348438352703cb915ede612937c048ee776eeb..b999f6376ecb3fdc539dc24de46653adaa28d6bd 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1,3 +1,4 @@ +import { applyThreadPullRequestLinkUpdate } from "@t3tools/shared/threadPullRequests"; import { projectTurnItemForWire } from "./WireProjection.ts"; import * as Stream from "effect/Stream"; import { makeThreadFind, findProjectedThreadItems } from "./ThreadFind.ts"; @@ -718,6 +719,11 @@ export function applyToProjection( }; switch (event.type) { + case "thread.pull-request-link-updated": + return { + ...base, + thread: applyThreadPullRequestLinkUpdate(projection.thread, event.payload), + }; case "thread.created": case "thread.archived": case "thread.unarchived": @@ -1823,6 +1829,15 @@ export const layer: Layer.Layer = const apply: ProjectionStoreV2Shape["apply"] = (event) => Effect.gen(function* () { switch (event.type) { + case "thread.pull-request-link-updated": { + const thread = yield* getThread(event.threadId); + yield* apply({ + ...event, + type: "thread.pull-request-synced", + payload: applyThreadPullRequestLinkUpdate(thread, event.payload), + }); + break; + } case "thread.created": case "thread.archived": case "thread.unarchived": diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index b1fe998c229608a49519a5d3c3e56d9e32203022..70be6e0b08780b0d9e6fd3df53c62883093541c4 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -440,6 +440,8 @@ export const make = Effect.gen(function* () { yield* forkParked( Stream.runForEach(events, (event) => { switch (event.type) { + case "thread.pull-request-link-updated": + return event.payload.snapshot === null ? requestSync(event.payload) : Effect.void; case "thread.pull-request-synced": return Effect.forEach( visibleThreadPullRequests(event.payload.pullRequests ?? []).filter( diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.ts index 0eb779639116b0fa67d991f047d8637d262159e3..bd7a7fd58dcb01491963e678884b413c8c7ad1ae 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.ts @@ -606,6 +606,7 @@ export const make = Effect.gen(function* () { switch (event.type) { case "thread.settled": return cleanUpSettledThread(event.threadId); + case "thread.pull-request-link-updated": case "thread.pull-request-synced": case "provider-session.detached": return worker.enqueue(event.threadId); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index c4f5b8295fd9c8144920b57b5089482792ab08de..9bd7f38ddb8bb5e7e9d23e371f1157b06e74a59b 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -4860,6 +4860,93 @@ it.layer(layerSharedApplicationDataPlaneTest)("visited projection", (it) => { ); }); +it.layer(Layer.fresh(layerSharedApplicationDataPlaneTest))("compact PR events", (it) => { + it.effect("persists only the changed PR link and replays it without losing other links", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("compact-pr-sync"); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("compact-pr-create"), + threadId, + projectId: ProjectId.make("compact-pr-project"), + title: "PR sync", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + const key = { host: "github.com", repository: "pingdotgg/t3code" }; + for (const number of [1, 2, 3]) { + yield* orchestrator.dispatch({ + type: "thread.pull-request.link", + commandId: CommandId.make(`compact-pr-link-${number}`), + threadId, + ...key, + number, + url: `https://github.com/pingdotgg/t3code/pull/${number}`, + source: "manual", + }); + } + const before = yield* orchestrator.getThreadShell(threadId); + yield* orchestrator.dispatch({ + type: "thread.pull-request-link.sync", + commandId: CommandId.make("compact-pr-sync"), + threadId, + ...key, + number: 2, + snapshot: { + state: "open", + title: "Updated PR", + headBranch: "feature/pr", + baseBranch: "main", + isDraft: false, + updatedAt: null, + syncedAt: "2026-10-13T12:00:00.000Z", + }, + stack: { + kind: "native", + id: "updated", + number: 2, + url: "https://github.com/pingdotgg/t3code/pull/2", + base: "main", + layers: [], + }, + }); + const rows = yield* sql<{ event_type: string; payload_json: string }>` + SELECT event_type, payload_json FROM orchestration_events + WHERE command_id = ${"compact-pr-sync"} + `; + assert.equal(rows.length, 1); + assert.equal(rows[0]?.event_type, "thread.pull-request-link-updated"); + const payload = JSON.parse(rows[0]!.payload_json); + assert.equal(payload.number, 2); + assert.isUndefined(payload.pullRequests); + const after = yield* orchestrator.getThreadShell(threadId); + assert.equal(after?.pullRequests?.length, 3); + assert.deepEqual(after?.pullRequests?.[0], before?.pullRequests?.[0]); + assert.deepEqual(after?.pullRequests?.[2], before?.pullRequests?.[2]); + assert.equal(after?.pullRequests?.[1]?.stack?.id, "updated"); + assert.isTrue((yield* maintenance.rebuild).valid); + assert.deepEqual( + (yield* orchestrator.getThreadShell(threadId))?.pullRequests, + after?.pullRequests, + ); + yield* maintenance.compactEventStore; + assert.isTrue((yield* maintenance.rebuild).valid); + assert.deepEqual( + (yield* orchestrator.getThreadShell(threadId))?.pullRequests, + after?.pullRequests, + ); + }), + ); +}); + it.layer(layerSharedApplicationDataPlaneTest)("shared application data plane", (it) => { it.effect("orders retained project transactions and V2 thread transactions in one source", () => Effect.gen(function* () { diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index 55ddeac252c256ca7433a89e0d1cd8d959a91f82..447192fce47b65aae0e767e6478a33b9710858b6 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -97,6 +97,7 @@ export function shouldPublishAgentAwarenessEvent( case "thread.unarchived": case "thread.deleted": case "thread.metadata-updated": + case "thread.pull-request-link-updated": case "thread.pull-request-synced": case "thread.model-selection-updated": case "thread.provider-switched": diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.test.ts b/packages/client-runtime/src/state/orchestrationV2Projection.test.ts index e41eb81257f9aecc03aad5214ef1adc1c9f9dccd..478365a2e5318b4f16e9f1c03045f4b38b31c09c 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.test.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from "vite-plus/test"; import { + type ThreadPullRequestLink, type OrchestrationV2DomainEvent, type OrchestrationV2Run, type OrchestrationV2ThreadProjection, @@ -105,6 +106,34 @@ const emptyProjection = { } as OrchestrationV2ThreadProjection; describe("applyOrchestrationV2ProjectionEvent", () => { + it("applies a compact PR sync without replacing other links or thread metadata", () => { + const link = (number: number): ThreadPullRequestLink => ({ + host: "github.com", + repository: "pingdotgg/t3code", + number, + url: `https://github.com/pingdotgg/t3code/pull/${number}`, + source: "manual", + linkedAt: DateTime.formatIso(now), + snapshot: null, + stack: null, + }); + const links = [link(1), link(2)]; + const projection = { + ...emptyProjection, + thread: { ...emptyProjection.thread, pullRequests: links }, + }; + const updated = { ...links[1]!, source: "agent" as const }; + const next = applyOrchestrationV2ProjectionEvent(projection, { + id: "compact-sync" as OrchestrationV2DomainEvent["id"], + type: "thread.pull-request-link-updated", + threadId, + occurredAt: now, + payload: updated, + }); + expect(next?.thread).toEqual({ ...projection.thread, pullRequests: [links[0], updated] }); + expect(next?.thread.pullRequests?.[0]).toBe(links[0]); + }); + it("keeps live token usage when the terminal provider turn omits it", () => { const providerTurnId = ProviderTurnId.make("provider-turn-reducer"); const running = { diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.ts b/packages/client-runtime/src/state/orchestrationV2Projection.ts index 4587221c07dffd059eaac0fc0ddbd4e3aa47ed21..3c9c6dfa79caf3d36f679bd5d37b2c7eb97a7224 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.ts @@ -1,3 +1,4 @@ +import { applyThreadPullRequestLinkUpdate } from "@t3tools/shared/threadPullRequests"; import type { OrchestrationV2DomainEvent, OrchestrationV2Run, @@ -201,6 +202,11 @@ export function applyOrchestrationV2ProjectionEvent( const latestLocalTurnOrdinal = options?.latestLocalTurnOrdinal; const base = { ...projection, updatedAt: event.occurredAt }; switch (event.type) { + case "thread.pull-request-link-updated": + return { + ...base, + thread: applyThreadPullRequestLinkUpdate(projection.thread, event.payload), + }; case "thread.created": case "thread.archived": case "thread.unarchived": diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 41eabae0aea6d77d00790ab1190e400e09c53512..028d536d3758f9220b5d865633f072d81f634f46 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1765,6 +1765,11 @@ export const OrchestrationV2DomainEvent = Schema.Union([ ]), payload: OrchestrationV2AppThread, }), + Schema.Struct({ + ...OrchestrationV2EventBase.fields, + type: Schema.Literal("thread.pull-request-link-updated"), + payload: ThreadPullRequestLink, + }), Schema.Struct({ ...OrchestrationV2EventBase.fields, type: Schema.Literal("run.created"), @@ -2585,6 +2590,11 @@ export const OrchestrationV2DomainEventJson = Schema.Union([ ]), payload: OrchestrationV2AppThreadJson, }), + Schema.Struct({ + ...OrchestrationV2JsonEventBaseFields, + type: Schema.Literal("thread.pull-request-link-updated"), + payload: ThreadPullRequestLink, + }), Schema.Struct({ ...OrchestrationV2JsonEventBaseFields, type: Schema.Literal("run.created"), diff --git a/packages/shared/src/threadPullRequests.ts b/packages/shared/src/threadPullRequests.ts index 97b800d8682ffe4a54d4863a6a99a38ce0a35f50..e926e10d3b43ed8c363bb0341a899b161ba1b205 100644 --- a/packages/shared/src/threadPullRequests.ts +++ b/packages/shared/src/threadPullRequests.ts @@ -1,4 +1,5 @@ import type { + OrchestrationV2AppThread, RepositoryIdentity, SourceControlProviderKind, ThreadLinkedPullRequest, @@ -354,3 +355,16 @@ export function threadPullRequestsOf(thread: { }, ]; } + +/** Apply a single-link sync without copying unrelated PRs into the event history. */ +export function applyThreadPullRequestLinkUpdate( + thread: OrchestrationV2AppThread, + updated: ThreadPullRequestLink, +): OrchestrationV2AppThread { + return { + ...thread, + pullRequests: threadPullRequestsOf(thread).map((link) => + threadPullRequestKeysEqual(link, updated) ? updated : link, + ), + }; +}