Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
114 changes: 114 additions & 0 deletions apps/server/src/orchestration/decider.import.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,114 @@
import {
CommandId,
MessageId,
ProjectId,
ProviderInstanceId,
ThreadId,
TurnId,
type OrchestrationReadModel,
} from "@t3tools/contracts";
import * as NodeServices from "@effect/platform-node/NodeServices";
import { expect, it } from "@effect/vitest";
import * as Effect from "effect/Effect";

import { decideOrchestrationCommand } from "./decider.ts";

const NOW = "2026-08-21T03:22:43.000Z";

const readModel: OrchestrationReadModel = {
snapshotSequence: 0,
projects: [],
threads: [
{
id: ThreadId.make("thread-1"),
projectId: ProjectId.make("project-1"),
title: "Imported Codex thread",
modelSelection: {
instanceId: ProviderInstanceId.make("codex"),
model: "gpt-5.6-sol",
},
runtimeMode: "full-access",
interactionMode: "default",
branch: null,
worktreePath: null,
latestTurn: null,
createdAt: NOW,
updatedAt: NOW,
archivedAt: null,
settledOverride: null,
settledAt: null,
snoozedUntil: null,
snoozedAt: null,
pinnedAt: null,
pinOrderKey: null,
deletedAt: null,
messages: [],
proposedPlans: [],
activities: [],
checkpoints: [],
session: null,
},
],
updatedAt: NOW,
};

it.layer(NodeServices.layer)("provider history import decider", (it) => {
it.effect("projects an imported assistant message without requesting a provider turn", () =>
Effect.gen(function* () {
const result = yield* decideOrchestrationCommand({
command: {
type: "thread.message.import",
commandId: CommandId.make("provider-import:message-1"),
threadId: ThreadId.make("thread-1"),
messageId: MessageId.make("message-1"),
role: "assistant",
text: "Imported response",
turnId: TurnId.make("turn-1"),
createdAt: NOW,
},
readModel,
});

const events = Array.isArray(result) ? result : [result];
expect(events).toHaveLength(1);
expect(events[0]?.type).toBe("thread.message-sent");
if (events[0]?.type === "thread.message-sent") {
expect(events[0].payload).toMatchObject({
role: "assistant",
text: "Imported response",
turnId: "turn-1",
streaming: false,
});
}
}),
);

it.effect("projects imported user history with its provider turn identity", () =>
Effect.gen(function* () {
const result = yield* decideOrchestrationCommand({
command: {
type: "thread.message.import",
commandId: CommandId.make("provider-import:user-message-1"),
threadId: ThreadId.make("thread-1"),
messageId: MessageId.make("user-message-1"),
role: "user",
text: "Imported prompt",
turnId: TurnId.make("turn-1"),
createdAt: NOW,
},
readModel,
});

const events = Array.isArray(result) ? result : [result];
expect(events[0]).toMatchObject({
type: "thread.message-sent",
payload: {
role: "user",
text: "Imported prompt",
turnId: "turn-1",
streaming: false,
},
});
}),
);
});
27 changes: 27 additions & 0 deletions apps/server/src/orchestration/decider.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1252,6 +1252,33 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"
};
}

case "thread.message.import": {
yield* requireThread({
readModel,
command,
threadId: command.threadId,
});
return {
...(yield* withEventBase({
aggregateKind: "thread",
aggregateId: command.threadId,
occurredAt: command.createdAt,
commandId: command.commandId,
})),
type: "thread.message-sent",
payload: {
threadId: command.threadId,
messageId: command.messageId,
role: command.role,

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Medium orchestration/decider.ts:1272

thread.message.import leaves latestTurn as null, so a newly imported conversation with a user message less than two minutes old is treated as having a queued turn even when a later imported assistant response completes it. During that grace window, thread.settle, thread.snooze, and conditional thread.session.stop are rejected; import handling must mark messages as historical or update latestTurn to represent the completed imported turn.

🤖 Copy this AI Prompt to have your agent fix this:
In file @apps/server/src/orchestration/decider.ts around line 1272:

`thread.message.import` leaves `latestTurn` as `null`, so a newly imported conversation with a user message less than two minutes old is treated as having a queued turn even when a later imported assistant response completes it. During that grace window, `thread.settle`, `thread.snooze`, and conditional `thread.session.stop` are rejected; import handling must mark messages as historical or update `latestTurn` to represent the completed imported turn.

text: command.text,
turnId: command.turnId,
streaming: false,
createdAt: command.createdAt,
updatedAt: command.createdAt,
},
};
}

