Skip to content

Commit 0400517

Browse files
debugrrordebugrror
committed
fix(server): preserve imported Codex titles in orchestrator V2
1 parent bf7121d commit 0400517

8 files changed

Lines changed: 933 additions & 12 deletions

File tree

‎apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts‎

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,70 @@ const testLayer = Layer.mergeAll(
4949
),
5050
);
5151

52+
for (const scenario of ["unchanged", "renamed", "regenerating"] as const) {
53+
it.effect(`guards an imported title repair against concurrent title changes (${scenario})`, () =>
54+
Effect.gen(function* () {
55+
const orchestrator = yield* Orchestrator.OrchestratorV2;
56+
const threadId = ThreadId.make("thread:guarded-import-title");
57+
yield* orchestrator.dispatch({
58+
type: "thread.create",
59+
commandId: CommandId.make("create-import-title"),
60+
threadId,
61+
projectId: ProjectId.make("project:import-title"),
62+
title: "<recommended_plugins>",
63+
modelSelection,
64+
runtimeMode: "full-access",
65+
interactionMode: "default",
66+
branch: null,
67+
worktreePath: null,
68+
createdBy: "user",
69+
creationSource: "web",
70+
});
71+
if (scenario === "renamed") {
72+
yield* orchestrator.dispatch({
73+
type: "thread.metadata.update",
74+
commandId: CommandId.make("manual-import-title"),
75+
threadId,
76+
title: "My custom title",
77+
});
78+
}
79+
if (scenario === "regenerating") {
80+
yield* orchestrator.dispatch({
81+
type: "thread.metadata.update",
82+
commandId: CommandId.make("regenerate-import-title"),
83+
threadId,
84+
regenerateTitle: true,
85+
});
86+
}
87+
const repaired = yield* orchestrator
88+
.dispatch({
89+
type: "thread.metadata.update",
90+
commandId: CommandId.make("repair-import-title"),
91+
threadId,
92+
title: "Prototype MetaApi trade replication",
93+
expectedTitle: "<recommended_plugins>",
94+
expectedTitleRegenerationRequestId: null,
95+
})
96+
.pipe(Effect.exit);
97+
assert.equal(repaired._tag, scenario === "unchanged" ? "Success" : "Failure");
98+
assert.equal(
99+
(yield* orchestrator.getThreadRecords(threadId, [])).thread.title,
100+
scenario === "renamed"
101+
? "My custom title"
102+
: scenario === "regenerating"
103+
? "<recommended_plugins>"
104+
: "Prototype MetaApi trade replication",
105+
);
106+
if (scenario === "regenerating") {
107+
assert.equal(
108+
(yield* orchestrator.getThreadRecords(threadId, [])).thread.titleRegeneration?.requestId,
109+
"regenerate-import-title",
110+
);
111+
}
112+
}).pipe(Effect.provide(testLayer)),
113+
);
114+
}
115+
52116
it.effect(
53117
"dispatches metadata, queue resume and request controls without hydrating unrelated history",
54118
() =>

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

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2199,6 +2199,28 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
21992199
cause: `Thread ${command.threadId} is deleted.`,
22002200
});
22012201
}
2202+
if (
2203+
command.type === "thread.metadata.update" &&
2204+
command.expectedTitle !== undefined &&
2205+
command.expectedTitle !== thread.title
2206+
) {
2207+
return yield* new OrchestratorDispatchError({
2208+
commandId: command.commandId,
2209+
commandType: command.type,
2210+
cause: `Thread ${command.threadId} title changed before the metadata update could be applied.`,
2211+
});
2212+
}
2213+
if (
2214+
command.type === "thread.metadata.update" &&
2215+
command.expectedTitleRegenerationRequestId !== undefined &&
2216+
command.expectedTitleRegenerationRequestId !== (thread.titleRegeneration?.requestId ?? null)
2217+
) {
2218+
return yield* new OrchestratorDispatchError({
2219+
commandId: command.commandId,
2220+
commandType: command.type,
2221+
cause: `Thread ${command.threadId} title regeneration changed before the metadata update could be applied.`,
2222+
});
2223+
}
22022224
if (
22032225
command.type === "thread.metadata.update" &&
22042226
command.expectedWorktreePath !== undefined &&

‎apps/server/src/project/AgentSessionImporter.test.ts‎

Lines changed: 148 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,13 @@
11
import { expect, it } from "@effect/vitest";
22
import {
3+
type AgentSessionImportSource,
34
ProjectId,
45
ProviderInstanceId,
56
ThreadId,
67
type OrchestrationV2DomainEvent,
78
} from "@t3tools/contracts";
89
import * as Effect from "effect/Effect";
10+
import * as DateTime from "effect/DateTime";
911
import * as Layer from "effect/Layer";
1012
import * as Option from "effect/Option";
1113
import * as Stream from "effect/Stream";
@@ -23,6 +25,152 @@ const providerInstanceId = ProviderInstanceId.make("codex");
2325
const providerSessionId = "native-codex-thread";
2426
const threadId = ThreadId.make(`import:${providerInstanceId}:${providerSessionId}`);
2527

28+
for (const scenario of [
29+
"canonical",
30+
"fallback",
31+
"manual-title",
32+
"other-project",
33+
"native-history",
34+
"other-provider",
35+
"repair-failure",
36+
"changed-transcript",
37+
"changed-repair-failure",
38+
"regenerating",
39+
"deleted",
40+
] as const) {
41+
it.effect(`repairs only eligible imported Codex titles (${scenario})`, () => {
42+
let title = scenario === "manual-title" ? "My custom title" : "<recommended_plugins>";
43+
const originalTitle = title;
44+
const source: AgentSessionImportSource = {
45+
provider: "codex",
46+
providerInstanceId,
47+
providerSessionId,
48+
filePath: "/tmp/native-codex-thread.jsonl",
49+
size: 100,
50+
mtimeMs: 2,
51+
device: 3,
52+
inode: 4,
53+
birthtimeMs: 1,
54+
};
55+
const messages = [
56+
{ role: "user", text: "<recommended_plugins>\nInjected setup\n</recommended_plugins>" },
57+
{ role: "user", text: "Build a trade replication prototype" },
58+
];
59+
const recorded: Array<unknown> = [];
60+
const renamed: Array<unknown> = [];
61+
const canonicalTitle = "Prototype MetaApi trade replication";
62+
const scanner = AgentSessionScanner.AgentSessionScanner.of({
63+
scan: Effect.die("unused"),
64+
recentThreads: () =>
65+
Stream.succeed(
66+
scenario === "changed-transcript" || scenario === "changed-repair-failure"
67+
? {
68+
_tag: "Importable",
69+
source,
70+
thread: {
71+
source: "codex",
72+
providerInstanceId,
73+
providerSessionId,
74+
title: canonicalTitle,
75+
model: null,
76+
createdAt: "2026-09-01T10:00:00.000Z",
77+
updatedAt: "2026-09-01T10:01:00.000Z",
78+
messages: [],
79+
},
80+
}
81+
: {
82+
_tag: "AlreadyImported",
83+
source,
84+
canonicalTitle: scenario === "fallback" ? null : canonicalTitle,
85+
},
86+
),
87+
});
88+
const testLayer = AgentSessionImporter.layer.pipe(
89+
Layer.provide(
90+
Layer.mergeAll(
91+
Layer.succeed(AgentSessionScanner.AgentSessionScanner, scanner),
92+
Layer.mock(ProjectService.ProjectService)({
93+
getById: () =>
94+
Effect.succeed(
95+
Option.some({ id: projectId, workspaceRoot: "/workspace/project" } as never),
96+
),
97+
}),
98+
Layer.mock(Orchestrator.OrchestratorV2)({
99+
getThreadRecords: () =>
100+
Effect.succeed({
101+
thread: {
102+
id: threadId,
103+
projectId: scenario === "other-project" ? ProjectId.make("other") : projectId,
104+
title,
105+
providerInstanceId:
106+
scenario === "other-provider"
107+
? ProviderInstanceId.make("claudeAgent")
108+
: providerInstanceId,
109+
historyOrigin: scenario === "native-history" ? "native" : "v1_import",
110+
deletedAt:
111+
scenario === "deleted" ? DateTime.makeUnsafe("2026-09-02T10:00:00.000Z") : null,
112+
titleRegeneration:
113+
scenario === "regenerating" ? { requestId: "regenerate-title" } : null,
114+
},
115+
messages: messages.map((message, index) => ({
116+
...message,
117+
id: `${threadId}:${String(index).padStart(6, "0")}`,
118+
createdAt: DateTime.makeUnsafe("2026-09-01T10:00:00.000Z"),
119+
})),
120+
} as never),
121+
dispatch: (command) => {
122+
if (scenario === "repair-failure" || scenario === "changed-repair-failure") {
123+
return Effect.fail(new Orchestrator.OrchestratorProjectionError({ threadId }));
124+
}
125+
return Effect.sync(() => {
126+
if (command.type === "thread.metadata.update" && command.title !== undefined) {
127+
title = command.title;
128+
renamed.push(command);
129+
}
130+
return { sequence: 1, storedEvents: [] };
131+
});
132+
},
133+
}),
134+
Layer.mock(EventSink.EventSinkV2)({ write: () => Effect.die("history must not change") }),
135+
Layer.mock(ProviderSessionRuntime.ProviderSessionRuntimeRepository)({
136+
list: () => Effect.succeed([]),
137+
recordImportedTranscript: (input) => Effect.sync(() => void recorded.push(input)),
138+
}),
139+
IdAllocator.layer,
140+
),
141+
),
142+
);
143+
return Effect.gen(function* () {
144+
const importer = yield* AgentSessionImporter.AgentSessionImporter;
145+
expect(yield* importer.importRecentAgentThreads({ projectId })).toEqual({
146+
importedCount: 1,
147+
skippedCount: 0,
148+
});
149+
const shouldRepair = ["canonical", "fallback", "changed-transcript"].includes(scenario);
150+
expect(title).toBe(
151+
shouldRepair
152+
? scenario === "fallback"
153+
? "Build a trade replication prototype"
154+
: canonicalTitle
155+
: originalTitle,
156+
);
157+
expect(renamed).toHaveLength(shouldRepair ? 1 : 0);
158+
if (shouldRepair)
159+
expect(renamed[0]).toMatchObject({
160+
expectedTitle: originalTitle,
161+
expectedTitleRegenerationRequestId: null,
162+
});
163+
expect(recorded).toHaveLength(
164+
scenario === "changed-transcript" || scenario === "changed-repair-failure" ? 1 : 0,
165+
);
166+
expect(messages.map((message) => message.text)).toEqual([
167+
"<recommended_plugins>\nInjected setup\n</recommended_plugins>",
168+
"Build a trade replication prototype",
169+
]);
170+
}).pipe(Effect.provide(testLayer));
171+
});
172+
}
173+
26174
it.effect("imports messages once and preserves the provider native resume binding", () => {
27175
const writes: Array<ReadonlyArray<OrchestrationV2DomainEvent>> = [];
28176
const upserts: Array<unknown> = [];

‎apps/server/src/project/AgentSessionImporter.ts‎

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -172,6 +172,55 @@ const make = Effect.gen(function* () {
172172
const eventSink = yield* EventSink.EventSinkV2;
173173
const idAllocator = yield* IdAllocator.IdAllocatorV2;
174174
const runtimes = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository;
175+
176+
/** Repair legacy import placeholders without replacing user-chosen titles or history. */
177+
const repairImportedCodexTitle = Effect.fn("repairImportedCodexTitleV2")(
178+
function* (
179+
threadId: ThreadId,
180+
projectId: ProjectId,
181+
providerInstanceId: AgentSessionImportSource["providerInstanceId"],
182+
canonicalTitle: string | null,
183+
) {
184+
const { thread } = yield* orchestrator.getThreadRecords(threadId, []);
185+
if (
186+
thread.projectId !== projectId ||
187+
thread.historyOrigin !== "v1_import" ||
188+
thread.providerInstanceId !== providerInstanceId ||
189+
thread.deletedAt !== null ||
190+
thread.titleRegeneration != null ||
191+
!AgentSessionScanner.hasLegacyCodexContextTitle(thread.title)
192+
) {
193+
return;
194+
}
195+
let replacementTitle = canonicalTitle;
196+
if (replacementTitle === null) {
197+
const records = yield* orchestrator.getThreadRecords(threadId, ["messages"], {
198+
messageRoles: ["user"],
199+
});
200+
for (const message of records.messages) {
201+
if (!message.id.startsWith(`${threadId}:`)) continue;
202+
replacementTitle = AgentSessionScanner.deriveImportedCodexTitle(message.text);
203+
if (replacementTitle !== null) break;
204+
}
205+
}
206+
if (replacementTitle === null || replacementTitle === thread.title) return;
207+
yield* orchestrator.dispatch({
208+
type: "thread.metadata.update",
209+
commandId: yield* idAllocator.allocate.command({
210+
fixtureName: "agent-session-import",
211+
commandName: "repair-codex-title",
212+
}),
213+
threadId,
214+
title: replacementTitle,
215+
expectedTitle: thread.title,
216+
expectedTitleRegenerationRequestId: null,
217+
});
218+
},
219+
Effect.catch((cause) =>
220+
Effect.logWarning("Could not repair an imported Codex thread title", { cause }),
221+
),
222+
);
223+
175224
const importRecentAgentThreads = Effect.fn("importRecentAgentThreadsV2")(function* (
176225
input: AgentSessionImportInput,
177226
) {
@@ -227,6 +276,14 @@ const make = Effect.gen(function* () {
227276
`import:${source.providerInstanceId}:${source.providerSessionId}`,
228277
);
229278
if (outcome._tag === "AlreadyImported") {
279+
if (source.provider === "codex") {
280+
yield* repairImportedCodexTitle(
281+
threadId,
282+
input.projectId,
283+
source.providerInstanceId,
284+
outcome.canonicalTitle ?? null,
285+
);
286+
}
230287
importedThreadIds.add(threadId);
231288
importedCount += 1;
232289
return;
@@ -261,6 +318,14 @@ const make = Effect.gen(function* () {
261318
if (existing.value.thread.historyOrigin !== "v1_import") {
262319
return yield* new AgentSessionThreadModifiedError({ threadId });
263320
}
321+
if (thread.source === "codex") {
322+
yield* repairImportedCodexTitle(
323+
threadId,
324+
input.projectId,
325+
thread.providerInstanceId,
326+
thread.title,
327+
);
328+
}
264329
yield* runtimes.recordImportedTranscript({ threadId, source });
265330
return true;
266331
}

0 commit comments

Comments
 (0)