Skip to content

Commit 4d73e1d

Browse files
juliusmarmingejuliusmarminge
authored
fix(server): persist only changed pull request links (#19248)
1 parent d765d03 commit 4d73e1d

10 files changed

Lines changed: 186 additions & 2 deletions

File tree

‎apps/server/src/orchestration-v2/Orchestrator.ts‎

Lines changed: 21 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3251,6 +3251,27 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
32513251
};
32523252
}
32533253
})();
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+
}
32543275
const eventType = (() => {
32553276
switch (command.type) {
32563277
case "thread.archive":
@@ -3282,9 +3303,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
32823303
return "thread.metadata-updated" as const;
32833304
case "thread.pull-request.link":
32843305
case "thread.pull-request.unlink":
3285-
case "thread.pull-request-link.sync":
32863306
case "thread.pull-request.watch":
3287-
case "thread.pull-request-watch.sync":
32883307
case "thread.pull-request.sync":
32893308
return "thread.pull-request-synced" as const;
32903309
case "thread.runtime-mode.set":

‎apps/server/src/orchestration-v2/ProjectionStore.ts‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
import { applyThreadPullRequestLinkUpdate } from "@t3tools/shared/threadPullRequests";
12
import { projectTurnItemForWire } from "./WireProjection.ts";
23
import * as Stream from "effect/Stream";
34
import { makeThreadFind, findProjectedThreadItems } from "./ThreadFind.ts";
@@ -718,6 +719,11 @@ export function applyToProjection(
718719
};
719720

720721
switch (event.type) {
722+
case "thread.pull-request-link-updated":
723+
return {
724+
...base,
725+
thread: applyThreadPullRequestLinkUpdate(projection.thread, event.payload),
726+
};
721727
case "thread.created":
722728
case "thread.archived":
723729
case "thread.unarchived":
@@ -1823,6 +1829,15 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
18231829
const apply: ProjectionStoreV2Shape["apply"] = (event) =>
18241830
Effect.gen(function* () {
18251831
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+
}
18261841
case "thread.created":
18271842
case "thread.archived":
18281843
case "thread.unarchived":

‎apps/server/src/orchestration-v2/PullRequestSyncReactor.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -440,6 +440,8 @@ export const make = Effect.gen(function* () {
440440
yield* forkParked(
441441
Stream.runForEach(events, (event) => {
442442
switch (event.type) {
443+
case "thread.pull-request-link-updated":
444+
return event.payload.snapshot === null ? requestSync(event.payload) : Effect.void;
443445
case "thread.pull-request-synced":
444446
return Effect.forEach(
445447
visibleThreadPullRequests(event.payload.pullRequests ?? []).filter(

‎apps/server/src/orchestration-v2/ThreadSettlementService.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -606,6 +606,7 @@ export const make = Effect.gen(function* () {
606606
switch (event.type) {
607607
case "thread.settled":
608608
return cleanUpSettledThread(event.threadId);
609+
case "thread.pull-request-link-updated":
609610
case "thread.pull-request-synced":
610611
case "provider-session.detached":
611612
return worker.enqueue(event.threadId);

‎apps/server/src/orchestration-v2/runtimeLayer.test.ts‎

Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4860,6 +4860,93 @@ it.layer(layerSharedApplicationDataPlaneTest)("visited projection", (it) => {
48604860
);
48614861
});
48624862

4863+
it.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+
48634950
it.layer(layerSharedApplicationDataPlaneTest)("shared application data plane", (it) => {
48644951
it.effect("orders retained project transactions and V2 thread transactions in one source", () =>
48654952
Effect.gen(function* () {

‎apps/server/src/relay/AgentAwarenessRelay.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,7 @@ export function shouldPublishAgentAwarenessEvent(
9797
case "thread.unarchived":
9898
case "thread.deleted":
9999
case "thread.metadata-updated":
100+
case "thread.pull-request-link-updated":
100101
case "thread.pull-request-synced":
101102
case "thread.model-selection-updated":
102103
case "thread.provider-switched":

‎packages/client-runtime/src/state/orchestrationV2Projection.test.ts‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import { describe, expect, it } from "vite-plus/test";
22
import {
3+
type ThreadPullRequestLink,
34
type OrchestrationV2DomainEvent,
45
type OrchestrationV2Run,
56
type OrchestrationV2ThreadProjection,
@@ -105,6 +106,34 @@ const emptyProjection = {
105106
} as OrchestrationV2ThreadProjection;
106107

107108
describe("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+
108137
it("keeps live token usage when the terminal provider turn omits it", () => {
109138
const providerTurnId = ProviderTurnId.make("provider-turn-reducer");
110139
const running = {

‎packages/client-runtime/src/state/orchestrationV2Projection.ts‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
import { applyThreadPullRequestLinkUpdate } from "@t3tools/shared/threadPullRequests";
12
import type {
23
OrchestrationV2DomainEvent,
34
OrchestrationV2Run,
@@ -201,6 +202,11 @@ export function applyOrchestrationV2ProjectionEvent(
201202
const latestLocalTurnOrdinal = options?.latestLocalTurnOrdinal;
202203
const base = { ...projection, updatedAt: event.occurredAt };
203204
switch (event.type) {
205+
case "thread.pull-request-link-updated":
206+
return {
207+
...base,
208+
thread: applyThreadPullRequestLinkUpdate(projection.thread, event.payload),
209+
};
204210
case "thread.created":
205211
case "thread.archived":
206212
case "thread.unarchived":

‎packages/contracts/src/orchestrationV2.ts‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1765,6 +1765,11 @@ export const OrchestrationV2DomainEvent = Schema.Union([
17651765
]),
17661766
payload: OrchestrationV2AppThread,
17671767
}),
1768+
Schema.Struct({
1769+
...OrchestrationV2EventBase.fields,
1770+
type: Schema.Literal("thread.pull-request-link-updated"),
1771+
payload: ThreadPullRequestLink,
1772+
}),
17681773
Schema.Struct({
17691774
...OrchestrationV2EventBase.fields,
17701775
type: Schema.Literal("run.created"),
@@ -2585,6 +2590,11 @@ export const OrchestrationV2DomainEventJson = Schema.Union([
25852590
]),
25862591
payload: OrchestrationV2AppThreadJson,
25872592
}),
2593+
Schema.Struct({
2594+
...OrchestrationV2JsonEventBaseFields,
2595+
type: Schema.Literal("thread.pull-request-link-updated"),
2596+
payload: ThreadPullRequestLink,
2597+
}),
25882598
Schema.Struct({
25892599
...OrchestrationV2JsonEventBaseFields,
25902600
type: Schema.Literal("run.created"),

‎packages/shared/src/threadPullRequests.ts‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import type {
2+
OrchestrationV2AppThread,
23
RepositoryIdentity,
34
SourceControlProviderKind,
45
ThreadLinkedPullRequest,
@@ -354,3 +355,16 @@ export function threadPullRequestsOf(thread: {
354355
},
355356
];
356357
}
358+
359+
/** Apply a single-link sync without copying unrelated PRs into the event history. */
360+
export 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+
}

0 commit comments

Comments
 (0)