case "thread.message.assistant.complete": {
yield* requireThread({
readModel,
Expand Down
21 changes: 21 additions & 0 deletions apps/server/src/provider/Layers/CodexAdapter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@ import {
} from "./CodexSessionRuntime.ts";
import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";
import { resolveCodexLaunchArgs } from "./codexLaunchArgs.ts";
import { discoverCodexThreads } from "./CodexThreadDiscovery.ts";
const isCodexAppServerProcessExitedError = Schema.is(CodexErrors.CodexAppServerProcessExitedError);
const isCodexAppServerTransportError = Schema.is(CodexErrors.CodexAppServerTransportError);
const isCodexSessionRuntimeThreadIdMissingError = Schema.is(
Expand Down Expand Up @@ -1866,6 +1867,25 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
})),
);

const discoverPersistedThreads: NonNullable<CodexAdapterShape["discoverPersistedThreads"]> = (
input,
) =>
Effect.scoped(
discoverCodexThreads(codexConfig, options?.environment, input).pipe(
Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, childProcessSpawner),
Effect.timeout("2 minutes"),
Effect.mapError(
(cause) =>
new ProviderAdapterRequestError({
provider: PROVIDER,
method: "thread/list",
detail: cause.message,
Comment on lines +1881 to +1882

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

detail here just copies cause.message, and ProviderAdapterRequestError.message is built from detail, so the wrapper's caller-visible message is derived from the underlying failure instead of stable structural attributes (and can splice in the spawn command line from CodexAppServerSpawnError). The method: "thread/list" label is also inaccurate: this pipeline covers spawn, initialize, thread/list, thread/read, and the 2-minute timeout. Consider a fixed structural detail and keeping the real failure only in cause.

Suggested change
method: "thread/list",
detail: cause.message,
method: "thread/discover",
detail: "Failed to discover persisted Codex threads.",

Posted via Macroscope — Effect Service Conventions

cause,
}),
),
),
);

