authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-10-13 22:46:11-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-13 22:46:11-07:00
log4d73e1d5cdd32dbfb470e9422421cc40475d5e99
treebb302023136c2d6abba27b43edab2bc0eddef565
parentd765d03cb7f980794c409c38e6ea5e25d89fde33
signature Signed by PGP key B5690EEEBB952194

fix(server): persist only changed pull request links (#19248)

10 files changed, 186 insertions(+), 2 deletions(-)

apps/server/src/orchestration-v2/Orchestrator.ts+21-2
...@@ -3251,6 +3251,27 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio...@@ -3251,6 +3251,27 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
3251 };3251 };
3252 }3252 }
3253 })();3253 })();
3254 if (
3255 command.type === "thread.pull-request-link.sync" ||
3256 command.type === "thread.pull-request-watch.sync"
3257 ) {
3258 if (updatedThread === thread) return;
3259 const link = threadPullRequestsOf(updatedThread).find((candidate) =>
3260 threadPullRequestKeysEqual(candidate, normalizeThreadPullRequestKey(command)),
3261 );
3262 if (link === undefined) return;
3263 yield* emit(
3264 events,
3265 command,
3266 )({
3267 type: "thread.pull-request-link-updated",
3268 threadId: command.threadId,
3269 providerInstanceId: updatedThread.providerInstanceId,
3270 occurredAt: now,
3271 payload: link,
3272 });
3273 return;
3274 }
3254 const eventType = (() => {3275 const eventType = (() => {
3255 switch (command.type) {3276 switch (command.type) {
3256 case "thread.archive":3277 case "thread.archive":
...@@ -3282,9 +3303,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio...@@ -3282,9 +3303,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
3282 return "thread.metadata-updated" as const;3303 return "thread.metadata-updated" as const;
3283 case "thread.pull-request.link":3304 case "thread.pull-request.link":
3284 case "thread.pull-request.unlink":3305 case "thread.pull-request.unlink":
3285 case "thread.pull-request-link.sync":
3286 case "thread.pull-request.watch":3306 case "thread.pull-request.watch":
3287 case "thread.pull-request-watch.sync":
3288 case "thread.pull-request.sync":3307 case "thread.pull-request.sync":
3289 return "thread.pull-request-synced" as const;3308 return "thread.pull-request-synced" as const;
3290 case "thread.runtime-mode.set":3309 case "thread.runtime-mode.set":
apps/server/src/orchestration-v2/ProjectionStore.ts+15
...@@ -1,3 +1,4 @@...@@ -1,3 +1,4 @@
1import { applyThreadPullRequestLinkUpdate } from "@t3tools/shared/threadPullRequests";
1import { projectTurnItemForWire } from "./WireProjection.ts";2import { projectTurnItemForWire } from "./WireProjection.ts";
2import * as Stream from "effect/Stream";3import * as Stream from "effect/Stream";
3import { makeThreadFind, findProjectedThreadItems } from "./ThreadFind.ts";4import { makeThreadFind, findProjectedThreadItems } from "./ThreadFind.ts";
...@@ -718,6 +719,11 @@ export function applyToProjection(...@@ -718,6 +719,11 @@ export function applyToProjection(
718 };719 };
719720
720 switch (event.type) {721 switch (event.type) {
722 case "thread.pull-request-link-updated":
723 return {
724 ...base,
725 thread: applyThreadPullRequestLinkUpdate(projection.thread, event.payload),
726 };
721 case "thread.created":727 case "thread.created":
722 case "thread.archived":728 case "thread.archived":
723 case "thread.unarchived":729 case "thread.unarchived":
...@@ -1823,6 +1829,15 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =...@@ -1823,6 +1829,15 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
1823 const apply: ProjectionStoreV2Shape["apply"] = (event) =>1829 const apply: ProjectionStoreV2Shape["apply"] = (event) =>
1824 Effect.gen(function* () {1830 Effect.gen(function* () {
1825 switch (event.type) {1831 switch (event.type) {
1832 case "thread.pull-request-link-updated": {
1833 const thread = yield* getThread(event.threadId);
1834 yield* apply({
1835 ...event,
1836 type: "thread.pull-request-synced",
1837 payload: applyThreadPullRequestLinkUpdate(thread, event.payload),
1838 });
1839 break;
1840 }
1826 case "thread.created":1841 case "thread.created":
1827 case "thread.archived":1842 case "thread.archived":
1828 case "thread.unarchived":1843 case "thread.unarchived":
apps/server/src/orchestration-v2/PullRequestSyncReactor.ts+2
...@@ -440,6 +440,8 @@ export const make = Effect.gen(function* () {...@@ -440,6 +440,8 @@ export const make = Effect.gen(function* () {
440 yield* forkParked(440 yield* forkParked(
441 Stream.runForEach(events, (event) => {441 Stream.runForEach(events, (event) => {
442 switch (event.type) {442 switch (event.type) {
443 case "thread.pull-request-link-updated":
444 return event.payload.snapshot === null ? requestSync(event.payload) : Effect.void;
443 case "thread.pull-request-synced":445 case "thread.pull-request-synced":
444 return Effect.forEach(446 return Effect.forEach(
445 visibleThreadPullRequests(event.payload.pullRequests ?? []).filter(447 visibleThreadPullRequests(event.payload.pullRequests ?? []).filter(
apps/server/src/orchestration-v2/ThreadSettlementService.ts+1
...@@ -606,6 +606,7 @@ export const make = Effect.gen(function* () {...@@ -606,6 +606,7 @@ export const make = Effect.gen(function* () {
606 switch (event.type) {606 switch (event.type) {
607 case "thread.settled":607 case "thread.settled":
608 return cleanUpSettledThread(event.threadId);608 return cleanUpSettledThread(event.threadId);
609 case "thread.pull-request-link-updated":
609 case "thread.pull-request-synced":610 case "thread.pull-request-synced":
610 case "provider-session.detached":611 case "provider-session.detached":
611 return worker.enqueue(event.threadId);612 return worker.enqueue(event.threadId);
apps/server/src/orchestration-v2/runtimeLayer.test.ts+87
...@@ -4860,6 +4860,93 @@ it.layer(layerSharedApplicationDataPlaneTest)("visited projection", (it) => {...@@ -4860,6 +4860,93 @@ it.layer(layerSharedApplicationDataPlaneTest)("visited projection", (it) => {
4860 );4860 );
4861});4861});
48624862
4863it.layer(Layer.fresh(layerSharedApplicationDataPlaneTest))("compact PR events", (it) => {
4864 it.effect("persists only the changed PR link and replays it without losing other links", () =>
4865 Effect.gen(function* () {
4866 const orchestrator = yield* Orchestrator.OrchestratorV2;
4867 const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
4868 const sql = yield* SqlClient.SqlClient;
4869 const threadId = ThreadId.make("compact-pr-sync");
4870 yield* orchestrator.dispatch({
4871 type: "thread.create",
4872 createdBy: "user",
4873 creationSource: "web",
4874 commandId: CommandId.make("compact-pr-create"),
4875 threadId,
4876 projectId: ProjectId.make("compact-pr-project"),
4877 title: "PR sync",
4878 modelSelection,
4879 runtimeMode: "full-access",
4880 interactionMode: "default",
4881 branch: null,
4882 worktreePath: null,
4883 });
4884 const key = { host: "github.com", repository: "pingdotgg/t3code" };
4885 for (const number of [1, 2, 3]) {
4886 yield* orchestrator.dispatch({
4887 type: "thread.pull-request.link",
4888 commandId: CommandId.make(`compact-pr-link-${number}`),
4889 threadId,
4890 ...key,
4891 number,
4892 url: `https://github.com/pingdotgg/t3code/pull/${number}`,
4893 source: "manual",
4894 });
4895 }
4896 const before = yield* orchestrator.getThreadShell(threadId);
4897 yield* orchestrator.dispatch({
4898 type: "thread.pull-request-link.sync",
4899 commandId: CommandId.make("compact-pr-sync"),
4900 threadId,
4901 ...key,
4902 number: 2,
4903 snapshot: {
4904 state: "open",
4905 title: "Updated PR",
4906 headBranch: "feature/pr",
4907 baseBranch: "main",
4908 isDraft: false,
4909 updatedAt: null,
4910 syncedAt: "2026-10-13T12:00:00.000Z",
4911 },
4912 stack: {
4913 kind: "native",
4914 id: "updated",
4915 number: 2,
4916 url: "https://github.com/pingdotgg/t3code/pull/2",
4917 base: "main",
4918 layers: [],
4919 },
4920 });
4921 const rows = yield* sql<{ event_type: string; payload_json: string }>`
4922 SELECT event_type, payload_json FROM orchestration_events
4923 WHERE command_id = ${"compact-pr-sync"}
4924 `;
4925 assert.equal(rows.length, 1);
4926 assert.equal(rows[0]?.event_type, "thread.pull-request-link-updated");
4927 const payload = JSON.parse(rows[0]!.payload_json);
4928 assert.equal(payload.number, 2);
4929 assert.isUndefined(payload.pullRequests);
4930 const after = yield* orchestrator.getThreadShell(threadId);
4931 assert.equal(after?.pullRequests?.length, 3);
4932 assert.deepEqual(after?.pullRequests?.[0], before?.pullRequests?.[0]);
4933 assert.deepEqual(after?.pullRequests?.[2], before?.pullRequests?.[2]);
4934 assert.equal(after?.pullRequests?.[1]?.stack?.id, "updated");
4935 assert.isTrue((yield* maintenance.rebuild).valid);
4936 assert.deepEqual(
4937 (yield* orchestrator.getThreadShell(threadId))?.pullRequests,
4938 after?.pullRequests,
4939 );
4940 yield* maintenance.compactEventStore;
4941 assert.isTrue((yield* maintenance.rebuild).valid);
4942 assert.deepEqual(
4943 (yield* orchestrator.getThreadShell(threadId))?.pullRequests,
4944 after?.pullRequests,
4945 );
4946 }),
4947 );
4948});
4949
4863it.layer(layerSharedApplicationDataPlaneTest)("shared application data plane", (it) => {4950it.layer(layerSharedApplicationDataPlaneTest)("shared application data plane", (it) => {
4864 it.effect("orders retained project transactions and V2 thread transactions in one source", () =>4951 it.effect("orders retained project transactions and V2 thread transactions in one source", () =>
4865 Effect.gen(function* () {4952 Effect.gen(function* () {
apps/server/src/relay/AgentAwarenessRelay.ts+1
...@@ -97,6 +97,7 @@ export function shouldPublishAgentAwarenessEvent(...@@ -97,6 +97,7 @@ export function shouldPublishAgentAwarenessEvent(
97 case "thread.unarchived":97 case "thread.unarchived":
98 case "thread.deleted":98 case "thread.deleted":
99 case "thread.metadata-updated":99 case "thread.metadata-updated":
100 case "thread.pull-request-link-updated":
100 case "thread.pull-request-synced":101 case "thread.pull-request-synced":
101 case "thread.model-selection-updated":102 case "thread.model-selection-updated":
102 case "thread.provider-switched":103 case "thread.provider-switched":
packages/client-runtime/src/state/orchestrationV2Projection.test.ts+29
...@@ -1,5 +1,6 @@...@@ -1,5 +1,6 @@
1import { describe, expect, it } from "vite-plus/test";1import { describe, expect, it } from "vite-plus/test";
2import {2import {
3 type ThreadPullRequestLink,
3 type OrchestrationV2DomainEvent,4 type OrchestrationV2DomainEvent,
4 type OrchestrationV2Run,5 type OrchestrationV2Run,
5 type OrchestrationV2ThreadProjection,6 type OrchestrationV2ThreadProjection,
...@@ -105,6 +106,34 @@ const emptyProjection = {...@@ -105,6 +106,34 @@ const emptyProjection = {
105} as OrchestrationV2ThreadProjection;106} as OrchestrationV2ThreadProjection;
106107
107describe("applyOrchestrationV2ProjectionEvent", () => {108describe("applyOrchestrationV2ProjectionEvent", () => {
109 it("applies a compact PR sync without replacing other links or thread metadata", () => {
110 const link = (number: number): ThreadPullRequestLink => ({
111 host: "github.com",
112 repository: "pingdotgg/t3code",
113 number,
114 url: `https://github.com/pingdotgg/t3code/pull/${number}`,
115 source: "manual",
116 linkedAt: DateTime.formatIso(now),
117 snapshot: null,
118 stack: null,
119 });
120 const links = [link(1), link(2)];
121 const projection = {
122 ...emptyProjection,
123 thread: { ...emptyProjection.thread, pullRequests: links },
124 };
125 const updated = { ...links[1]!, source: "agent" as const };
126 const next = applyOrchestrationV2ProjectionEvent(projection, {
127 id: "compact-sync" as OrchestrationV2DomainEvent["id"],
128 type: "thread.pull-request-link-updated",
129 threadId,
130 occurredAt: now,
131 payload: updated,
132 });
133 expect(next?.thread).toEqual({ ...projection.thread, pullRequests: [links[0], updated] });
134 expect(next?.thread.pullRequests?.[0]).toBe(links[0]);
135 });
136
108 it("keeps live token usage when the terminal provider turn omits it", () => {137 it("keeps live token usage when the terminal provider turn omits it", () => {
109 const providerTurnId = ProviderTurnId.make("provider-turn-reducer");138 const providerTurnId = ProviderTurnId.make("provider-turn-reducer");
110 const running = {139 const running = {
packages/client-runtime/src/state/orchestrationV2Projection.ts+6
...@@ -1,3 +1,4 @@...@@ -1,3 +1,4 @@
1import { applyThreadPullRequestLinkUpdate } from "@t3tools/shared/threadPullRequests";
1import type {2import type {
2 OrchestrationV2DomainEvent,3 OrchestrationV2DomainEvent,
3 OrchestrationV2Run,4 OrchestrationV2Run,
...@@ -201,6 +202,11 @@ export function applyOrchestrationV2ProjectionEvent(...@@ -201,6 +202,11 @@ export function applyOrchestrationV2ProjectionEvent(
201 const latestLocalTurnOrdinal = options?.latestLocalTurnOrdinal;202 const latestLocalTurnOrdinal = options?.latestLocalTurnOrdinal;
202 const base = { ...projection, updatedAt: event.occurredAt };203 const base = { ...projection, updatedAt: event.occurredAt };
203 switch (event.type) {204 switch (event.type) {
205 case "thread.pull-request-link-updated":
206 return {
207 ...base,
208 thread: applyThreadPullRequestLinkUpdate(projection.thread, event.payload),
209 };
204 case "thread.created":210 case "thread.created":
205 case "thread.archived":211 case "thread.archived":
206 case "thread.unarchived":212 case "thread.unarchived":
packages/contracts/src/orchestrationV2.ts+10
...@@ -1765,6 +1765,11 @@ export const OrchestrationV2DomainEvent = Schema.Union([...@@ -1765,6 +1765,11 @@ export const OrchestrationV2DomainEvent = Schema.Union([
1765 ]),1765 ]),
1766 payload: OrchestrationV2AppThread,1766 payload: OrchestrationV2AppThread,
1767 }),1767 }),
1768 Schema.Struct({
1769 ...OrchestrationV2EventBase.fields,
1770 type: Schema.Literal("thread.pull-request-link-updated"),
1771 payload: ThreadPullRequestLink,
1772 }),
1768 Schema.Struct({1773 Schema.Struct({
1769 ...OrchestrationV2EventBase.fields,1774 ...OrchestrationV2EventBase.fields,
1770 type: Schema.Literal("run.created"),1775 type: Schema.Literal("run.created"),
...@@ -2585,6 +2590,11 @@ export const OrchestrationV2DomainEventJson = Schema.Union([...@@ -2585,6 +2590,11 @@ export const OrchestrationV2DomainEventJson = Schema.Union([
2585 ]),2590 ]),
2586 payload: OrchestrationV2AppThreadJson,2591 payload: OrchestrationV2AppThreadJson,
2587 }),2592 }),
2593 Schema.Struct({
2594 ...OrchestrationV2JsonEventBaseFields,
2595 type: Schema.Literal("thread.pull-request-link-updated"),
2596 payload: ThreadPullRequestLink,
2597 }),
2588 Schema.Struct({2598 Schema.Struct({
2589 ...OrchestrationV2JsonEventBaseFields,2599 ...OrchestrationV2JsonEventBaseFields,
2590 type: Schema.Literal("run.created"),2600 type: Schema.Literal("run.created"),
packages/shared/src/threadPullRequests.ts+14
...@@ -1,4 +1,5 @@...@@ -1,4 +1,5 @@
1import type {1import type {
2 OrchestrationV2AppThread,
2 RepositoryIdentity,3 RepositoryIdentity,
3 SourceControlProviderKind,4 SourceControlProviderKind,
4 ThreadLinkedPullRequest,5 ThreadLinkedPullRequest,
...@@ -354,3 +355,16 @@ export function threadPullRequestsOf(thread: {...@@ -354,3 +355,16 @@ export function threadPullRequestsOf(thread: {
354 },355 },
355 ];356 ];
356}357}
358
359/** Apply a single-link sync without copying unrelated PRs into the event history. */
360export function applyThreadPullRequestLinkUpdate(
361 thread: OrchestrationV2AppThread,
362 updated: ThreadPullRequestLink,
363): OrchestrationV2AppThread {
364 return {
365 ...thread,
366 pullRequests: threadPullRequestsOf(thread).map((link) =>
367 threadPullRequestKeysEqual(link, updated) ? updated : link,
368 ),
369 };
370}