const rollbackThread: CodexAdapterShape["rollbackThread"] = (threadId, numTurns) => {
if (!Number.isInteger(numTurns) || numTurns < 1) {
return Effect.fail(
Expand Down Expand Up @@ -1988,6 +2008,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
sendTurn,
interruptTurn,
readThread,
discoverPersistedThreads,
rollbackThread,
uploadFeedback,
respondToRequest,
Expand Down
200 changes: 200 additions & 0 deletions apps/server/src/provider/Layers/CodexThreadDiscovery.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,200 @@
import { expect, it } from "@effect/vitest";
import * as Effect from "effect/Effect";
import type * as EffectCodexSchema from "effect-codex-app-server/schema";

import {
readCodexThreadSnapshots,
selectCodexThreadsForRead,
toPersistedThread,
} from "./CodexThreadDiscovery.ts";

type ReadThread = EffectCodexSchema.V2ThreadReadResponse["thread"];
type ListedThread = EffectCodexSchema.V2ThreadListResponse["data"][number];

function makeReadThread(overrides: Record<string, unknown> = {}): ReadThread {
return {
id: "provider-thread-1",
cwd: "/work/project",
name: null,
preview: "Preview title",
createdAt: 1_777_777_700,
updatedAt: 1_777_777_760,
source: "cli",
status: { type: "idle" },
threadSource: null,
cliVersion: "1.2.3",
modelProvider: "openai",
gitInfo: null,
turns: [],
...overrides,
} as unknown as ReadThread;
}

function makeListedThread(id: string, overrides: Record<string, unknown> = {}): ListedThread {
return {
...makeReadThread({ id, turns: undefined }),
ephemeral: false,
...overrides,
} as unknown as ListedThread;
}

it("maps persisted Codex turns to ordered user and assistant history", () => {
const thread = {
id: "provider-thread-1",
cwd: "/work/project",
name: null,
preview: "Investigate the importer\nMore context",
createdAt: 1_777_777_700,
updatedAt: 1_777_777_760,
source: "cli",
status: { type: "idle" },
threadSource: null,
cliVersion: "1.2.3",
modelProvider: "openai",
gitInfo: null,
turns: [
{
id: "turn-1",
startedAt: 1_777_777_710,
completedAt: 1_777_777_750,
status: "completed",
items: [
{
id: "user-1",
type: "userMessage",
content: [
{ type: "text", text: "Investigate the importer" },
{ type: "localImage", path: "/tmp/screenshot.png" },
],
},
{
id: "tool-1",
type: "commandExecution",
command: "pwd",
commandActions: [],
cwd: "/work/project",
status: "completed",
},
{
id: "assistant-1",
type: "agentMessage",
text: "The importer is ready.",
},
],
},
],
} as unknown as EffectCodexSchema.V2ThreadReadResponse["thread"];

expect(toPersistedThread(thread)).toMatchObject({
providerThreadId: "provider-thread-1",
cwd: "/work/project",
title: "Investigate the importer",
sourceMetadata: { source: "cli", cliVersion: "1.2.3" },
messages: [
{
id: "user-1",
sourceOrdinal: 0,
role: "user",
text: "Investigate the importer",
turnId: "turn-1",
},
{
id: "assistant-1",
sourceOrdinal: 1,
role: "assistant",
text: "The importer is ready.",
turnId: "turn-1",
},
],
});
});

it("prefers a persisted name and falls back when all title text is blank", () => {
expect(toPersistedThread(makeReadThread({ name: " Named thread " })).title).toBe(
"Named thread",
);
expect(toPersistedThread(makeReadThread({ name: " ", preview: "\n" })).title).toBe(
"Imported Codex thread",
);
});

it("omits partial and empty messages while applying timestamp fallbacks", () => {
const persisted = toPersistedThread(
makeReadThread({
status: { type: "inProgress" },
turns: [
{
id: "turn-active",
startedAt: 1_777_777_710,
completedAt: null,
status: "inProgress",
items: [{ id: "active-agent", type: "agentMessage", text: "Partial" }],
},
{
id: "turn-complete",
startedAt: null,
completedAt: null,
status: "completed",
items: [
{
id: "empty-user",
type: "userMessage",
content: [{ type: "text", text: " " }],
},
{ id: "empty-agent", type: "agentMessage", text: "" },
{ id: "agent", type: "agentMessage", text: "Complete" },
],
},
],
}),
);

expect(persisted.messages).toEqual([
expect.objectContaining({
id: "agent",
text: "Complete",
createdAt: "2026-05-03T03:09:20.000Z",
}),
]);
expect(persisted.discoveryCursor).toBe("2026-05-03T03:09:20.000Z:inProgress");
});

it("selects only user-visible threads whose discovery cursor changed", () => {
const unchanged = makeListedThread("unchanged");
const statusChanged = makeListedThread("status-changed", { status: { type: "idle" } });
const selected = selectCodexThreadsForRead(
[
unchanged,
statusChanged,
makeListedThread("new"),
makeListedThread("ephemeral", { ephemeral: true }),
makeListedThread("subagent", { source: { subAgent: { threadId: "parent" } } }),
makeListedThread("memory", { threadSource: "memory_consolidation" }),
makeListedThread("native"),
],
{
excludeProviderThreadIds: new Set(["native"]),
cursorByProviderThreadId: new Map([
["unchanged", "2026-05-03T03:09:20.000Z:idle"],
["status-changed", "2026-05-03T03:09:20.000Z:active"],
]),
},
);

expect(selected.map((thread) => thread.id)).toEqual(["status-changed", "new"]);
});

it.effect("keeps readable threads when a sibling disappears between list and read", () =>
Effect.gen(function* () {
const threads = [makeListedThread("deleted"), makeListedThread("healthy")];
const discovered = yield* readCodexThreadSnapshots(threads, (threadId) =>
threadId === "deleted"
? Effect.fail({ _tag: "ThreadReadFailed" as const })
: Effect.succeed({
thread: makeReadThread({ id: threadId, preview: "Healthy thread" }),
} as EffectCodexSchema.V2ThreadReadResponse),
);

expect(discovered.map((thread) => thread.providerThreadId)).toEqual(["healthy"]);
}),
);
Loading
Loading