mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-15 15:13:48 -06:00
45dd558d92
* fix(codex): preserve thread ownership across session lifecycles * test(codex): keep canonical context-engine session fixtures * fix(codex): fence replacement thread rollback by ownership * fix(codex): fence thread handoffs by physical client ownership * fix(codex): derive conversation privacy from its source session * fix(codex): claim active native subagents at their lifecycle owner * fix(codex): preserve conversation ownership when detach fails * fix(codex): fence stale native child close notifications * fix(codex): preserve active child ownership during rollback * test(codex): keep lifecycle fixtures aligned with public contracts * fix(codex): preserve rollback failure causes across ownership recovery * fix(codex): retain aggregate rollback causes through lint analysis * fix(codex): release native children when idle retention fails
4184 lines
147 KiB
TypeScript
4184 lines
147 KiB
TypeScript
// Codex tests cover conversation binding plugin behavior.
|
|
import fs from "node:fs/promises";
|
|
import os from "node:os";
|
|
import path from "node:path";
|
|
import {
|
|
clearActiveEmbeddedRun,
|
|
setActiveEmbeddedRun,
|
|
} from "openclaw/plugin-sdk/agent-harness-runtime";
|
|
import type { ExecApprovalsFile } from "openclaw/plugin-sdk/exec-approvals-runtime";
|
|
import type { PluginConversationBinding } from "openclaw/plugin-sdk/plugin-entry";
|
|
import { upsertSessionEntry } from "openclaw/plugin-sdk/session-store-runtime";
|
|
import { appendSessionTranscriptMessageByIdentity } from "openclaw/plugin-sdk/session-transcript-runtime";
|
|
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
|
|
|
const sharedClientMocks = vi.hoisted(() => ({
|
|
getSharedCodexAppServerClient: vi.fn(),
|
|
retainSharedCodexAppServerClientByInstanceId: vi.fn(
|
|
(_clientId?: string): { client: unknown; release: () => void } | undefined => undefined,
|
|
),
|
|
clearSharedCodexAppServerClientIfCurrentAndUnclaimed: vi.fn((_client: unknown) => ({
|
|
found: false,
|
|
closed: false,
|
|
})),
|
|
retireSharedCodexAppServerClientIfCurrent: vi.fn(
|
|
(_client: unknown): { activeLeases: number; closed: boolean } | undefined => undefined,
|
|
),
|
|
clearSharedCodexAppServerClientIfCurrent: vi.fn((_client: unknown) => false),
|
|
}));
|
|
|
|
const publicBindingMocks = vi.hoisted(() => ({
|
|
resolveByConversation: vi.fn((_conversation: unknown): { bindingId: string } | null => ({
|
|
bindingId: "binding-1",
|
|
})),
|
|
}));
|
|
|
|
const execApprovalsRuntimeMocks = vi.hoisted(() => ({
|
|
loadExecApprovals: vi.fn<() => ExecApprovalsFile>(() => ({ version: 1, agents: {} })),
|
|
}));
|
|
|
|
const agentRuntimeMocks = vi.hoisted(() => ({
|
|
ensureAuthProfileStore: vi.fn(),
|
|
loadAuthProfileStoreForSecretsRuntime: vi.fn(),
|
|
resolveApiKeyForProfile: vi.fn(),
|
|
resolveAuthProfileOrder: vi.fn(),
|
|
resolveDefaultAgentDir: vi.fn(() => "/agent"),
|
|
resolvePersistedAuthProfileOwnerAgentDir: vi.fn(),
|
|
resolveProviderIdForAuth: vi.fn((provider: string, _lookup?: { config?: unknown }) => provider),
|
|
resolveSessionAgentIds: vi.fn(() => ({ defaultAgentId: "main", sessionAgentId: "main" })),
|
|
saveAuthProfileStore: vi.fn(),
|
|
}));
|
|
|
|
const codexRequirementsTomlMock = vi.hoisted(() => vi.fn<() => string | undefined>());
|
|
const resolveSandboxContextMock = vi.hoisted(() =>
|
|
vi.fn<(...args: unknown[]) => Promise<{ enabled: boolean } | null>>(async () => null),
|
|
);
|
|
|
|
vi.mock("node:fs", async (importOriginal) => {
|
|
const actual = await importOriginal<typeof import("node:fs")>();
|
|
return {
|
|
...actual,
|
|
readFileSync(filePath: string | URL | number, options?: BufferEncoding | object | null) {
|
|
if (filePath === "/etc/codex/requirements.toml") {
|
|
const content = codexRequirementsTomlMock();
|
|
if (content !== undefined) {
|
|
return content;
|
|
}
|
|
}
|
|
return actual.readFileSync(filePath, options);
|
|
},
|
|
};
|
|
});
|
|
|
|
vi.mock("openclaw/plugin-sdk/agent-harness-runtime", async (importOriginal) => {
|
|
const actual = await importOriginal<typeof import("openclaw/plugin-sdk/agent-harness-runtime")>();
|
|
return {
|
|
...actual,
|
|
resolveSandboxContext: resolveSandboxContextMock,
|
|
};
|
|
});
|
|
|
|
vi.mock("openclaw/plugin-sdk/conversation-binding-runtime", async (importOriginal) => {
|
|
const actual =
|
|
await importOriginal<typeof import("openclaw/plugin-sdk/conversation-binding-runtime")>();
|
|
return {
|
|
...actual,
|
|
getSessionBindingService: () => ({
|
|
resolveByConversation: publicBindingMocks.resolveByConversation,
|
|
}),
|
|
};
|
|
});
|
|
|
|
vi.mock("./app-server/shared-client.js", () => ({
|
|
...sharedClientMocks,
|
|
isCodexAppServerStartSelectionChangedError: (error: unknown) =>
|
|
error instanceof Error &&
|
|
"code" in error &&
|
|
error.code === "CODEX_APP_SERVER_START_SELECTION_CHANGED",
|
|
getLeasedSharedCodexAppServerClient: async (...args: unknown[]) => {
|
|
const client = (await sharedClientMocks.getSharedCodexAppServerClient(...args)) as {
|
|
getInstanceId?: () => string;
|
|
addCloseHandler?: () => () => void;
|
|
};
|
|
client.getInstanceId ??= () => "test-client";
|
|
client.addCloseHandler ??= () => () => undefined;
|
|
return client;
|
|
},
|
|
releaseLeasedSharedCodexAppServerClient: vi.fn(),
|
|
releaseCodexAppServerClientLease: vi.fn((lease: { client?: unknown }) => {
|
|
lease.client = undefined;
|
|
}),
|
|
withLeasedCodexAppServerClientStartSelectionRetry: async (params: {
|
|
lease: { client?: unknown };
|
|
run: (client: unknown) => Promise<unknown>;
|
|
}) => await params.run(params.lease.client),
|
|
}));
|
|
vi.mock("openclaw/plugin-sdk/exec-approvals-runtime", async (importOriginal) => {
|
|
const actual =
|
|
await importOriginal<typeof import("openclaw/plugin-sdk/exec-approvals-runtime")>();
|
|
return {
|
|
...actual,
|
|
loadExecApprovals: execApprovalsRuntimeMocks.loadExecApprovals,
|
|
};
|
|
});
|
|
vi.mock("openclaw/plugin-sdk/agent-runtime", () => agentRuntimeMocks);
|
|
|
|
import {
|
|
consumeCodexAppServerLiveThread,
|
|
ensureCodexAppServerClientRuntime,
|
|
isCodexAppServerLiveThreadClaimed,
|
|
retainCodexAppServerLiveThread,
|
|
} from "./app-server/client-runtime.js";
|
|
import type { CodexAppServerClient } from "./app-server/client.js";
|
|
import { resolveCodexAppServerRuntimeOptions } from "./app-server/config.js";
|
|
import { codexNativeSubagentMonitorRuntime } from "./app-server/native-subagent-monitor.js";
|
|
import type { JsonValue } from "./app-server/protocol.js";
|
|
import {
|
|
readCodexAppServerBinding,
|
|
registerCodexTestSessionIdentity,
|
|
resetCodexTestBindingStore,
|
|
testCodexAppServerBindingStore,
|
|
type CodexAppServerThreadBinding,
|
|
writeCodexAppServerBinding,
|
|
} from "./app-server/session-binding.test-helpers.js";
|
|
import { createClientHarness } from "./app-server/test-support.js";
|
|
import { withCodexConversationThreadActivity } from "./app-server/thread-ownership.js";
|
|
import { getCodexAppServerTurnRouter } from "./app-server/turn-router.js";
|
|
import { legacyCodexConversationBindingId } from "./conversation-binding-data.js";
|
|
import { codexConversationBindingRuntime } from "./conversation-binding.js";
|
|
import { readCodexConversationActiveTurn } from "./conversation-control.js";
|
|
|
|
const handleCodexConversationBindingResolvedImpl =
|
|
codexConversationBindingRuntime.handleBindingResolved;
|
|
const handleCodexConversationInboundClaimImpl = codexConversationBindingRuntime.handleInboundClaim;
|
|
const startCodexConversationThreadImpl = codexConversationBindingRuntime.startThread;
|
|
|
|
function testConversationIdentity(sessionFile: string) {
|
|
return {
|
|
kind: "conversation" as const,
|
|
bindingId: legacyCodexConversationBindingId(sessionFile),
|
|
};
|
|
}
|
|
|
|
async function writeTestConversationBinding(
|
|
sessionFile: string,
|
|
binding: CodexAppServerThreadBinding,
|
|
): Promise<void> {
|
|
await testCodexAppServerBindingStore.mutate(testConversationIdentity(sessionFile), {
|
|
kind: "set",
|
|
binding: { clientId: "test-client", ...binding },
|
|
});
|
|
}
|
|
|
|
async function readTestConversationBinding(sessionFile: string) {
|
|
return await testCodexAppServerBindingStore.read(testConversationIdentity(sessionFile));
|
|
}
|
|
|
|
function boundConversationClaim(sessionFile: string, sessionKey?: string) {
|
|
const pluginBinding: PluginConversationBinding = {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "telegram",
|
|
accountId: "default",
|
|
conversationId: "5185575566",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session" as const,
|
|
version: 1 as const,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
},
|
|
};
|
|
return {
|
|
event: {
|
|
content: "continue",
|
|
bodyForAgent: "continue",
|
|
channel: "telegram",
|
|
isGroup: false,
|
|
commandAuthorized: true,
|
|
...(sessionKey ? { sessionKey } : {}),
|
|
},
|
|
ctx: {
|
|
channelId: "telegram",
|
|
...(sessionKey ? { sessionKey } : {}),
|
|
pluginBinding,
|
|
},
|
|
};
|
|
}
|
|
|
|
async function createSameThreadClientMigrationFixture(
|
|
sessionFile: string,
|
|
options: { rejectOldRelease: boolean; owner: "session" | "conversation" },
|
|
) {
|
|
const binding = {
|
|
threadId: "thread-migrated",
|
|
clientId: "client-before-migration",
|
|
cwd: tempDir,
|
|
};
|
|
const readOwner =
|
|
options.owner === "session"
|
|
? async () => readCodexAppServerBinding(sessionFile)
|
|
: async () => readTestConversationBinding(sessionFile);
|
|
if (options.owner === "session") {
|
|
await writeCodexAppServerBinding(sessionFile, binding);
|
|
} else {
|
|
await writeTestConversationBinding(sessionFile, binding);
|
|
}
|
|
const operations: string[] = [];
|
|
const ownerDuringRelease: Array<string | undefined> = [];
|
|
const notificationHandlers = new Set<(notification: unknown) => void>();
|
|
const previousClient = {
|
|
getInstanceId: () => "client-before-migration",
|
|
request: vi.fn(async (method: string) => {
|
|
operations.push(`previous:${method}`);
|
|
ownerDuringRelease.push((await readOwner())?.clientId);
|
|
if (options.rejectOldRelease) {
|
|
throw new Error("previous physical client unsubscribe failed");
|
|
}
|
|
return {};
|
|
}),
|
|
addNotificationHandler: vi.fn(() => () => undefined),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
addCloseHandler: vi.fn(() => () => undefined),
|
|
} as unknown as CodexAppServerClient;
|
|
const replacementClient = {
|
|
getInstanceId: () => "client-after-migration",
|
|
request: vi.fn(async (method: string) => {
|
|
operations.push(`replacement:${method}`);
|
|
if (method === "thread/resume") {
|
|
return conversationThreadStartResult("thread-migrated");
|
|
}
|
|
if (method === "thread/unsubscribe") {
|
|
return {};
|
|
}
|
|
if (method === "turn/start") {
|
|
queueMicrotask(() => {
|
|
for (const handler of notificationHandlers) {
|
|
handler({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "thread-migrated",
|
|
turn: {
|
|
id: "turn-migrated",
|
|
status: "completed",
|
|
items: [{ type: "agentMessage", id: "answer", text: "Migrated reply" }],
|
|
},
|
|
},
|
|
});
|
|
}
|
|
});
|
|
return { turn: { id: "turn-migrated" } };
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}),
|
|
addNotificationHandler: vi.fn((handler: (notification: unknown) => void) => {
|
|
notificationHandlers.add(handler);
|
|
return () => notificationHandlers.delete(handler);
|
|
}),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
addCloseHandler: vi.fn(() => () => undefined),
|
|
} as unknown as CodexAppServerClient;
|
|
ensureCodexAppServerClientRuntime(previousClient, { agentDir: tempDir });
|
|
ensureCodexAppServerClientRuntime(replacementClient, { agentDir: tempDir });
|
|
await expect(retainCodexAppServerLiveThread(previousClient, "thread-migrated")).resolves.toBe(
|
|
true,
|
|
);
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue(replacementClient);
|
|
sharedClientMocks.retainSharedCodexAppServerClientByInstanceId.mockImplementation((clientId) =>
|
|
clientId === previousClient.getInstanceId()
|
|
? { client: previousClient, release: vi.fn() }
|
|
: undefined,
|
|
);
|
|
return { previousClient, replacementClient, operations, ownerDuringRelease, readOwner };
|
|
}
|
|
|
|
function handleCodexConversationInboundClaim(
|
|
event: Parameters<typeof handleCodexConversationInboundClaimImpl>[0],
|
|
ctx: Parameters<typeof handleCodexConversationInboundClaimImpl>[1],
|
|
options: Omit<Parameters<typeof handleCodexConversationInboundClaimImpl>[2], "bindingStore"> = {},
|
|
) {
|
|
return handleCodexConversationInboundClaimImpl({ senderIsOwner: true, ...event }, ctx, {
|
|
...options,
|
|
bindingStore: testCodexAppServerBindingStore,
|
|
});
|
|
}
|
|
|
|
function startCodexConversationThread(
|
|
params: Omit<Parameters<typeof startCodexConversationThreadImpl>[0], "bindingStore">,
|
|
) {
|
|
return startCodexConversationThreadImpl({
|
|
...params,
|
|
bindingStore: testCodexAppServerBindingStore,
|
|
});
|
|
}
|
|
|
|
function handleCodexConversationBindingResolved(
|
|
event: Parameters<typeof handleCodexConversationBindingResolvedImpl>[0],
|
|
) {
|
|
return handleCodexConversationBindingResolvedImpl(event, {
|
|
bindingStore: testCodexAppServerBindingStore,
|
|
});
|
|
}
|
|
|
|
let tempDir: string;
|
|
|
|
const NETWORK_PROXY_PLUGIN_CONFIG = {
|
|
appServer: {
|
|
networkProxy: {
|
|
enabled: true,
|
|
domains: { "api.openai.com": "allow" },
|
|
allowUpstreamProxy: true,
|
|
proxyUrl: "http://127.0.0.1:3128",
|
|
},
|
|
},
|
|
};
|
|
const NETWORK_PROXY_RUNTIME = resolveCodexAppServerRuntimeOptions({
|
|
env: {},
|
|
requirementsToml: null,
|
|
pluginConfig: NETWORK_PROXY_PLUGIN_CONFIG,
|
|
});
|
|
const NETWORK_PROXY_PROFILE_NAME = NETWORK_PROXY_RUNTIME.networkProxy?.profileName ?? "missing";
|
|
const NETWORK_PROXY_CONFIG_PATCH = NETWORK_PROXY_RUNTIME.networkProxy?.configPatch ?? {};
|
|
const NETWORK_PROXY_CONFIG_FINGERPRINT =
|
|
NETWORK_PROXY_RUNTIME.networkProxy?.configFingerprint ?? "missing";
|
|
|
|
function conversationThreadStartResult(threadId: string) {
|
|
return {
|
|
approvalPolicy: "never",
|
|
approvalsReviewer: "user",
|
|
cwd: tempDir,
|
|
model: "gpt-5.4-mini",
|
|
modelProvider: "openai",
|
|
sandbox: { type: "workspaceWrite", networkAccess: false },
|
|
serviceTier: null,
|
|
activePermissionProfile: null,
|
|
thread: {
|
|
id: threadId,
|
|
sessionId: "session-1",
|
|
preview: "",
|
|
ephemeral: false,
|
|
modelProvider: "openai",
|
|
createdAt: 1,
|
|
updatedAt: 1,
|
|
status: { type: "idle" },
|
|
path: null,
|
|
cwd: tempDir,
|
|
cliVersion: "0.147.0",
|
|
source: "unknown",
|
|
agentNickname: null,
|
|
agentRole: null,
|
|
gitInfo: null,
|
|
name: null,
|
|
turns: [],
|
|
},
|
|
};
|
|
}
|
|
|
|
function mockCallArg(mock: ReturnType<typeof vi.fn>, callIndex = 0, argIndex = 0): unknown {
|
|
const call = mock.mock.calls[callIndex];
|
|
if (!call) {
|
|
throw new Error(`Expected mock call ${callIndex}`);
|
|
}
|
|
return call[argIndex];
|
|
}
|
|
|
|
describe("codex conversation binding", () => {
|
|
beforeEach(async () => {
|
|
resetCodexTestBindingStore();
|
|
tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-codex-binding-"));
|
|
});
|
|
|
|
afterEach(async () => {
|
|
publicBindingMocks.resolveByConversation.mockReset();
|
|
publicBindingMocks.resolveByConversation.mockReturnValue({ bindingId: "binding-1" });
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockReset();
|
|
sharedClientMocks.retainSharedCodexAppServerClientByInstanceId.mockReset();
|
|
sharedClientMocks.retainSharedCodexAppServerClientByInstanceId.mockReturnValue(undefined);
|
|
sharedClientMocks.clearSharedCodexAppServerClientIfCurrentAndUnclaimed.mockReset();
|
|
sharedClientMocks.clearSharedCodexAppServerClientIfCurrentAndUnclaimed.mockReturnValue({
|
|
found: false,
|
|
closed: false,
|
|
});
|
|
sharedClientMocks.retireSharedCodexAppServerClientIfCurrent.mockReset();
|
|
sharedClientMocks.clearSharedCodexAppServerClientIfCurrent.mockReset();
|
|
sharedClientMocks.clearSharedCodexAppServerClientIfCurrent.mockReturnValue(false);
|
|
execApprovalsRuntimeMocks.loadExecApprovals.mockReset();
|
|
execApprovalsRuntimeMocks.loadExecApprovals.mockReturnValue({ version: 1, agents: {} });
|
|
agentRuntimeMocks.ensureAuthProfileStore.mockReset();
|
|
agentRuntimeMocks.loadAuthProfileStoreForSecretsRuntime.mockReset();
|
|
agentRuntimeMocks.resolveApiKeyForProfile.mockReset();
|
|
agentRuntimeMocks.resolveAuthProfileOrder.mockReset();
|
|
agentRuntimeMocks.resolveDefaultAgentDir.mockClear();
|
|
agentRuntimeMocks.resolvePersistedAuthProfileOwnerAgentDir.mockReset();
|
|
agentRuntimeMocks.resolveProviderIdForAuth.mockClear();
|
|
agentRuntimeMocks.resolveSessionAgentIds.mockClear();
|
|
agentRuntimeMocks.saveAuthProfileStore.mockReset();
|
|
codexRequirementsTomlMock.mockReset();
|
|
resolveSandboxContextMock.mockReset();
|
|
resolveSandboxContextMock.mockResolvedValue(null);
|
|
await fs.rm(tempDir, { recursive: true, force: true });
|
|
});
|
|
|
|
beforeEach(() => {
|
|
agentRuntimeMocks.ensureAuthProfileStore.mockReturnValue({
|
|
version: 1,
|
|
profiles: {},
|
|
});
|
|
agentRuntimeMocks.resolveAuthProfileOrder.mockReturnValue([]);
|
|
agentRuntimeMocks.resolveDefaultAgentDir.mockReturnValue("/agent");
|
|
agentRuntimeMocks.resolveProviderIdForAuth.mockImplementation(
|
|
(provider: string, _lookup?: { config?: unknown }) => provider,
|
|
);
|
|
agentRuntimeMocks.resolveSessionAgentIds.mockReturnValue({
|
|
defaultAgentId: "main",
|
|
sessionAgentId: "main",
|
|
});
|
|
});
|
|
|
|
it("isolates concurrent turn requests and buffers early bound-turn completion", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, { threadId: "bound-thread", cwd: tempDir });
|
|
const notificationHandlers = new Set<(notification: unknown) => unknown>();
|
|
const requestHandlers = new Set<(request: unknown) => unknown>();
|
|
const siblingRequestOwner = vi.fn(
|
|
(request: { method: string }): JsonValue =>
|
|
request.method === "item/tool/call"
|
|
? { contentItems: [], success: true }
|
|
: { decision: "accept" },
|
|
);
|
|
const siblingResponses: unknown[] = [];
|
|
let releaseSiblingRoute: (() => void) | undefined;
|
|
const client = {
|
|
request: vi.fn(async (method: string) => {
|
|
if (method !== "turn/start") {
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}
|
|
const siblingRoute = getCodexAppServerTurnRouter(clientForRouter).reserveThread({
|
|
threadId: "sibling-thread",
|
|
onRequest: siblingRequestOwner,
|
|
});
|
|
releaseSiblingRoute = siblingRoute.release;
|
|
siblingRoute.armTurn();
|
|
await siblingRoute.bindTurn("sibling-turn");
|
|
for (const requestMethod of ["item/tool/call", "item/commandExecution/requestApproval"]) {
|
|
const request = {
|
|
id: requestMethod,
|
|
method: requestMethod,
|
|
params: { threadId: "sibling-thread", turnId: "sibling-turn" },
|
|
};
|
|
for (const handler of requestHandlers) {
|
|
const response = await handler(request);
|
|
if (response !== undefined) {
|
|
siblingResponses.push(response);
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
for (const handler of notificationHandlers) {
|
|
handler({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "bound-thread",
|
|
turn: {
|
|
id: "bound-turn",
|
|
status: "completed",
|
|
items: [{ type: "agentMessage", id: "bound-answer", text: "Bound answer" }],
|
|
},
|
|
},
|
|
});
|
|
}
|
|
return { turn: { id: "bound-turn" } };
|
|
}),
|
|
addNotificationHandler: vi.fn((handler: (notification: unknown) => unknown) => {
|
|
notificationHandlers.add(handler);
|
|
return () => notificationHandlers.delete(handler);
|
|
}),
|
|
addRequestHandler: vi.fn((handler: (request: unknown) => unknown) => {
|
|
requestHandlers.add(handler);
|
|
return () => requestHandlers.delete(handler);
|
|
}),
|
|
addCloseHandler: vi.fn(() => () => undefined),
|
|
};
|
|
const clientForRouter = client as never;
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue(client);
|
|
const { event, ctx } = boundConversationClaim(sessionFile);
|
|
|
|
try {
|
|
await expect(handleCodexConversationInboundClaim(event, ctx)).resolves.toEqual({
|
|
handled: true,
|
|
reply: { text: "Bound answer" },
|
|
});
|
|
expect(siblingRequestOwner).toHaveBeenCalledTimes(2);
|
|
expect(siblingResponses).toEqual([
|
|
{ contentItems: [], success: true },
|
|
{ decision: "accept" },
|
|
]);
|
|
expect(client.addNotificationHandler).toHaveBeenCalledOnce();
|
|
expect(client.addRequestHandler).toHaveBeenCalledOnce();
|
|
} finally {
|
|
releaseSiblingRoute?.();
|
|
}
|
|
});
|
|
|
|
it("keeps queued bound turns ahead of retirement on their shared owner lane", async () => {
|
|
const sessionFile = path.join(tempDir, "queued-session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, { threadId: "bound-thread", cwd: tempDir });
|
|
const notificationHandlers = new Set<(notification: unknown) => void>();
|
|
const order: string[] = [];
|
|
const client = {
|
|
request: vi.fn(async (method: string) => {
|
|
if (method !== "turn/start") {
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}
|
|
const turnId = `turn-${order.filter((entry) => entry.startsWith("turn-")).length + 1}`;
|
|
order.push(turnId);
|
|
return { turn: { id: turnId } };
|
|
}),
|
|
addNotificationHandler: vi.fn((handler: (notification: unknown) => void) => {
|
|
notificationHandlers.add(handler);
|
|
return () => notificationHandlers.delete(handler);
|
|
}),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
addCloseHandler: vi.fn(() => () => undefined),
|
|
};
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue(client);
|
|
const { event, ctx } = boundConversationClaim(sessionFile);
|
|
const completeTurn = (turnId: string) => {
|
|
for (const handler of notificationHandlers) {
|
|
handler({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "bound-thread",
|
|
turn: {
|
|
id: turnId,
|
|
status: "completed",
|
|
items: [{ type: "agentMessage", id: `${turnId}-answer`, text: turnId }],
|
|
},
|
|
},
|
|
});
|
|
}
|
|
};
|
|
|
|
const firstTurn = handleCodexConversationInboundClaim(event, ctx);
|
|
await vi.waitFor(() => expect(client.request).toHaveBeenCalledOnce());
|
|
const secondTurn = handleCodexConversationInboundClaim(event, ctx);
|
|
const retirement = withCodexConversationThreadActivity(
|
|
legacyCodexConversationBindingId(sessionFile),
|
|
async () => {
|
|
order.push("retired");
|
|
},
|
|
);
|
|
|
|
completeTurn("turn-1");
|
|
await vi.waitFor(() => expect(client.request).toHaveBeenCalledTimes(2));
|
|
expect(order).toEqual(["turn-1", "turn-2"]);
|
|
completeTurn("turn-2");
|
|
|
|
await expect(firstTurn).resolves.toMatchObject({ reply: { text: "turn-1" } });
|
|
await expect(secondTurn).resolves.toMatchObject({ reply: { text: "turn-2" } });
|
|
await retirement;
|
|
expect(order).toEqual(["turn-1", "turn-2", "retired"]);
|
|
});
|
|
|
|
it.each(["detached", "replaced", "cleared-before-capture"] as const)(
|
|
"does not recreate a %s conversation from an inbound claim queued behind retirement",
|
|
async (outcome) => {
|
|
const bindingId = "binding-retiring";
|
|
const identity = { kind: "conversation" as const, bindingId };
|
|
if (outcome === "detached" || outcome === "cleared-before-capture") {
|
|
await testCodexAppServerBindingStore.mutate(identity, {
|
|
kind: "set",
|
|
binding: {
|
|
threadId: "thread-retiring",
|
|
cwd: tempDir,
|
|
conversationStartId: "start-original",
|
|
},
|
|
});
|
|
}
|
|
let retireConversation: (() => void) | undefined;
|
|
const retirementReady = new Promise<void>((resolve) => {
|
|
retireConversation = resolve;
|
|
});
|
|
let retirementStarted: (() => void) | undefined;
|
|
const enteredRetirement = new Promise<void>((resolve) => {
|
|
retirementStarted = resolve;
|
|
});
|
|
const retirement = withCodexConversationThreadActivity(bindingId, async () => {
|
|
if (outcome === "cleared-before-capture") {
|
|
await testCodexAppServerBindingStore.mutate(identity, {
|
|
kind: "clear",
|
|
threadId: "thread-retiring",
|
|
});
|
|
}
|
|
retirementStarted?.();
|
|
await retirementReady;
|
|
if (outcome === "detached") {
|
|
await testCodexAppServerBindingStore.mutate(identity, {
|
|
kind: "clear",
|
|
threadId: "thread-retiring",
|
|
});
|
|
} else if (outcome === "replaced") {
|
|
await testCodexAppServerBindingStore.mutate(identity, {
|
|
kind: "set",
|
|
binding: {
|
|
threadId: "thread-replacement",
|
|
cwd: tempDir,
|
|
conversationStartId: "start-replacement",
|
|
},
|
|
});
|
|
}
|
|
publicBindingMocks.resolveByConversation.mockReturnValue(
|
|
outcome === "replaced" ? { bindingId: "binding-replacement" } : null,
|
|
);
|
|
});
|
|
await enteredRetirement;
|
|
let capturedOwner: (() => void) | undefined;
|
|
const ownerCaptured = new Promise<void>((resolve) => {
|
|
capturedOwner = resolve;
|
|
});
|
|
const bindingStore = {
|
|
...testCodexAppServerBindingStore,
|
|
read: async (
|
|
requestedIdentity: Parameters<typeof testCodexAppServerBindingStore.read>[0],
|
|
) => {
|
|
const binding = await testCodexAppServerBindingStore.read(requestedIdentity);
|
|
capturedOwner?.();
|
|
return binding;
|
|
},
|
|
};
|
|
const { event, ctx } = boundConversationClaim(path.join(tempDir, "retiring-session.jsonl"));
|
|
ctx.pluginBinding.data = {
|
|
kind: "codex-app-server-session" as const,
|
|
version: 2 as const,
|
|
bindingId,
|
|
workspaceDir: tempDir,
|
|
start: { id: "start-original" },
|
|
};
|
|
const queued = handleCodexConversationInboundClaimImpl(
|
|
{ senderIsOwner: true, ...event },
|
|
ctx,
|
|
{
|
|
bindingStore,
|
|
},
|
|
);
|
|
await ownerCaptured;
|
|
|
|
retireConversation?.();
|
|
|
|
await retirement;
|
|
await expect(queued).resolves.toEqual({
|
|
handled: true,
|
|
reply: {
|
|
text: "This Codex conversation was detached or changed before its message could run.",
|
|
},
|
|
});
|
|
expect(sharedClientMocks.getSharedCodexAppServerClient).not.toHaveBeenCalled();
|
|
if (outcome !== "replaced") {
|
|
await expect(testCodexAppServerBindingStore.read(identity)).resolves.toBeUndefined();
|
|
} else {
|
|
await expect(testCodexAppServerBindingStore.read(identity)).resolves.toMatchObject({
|
|
threadId: "thread-replacement",
|
|
});
|
|
}
|
|
},
|
|
);
|
|
|
|
it("uses the default Codex auth profile and omits the public OpenAI provider for new binds", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
const sessionKey = "agent:main:dashboard:incognito-native-bind";
|
|
registerCodexTestSessionIdentity(sessionFile, sessionFile, sessionKey);
|
|
const config = {
|
|
auth: { order: { openai: ["openai:default"] } },
|
|
};
|
|
const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
|
|
agentRuntimeMocks.ensureAuthProfileStore.mockReturnValue({
|
|
version: 1,
|
|
profiles: {
|
|
"openai:default": {
|
|
type: "oauth",
|
|
provider: "openai",
|
|
access: "access-token",
|
|
},
|
|
},
|
|
});
|
|
agentRuntimeMocks.resolveAuthProfileOrder.mockReturnValue(["openai:default"]);
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
requests.push({ method, params: requestParams });
|
|
return {
|
|
thread: { id: "thread-new", sessionId: "session-1", cwd: tempDir },
|
|
model: "gpt-5.4-mini",
|
|
};
|
|
}),
|
|
});
|
|
|
|
await startCodexConversationThread({
|
|
config: config as never,
|
|
sessionFile,
|
|
sessionKey,
|
|
workspaceDir: tempDir,
|
|
model: "gpt-5.4-mini",
|
|
modelProvider: "openai",
|
|
});
|
|
|
|
const authOrderParams = mockCallArg(agentRuntimeMocks.resolveAuthProfileOrder) as {
|
|
cfg?: unknown;
|
|
provider?: unknown;
|
|
};
|
|
expect(authOrderParams?.cfg).toBe(config);
|
|
expect(authOrderParams?.provider).toBe("openai");
|
|
const sharedClientParams = mockCallArg(sharedClientMocks.getSharedCodexAppServerClient) as {
|
|
authProfileId?: unknown;
|
|
};
|
|
expect(sharedClientParams?.authProfileId).toBe("openai:default");
|
|
expect(requests).toHaveLength(1);
|
|
expect(requests[0]?.method).toBe("thread/start");
|
|
expect(requests[0]?.params.model).toBe("gpt-5.4-mini");
|
|
expect(requests[0]?.params.personality).toBe("none");
|
|
expect(requests[0]?.params.ephemeral).toBe(true);
|
|
expect(requests[0]?.params).not.toHaveProperty("modelProvider");
|
|
await expect(readCodexAppServerBinding(sessionFile)).resolves.toMatchObject({
|
|
authProfileId: "openai:default",
|
|
});
|
|
});
|
|
|
|
it.each([
|
|
{
|
|
label: "an incognito source bound to an ordinary destination",
|
|
sourceSessionKey: "agent:main:dashboard:incognito-source",
|
|
destinationSessionKey: "agent:main:telegram:ordinary-destination",
|
|
ephemeral: true,
|
|
turnFails: false,
|
|
},
|
|
{
|
|
label: "an ordinary source bound to an incognito destination",
|
|
sourceSessionKey: "agent:main:telegram:ordinary-source",
|
|
destinationSessionKey: "agent:main:dashboard:incognito-destination",
|
|
ephemeral: false,
|
|
turnFails: false,
|
|
},
|
|
{
|
|
label: "a source without a session key bound to an incognito destination",
|
|
sourceSessionKey: undefined,
|
|
destinationSessionKey: "agent:main:dashboard:incognito-destination",
|
|
ephemeral: false,
|
|
turnFails: false,
|
|
},
|
|
{
|
|
label: "a failing incognito source bound to an ordinary destination",
|
|
sourceSessionKey: "agent:main:dashboard:incognito-source",
|
|
destinationSessionKey: "agent:main:telegram:ordinary-destination",
|
|
ephemeral: true,
|
|
turnFails: true,
|
|
},
|
|
{
|
|
label: "a failing ordinary source bound to an incognito destination",
|
|
sourceSessionKey: "agent:main:telegram:ordinary-source",
|
|
destinationSessionKey: "agent:main:dashboard:incognito-destination",
|
|
ephemeral: false,
|
|
turnFails: true,
|
|
},
|
|
])(
|
|
"uses the persisted source lifecycle for $label",
|
|
async ({ sourceSessionKey, destinationSessionKey, ephemeral, turnFails }) => {
|
|
const sessionFile = path.join(tempDir, "mixed-source-lifecycle.jsonl");
|
|
const bindingId = "binding-mixed-source-lifecycle";
|
|
const operations: Array<{ method: string; params: Record<string, unknown> }> = [];
|
|
const notificationHandlers = new Set<(notification: unknown) => void>();
|
|
const client = {
|
|
getInstanceId: () => "client-mixed-source-lifecycle",
|
|
request: vi.fn(async (method: string, params: Record<string, unknown>) => {
|
|
operations.push({ method, params });
|
|
if (method === "thread/start") {
|
|
return conversationThreadStartResult("thread-mixed-source-lifecycle");
|
|
}
|
|
if (method === "turn/start") {
|
|
if (turnFails) {
|
|
throw new Error("mixed source lifecycle turn failed");
|
|
}
|
|
queueMicrotask(() => {
|
|
for (const handler of notificationHandlers) {
|
|
handler({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "thread-mixed-source-lifecycle",
|
|
turn: {
|
|
id: "turn-mixed-source-lifecycle",
|
|
status: "completed",
|
|
items: [{ type: "agentMessage", id: "answer", text: "Bound reply" }],
|
|
},
|
|
},
|
|
});
|
|
}
|
|
});
|
|
return { turn: { id: "turn-mixed-source-lifecycle" } };
|
|
}
|
|
if (method === "thread/unsubscribe") {
|
|
return {};
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}),
|
|
addNotificationHandler: vi.fn((handler: (notification: unknown) => void) => {
|
|
notificationHandlers.add(handler);
|
|
return () => notificationHandlers.delete(handler);
|
|
}),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
addCloseHandler: vi.fn(() => () => undefined),
|
|
} as unknown as CodexAppServerClient;
|
|
ensureCodexAppServerClientRuntime(client, { agentDir: tempDir });
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue(client);
|
|
sharedClientMocks.retainSharedCodexAppServerClientByInstanceId.mockReturnValue({
|
|
client,
|
|
release: vi.fn(),
|
|
});
|
|
const { event, ctx } = boundConversationClaim(sessionFile, destinationSessionKey);
|
|
const data = {
|
|
kind: "codex-app-server-session" as const,
|
|
version: 2 as const,
|
|
bindingId,
|
|
workspaceDir: tempDir,
|
|
source: {
|
|
agentId: "main",
|
|
sessionId: "source-mixed-lifecycle",
|
|
threadId: "thread-source-mixed-lifecycle",
|
|
...(sourceSessionKey ? { sessionKey: sourceSessionKey } : {}),
|
|
},
|
|
start: { id: "start-mixed-source-lifecycle" },
|
|
};
|
|
ctx.pluginBinding.data = data;
|
|
|
|
await expect(handleCodexConversationInboundClaim(event, ctx)).resolves.toMatchObject({
|
|
handled: true,
|
|
reply: {
|
|
text: turnFails
|
|
? "Codex app-server turn failed: mixed source lifecycle turn failed"
|
|
: "Bound reply",
|
|
},
|
|
});
|
|
|
|
expect(operations.map(({ method }) => method)).toEqual(
|
|
turnFails
|
|
? ["thread/start", "turn/start", "thread/unsubscribe"]
|
|
: ["thread/start", "turn/start"],
|
|
);
|
|
if (ephemeral) {
|
|
expect(operations[0]?.params.ephemeral).toBe(true);
|
|
await expect(
|
|
consumeCodexAppServerLiveThread(client, "thread-mixed-source-lifecycle"),
|
|
).resolves.toBeUndefined();
|
|
} else {
|
|
expect(operations[0]?.params).not.toHaveProperty("ephemeral");
|
|
if (turnFails) {
|
|
await expect(
|
|
consumeCodexAppServerLiveThread(client, "thread-mixed-source-lifecycle"),
|
|
).resolves.toBeUndefined();
|
|
} else {
|
|
const ownership = await consumeCodexAppServerLiveThread(
|
|
client,
|
|
"thread-mixed-source-lifecycle",
|
|
);
|
|
expect(ownership).toEqual(expect.objectContaining({ release: expect.any(Function) }));
|
|
await expect(
|
|
retainCodexAppServerLiveThread(
|
|
client,
|
|
"thread-mixed-source-lifecycle",
|
|
ownership?.release,
|
|
ownership?.configFingerprint,
|
|
ownership?.serviceTier,
|
|
),
|
|
).resolves.toBe(true);
|
|
}
|
|
}
|
|
|
|
if (turnFails && ephemeral) {
|
|
await expect(
|
|
testCodexAppServerBindingStore.read({ kind: "conversation", bindingId }),
|
|
).resolves.toBeUndefined();
|
|
return;
|
|
}
|
|
|
|
await handleCodexConversationBindingResolved({
|
|
status: "denied",
|
|
decision: "deny",
|
|
request: {
|
|
data,
|
|
conversation: {
|
|
channel: "telegram",
|
|
accountId: "default",
|
|
conversationId: "5185575566",
|
|
},
|
|
},
|
|
});
|
|
|
|
expect(operations.at(-1)).toEqual({
|
|
method: "thread/unsubscribe",
|
|
params: { threadId: "thread-mixed-source-lifecycle" },
|
|
});
|
|
await expect(
|
|
testCodexAppServerBindingStore.read({ kind: "conversation", bindingId }),
|
|
).resolves.toBeUndefined();
|
|
},
|
|
);
|
|
|
|
it("selects Codex network-proxy permissions through app-server bind thread config", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
requests.push({ method, params: requestParams });
|
|
return {
|
|
thread: { id: "thread-new", sessionId: "session-1", cwd: tempDir },
|
|
model: "gpt-5.4-mini",
|
|
};
|
|
}),
|
|
});
|
|
|
|
await startCodexConversationThread({
|
|
pluginConfig: NETWORK_PROXY_PLUGIN_CONFIG,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
model: "gpt-5.4-mini",
|
|
modelProvider: "openai",
|
|
});
|
|
|
|
expect(requests).toHaveLength(1);
|
|
expect(requests[0]?.method).toBe("thread/start");
|
|
expect(requests[0]?.params).not.toHaveProperty("permissions");
|
|
expect(requests[0]?.params).not.toHaveProperty("sandbox");
|
|
expect(requests[0]?.params.config).toMatchObject(NETWORK_PROXY_CONFIG_PATCH);
|
|
});
|
|
|
|
it("starts a fresh proxy-backed thread when binding an explicit app-server thread id", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
requests.push({ method, params: requestParams });
|
|
if (method === "thread/resume") {
|
|
throw new Error("thread/resume should not receive network proxy config");
|
|
}
|
|
return conversationThreadStartResult("thread-new");
|
|
}),
|
|
});
|
|
|
|
await startCodexConversationThread({
|
|
pluginConfig: NETWORK_PROXY_PLUGIN_CONFIG,
|
|
sessionFile,
|
|
threadId: "thread-old",
|
|
workspaceDir: tempDir,
|
|
model: "gpt-5.4-mini",
|
|
modelProvider: "openai",
|
|
});
|
|
|
|
expect(requests.map((request) => request.method)).toEqual(["thread/start"]);
|
|
expect(requests[0]?.params).not.toHaveProperty("threadId");
|
|
expect(requests[0]?.params).not.toHaveProperty("sandbox");
|
|
expect(requests[0]?.params.config).toMatchObject(NETWORK_PROXY_CONFIG_PATCH);
|
|
const bindingAfterStart = await readCodexAppServerBinding(sessionFile);
|
|
expect(bindingAfterStart?.threadId).toBe("thread-new");
|
|
expect(bindingAfterStart?.networkProxyProfileName).toBe(NETWORK_PROXY_PROFILE_NAME);
|
|
expect(bindingAfterStart?.networkProxyConfigFingerprint).toBe(NETWORK_PROXY_CONFIG_FINGERPRINT);
|
|
});
|
|
|
|
it("drops a retained native child before applying bound-only apps and sandbox policy", async () => {
|
|
const sessionFile = path.join(tempDir, "retained-child-session.jsonl");
|
|
const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
|
|
const client = {
|
|
getInstanceId: () => "client-native-child",
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
requests.push({ method, params: requestParams });
|
|
if (method === "thread/unsubscribe") {
|
|
return {};
|
|
}
|
|
if (method === "thread/resume") {
|
|
return conversationThreadStartResult("thread-native-child");
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}),
|
|
addNotificationHandler: vi.fn(() => () => undefined),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
addCloseHandler: vi.fn(() => () => undefined),
|
|
} as unknown as CodexAppServerClient;
|
|
ensureCodexAppServerClientRuntime(client, { agentDir: tempDir });
|
|
await retainCodexAppServerLiveThread(client, "thread-native-child");
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue(client);
|
|
|
|
await startCodexConversationThread({
|
|
sessionFile,
|
|
threadId: "thread-native-child",
|
|
workspaceDir: tempDir,
|
|
sandbox: "read-only",
|
|
});
|
|
|
|
expect(requests.map(({ method }) => method)).toEqual(["thread/unsubscribe", "thread/resume"]);
|
|
expect(requests[0]?.params).toEqual({ threadId: "thread-native-child" });
|
|
expect(requests[1]?.params).toMatchObject({
|
|
threadId: "thread-native-child",
|
|
sandbox: "read-only",
|
|
config: { apps: { _default: { enabled: false } }, "features.apps": false },
|
|
});
|
|
await expect(consumeCodexAppServerLiveThread(client, "thread-native-child")).resolves.toEqual(
|
|
expect.objectContaining({ release: expect.any(Function) }),
|
|
);
|
|
});
|
|
|
|
it("never resumes or unsubscribes an actively claimed native child for a conversation", async () => {
|
|
const sessionFile = path.join(tempDir, "active-child-session.jsonl");
|
|
const harness = createClientHarness();
|
|
const request = vi
|
|
.spyOn(harness.client, "request")
|
|
.mockResolvedValue(conversationThreadStartResult("thread-active-child") as never);
|
|
ensureCodexAppServerClientRuntime(harness.client, { agentDir: tempDir });
|
|
const parent = codexNativeSubagentMonitorRuntime.register({
|
|
client: harness.client,
|
|
parentThreadId: "thread-parent",
|
|
});
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue(harness.client);
|
|
|
|
try {
|
|
harness.send({
|
|
method: "thread/started",
|
|
params: {
|
|
thread: {
|
|
id: "thread-active-child",
|
|
parentThreadId: "thread-parent",
|
|
source: {
|
|
subAgent: {
|
|
thread_spawn: {
|
|
parent_thread_id: "thread-parent",
|
|
depth: 1,
|
|
agent_path: "thread-active-child",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
});
|
|
await vi.waitFor(() =>
|
|
expect(isCodexAppServerLiveThreadClaimed(harness.client, "thread-active-child")).toBe(true),
|
|
);
|
|
|
|
await expect(
|
|
startCodexConversationThread({
|
|
sessionFile,
|
|
threadId: "thread-active-child",
|
|
workspaceDir: tempDir,
|
|
}),
|
|
).rejects.toThrow("active run");
|
|
|
|
expect(request).not.toHaveBeenCalled();
|
|
expect(isCodexAppServerLiveThreadClaimed(harness.client, "thread-active-child")).toBe(true);
|
|
} finally {
|
|
parent.unregister();
|
|
harness.client.close();
|
|
}
|
|
});
|
|
|
|
it("keeps a retained native child owned when its pre-resume unsubscribe fails", async () => {
|
|
const sessionFile = path.join(tempDir, "failed-child-session.jsonl");
|
|
const request = vi.fn(async (method: string) => {
|
|
if (method === "thread/unsubscribe") {
|
|
throw new Error("native child unsubscribe failed");
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
});
|
|
const client = {
|
|
getInstanceId: () => "client-failed-child",
|
|
request,
|
|
addNotificationHandler: vi.fn(() => () => undefined),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
addCloseHandler: vi.fn(() => () => undefined),
|
|
} as unknown as CodexAppServerClient;
|
|
ensureCodexAppServerClientRuntime(client, { agentDir: tempDir });
|
|
await retainCodexAppServerLiveThread(client, "thread-failed-child");
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue(client);
|
|
|
|
await expect(
|
|
startCodexConversationThread({
|
|
sessionFile,
|
|
threadId: "thread-failed-child",
|
|
workspaceDir: tempDir,
|
|
}),
|
|
).rejects.toThrow("native child unsubscribe failed");
|
|
|
|
expect(request.mock.calls.map(([method]) => method)).toEqual(["thread/unsubscribe"]);
|
|
await expect(consumeCodexAppServerLiveThread(client, "thread-failed-child")).resolves.toEqual(
|
|
expect.objectContaining({ release: expect.any(Function) }),
|
|
);
|
|
});
|
|
|
|
it.each([
|
|
{ path: "binding" as const, rejectOldRelease: false },
|
|
{ path: "binding" as const, rejectOldRelease: true },
|
|
{ path: "bound turn" as const, rejectOldRelease: false },
|
|
{ path: "bound turn" as const, rejectOldRelease: true },
|
|
])(
|
|
"transfers exact physical ownership during $path migration (old release fails: $rejectOldRelease)",
|
|
async ({ path: migrationPath, rejectOldRelease }) => {
|
|
const sessionFile = path.join(tempDir, "same-thread-client-migration.jsonl");
|
|
const { previousClient, replacementClient, operations, ownerDuringRelease, readOwner } =
|
|
await createSameThreadClientMigrationFixture(sessionFile, {
|
|
rejectOldRelease,
|
|
owner: migrationPath === "binding" ? "session" : "conversation",
|
|
});
|
|
|
|
if (migrationPath === "binding") {
|
|
const binding = startCodexConversationThread({
|
|
sessionFile,
|
|
threadId: "thread-migrated",
|
|
workspaceDir: tempDir,
|
|
});
|
|
if (rejectOldRelease) {
|
|
await expect(binding).rejects.toThrow("previous physical client unsubscribe failed");
|
|
} else {
|
|
await expect(binding).resolves.toMatchObject({
|
|
kind: "codex-app-server-session",
|
|
source: { threadId: "thread-migrated" },
|
|
});
|
|
}
|
|
} else {
|
|
const { event, ctx } = boundConversationClaim(sessionFile);
|
|
const result = await handleCodexConversationInboundClaim(event, ctx, { timeoutMs: 500 });
|
|
expect(result?.reply?.text).toContain(
|
|
rejectOldRelease ? "previous physical client unsubscribe failed" : "Migrated reply",
|
|
);
|
|
}
|
|
|
|
const expectedOperations = ["replacement:thread/resume", "previous:thread/unsubscribe"];
|
|
if (rejectOldRelease) {
|
|
expectedOperations.push("replacement:thread/unsubscribe");
|
|
} else if (migrationPath === "bound turn") {
|
|
expectedOperations.push("replacement:turn/start");
|
|
}
|
|
expect(operations).toEqual(expectedOperations);
|
|
expect(ownerDuringRelease).toEqual(["client-before-migration"]);
|
|
await expect(readOwner()).resolves.toMatchObject({
|
|
threadId: "thread-migrated",
|
|
clientId: rejectOldRelease ? "client-before-migration" : "client-after-migration",
|
|
});
|
|
const survivingClient = rejectOldRelease ? previousClient : replacementClient;
|
|
const obsoleteClient = rejectOldRelease ? replacementClient : previousClient;
|
|
await expect(
|
|
consumeCodexAppServerLiveThread(survivingClient, "thread-migrated"),
|
|
).resolves.toEqual(expect.objectContaining({ release: expect.any(Function) }));
|
|
await expect(
|
|
consumeCodexAppServerLiveThread(obsoleteClient, "thread-migrated"),
|
|
).resolves.toBeUndefined();
|
|
},
|
|
);
|
|
|
|
it("starts a new bind thread when no model override is provided", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
requests.push({ method, params: requestParams });
|
|
return {
|
|
thread: { id: "thread-new", sessionId: "session-1", cwd: tempDir },
|
|
model: "gpt-5.5",
|
|
};
|
|
}),
|
|
});
|
|
|
|
await startCodexConversationThread({
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
});
|
|
|
|
expect(requests).toHaveLength(1);
|
|
expect(requests[0]?.method).toBe("thread/start");
|
|
expect(requests[0]?.params).not.toHaveProperty("model");
|
|
expect(requests[0]?.params).not.toHaveProperty("modelProvider");
|
|
await expect(readCodexAppServerBinding(sessionFile)).resolves.toMatchObject({
|
|
model: "gpt-5.5",
|
|
});
|
|
});
|
|
|
|
it("preserves Codex auth and omits the public OpenAI provider for native bind threads", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
agentRuntimeMocks.ensureAuthProfileStore.mockReturnValue({
|
|
version: 1,
|
|
profiles: {
|
|
work: {
|
|
type: "oauth",
|
|
provider: "openai",
|
|
access: "access-token",
|
|
refresh: "refresh-token",
|
|
expires: Date.now() + 60_000,
|
|
},
|
|
},
|
|
});
|
|
await writeCodexAppServerBinding(sessionFile, {
|
|
threadId: "thread-old",
|
|
cwd: tempDir,
|
|
authProfileId: "work",
|
|
modelProvider: "openai",
|
|
});
|
|
const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
requests.push({ method, params: requestParams });
|
|
return {
|
|
thread: { id: "thread-new", sessionId: "session-1", cwd: tempDir },
|
|
model: "gpt-5.4-mini",
|
|
modelProvider: "openai",
|
|
};
|
|
}),
|
|
});
|
|
|
|
await startCodexConversationThread({
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
model: "gpt-5.4-mini",
|
|
modelProvider: "openai",
|
|
});
|
|
|
|
const sharedClientParams = mockCallArg(sharedClientMocks.getSharedCodexAppServerClient) as {
|
|
authProfileId?: unknown;
|
|
};
|
|
expect(sharedClientParams?.authProfileId).toBe("work");
|
|
expect(requests).toHaveLength(1);
|
|
expect(requests[0]?.method).toBe("thread/start");
|
|
expect(requests[0]?.params.model).toBe("gpt-5.4-mini");
|
|
expect(requests[0]?.params.personality).toBe("none");
|
|
expect(requests[0]?.params).not.toHaveProperty("modelProvider");
|
|
const savedBinding = await readCodexAppServerBinding(sessionFile);
|
|
expect(savedBinding?.authProfileId).toBe("work");
|
|
expect(savedBinding?.modelProvider).toBeUndefined();
|
|
});
|
|
|
|
it("stores and uses the owning agent dir for bound app-server sessions", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
const agentDir = path.join(tempDir, "agents", "bot-a", "agent");
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async () => ({
|
|
thread: { id: "thread-new", sessionId: "session-1", cwd: tempDir },
|
|
model: "gpt-5.4-mini",
|
|
})),
|
|
});
|
|
|
|
const data = await startCodexConversationThread({
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
agentDir,
|
|
model: "gpt-5.4-mini",
|
|
});
|
|
|
|
const sharedClientParams = mockCallArg(sharedClientMocks.getSharedCodexAppServerClient) as {
|
|
agentDir?: unknown;
|
|
};
|
|
expect(sharedClientParams?.agentDir).toBe(agentDir);
|
|
expect(data.agentDir).toBe(agentDir);
|
|
});
|
|
|
|
it("rejects direct conversation start over a private supervised binding", async () => {
|
|
const sessionFile = path.join(tempDir, "supervised-session.jsonl");
|
|
await writeCodexAppServerBinding(sessionFile, {
|
|
threadId: "thread-supervised",
|
|
connectionScope: "supervision",
|
|
supervisionSourceThreadId: "thread-source",
|
|
cwd: tempDir,
|
|
model: "gpt-5.5",
|
|
modelProvider: "openai",
|
|
preserveNativeModel: true,
|
|
conversationSourceTransferComplete: true,
|
|
});
|
|
|
|
await expect(
|
|
startCodexConversationThread({
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
model: "gpt-5.4",
|
|
}),
|
|
).rejects.toThrow("Refusing to replace supervised Codex thread");
|
|
expect(sharedClientMocks.getSharedCodexAppServerClient).not.toHaveBeenCalled();
|
|
await expect(readCodexAppServerBinding(sessionFile)).resolves.toMatchObject({
|
|
threadId: "thread-supervised",
|
|
connectionScope: "supervision",
|
|
});
|
|
});
|
|
|
|
it("rejects binding when configured exec auto mode may need unrouted human approvals", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
requests.push({ method, params: requestParams });
|
|
return {
|
|
thread: { id: "thread-new", sessionId: "session-1", cwd: tempDir },
|
|
model: "gpt-5.4-mini",
|
|
};
|
|
}),
|
|
});
|
|
|
|
await expect(
|
|
startCodexConversationThread({
|
|
config: {
|
|
tools: {
|
|
exec: {
|
|
mode: "auto",
|
|
},
|
|
},
|
|
} as never,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
model: "gpt-5.4-mini",
|
|
}),
|
|
).rejects.toThrow(
|
|
"OpenClaw native Codex conversation binding cannot route interactive approvals yet",
|
|
);
|
|
expect(requests).toEqual([]);
|
|
});
|
|
|
|
it("rejects binding when the binding agent exec auto mode may need unrouted approvals", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
const request = vi.fn();
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request,
|
|
});
|
|
|
|
await expect(
|
|
startCodexConversationThread({
|
|
config: {
|
|
tools: {
|
|
exec: {
|
|
mode: "full",
|
|
},
|
|
},
|
|
agents: {
|
|
list: [
|
|
{
|
|
id: "bot-a",
|
|
tools: {
|
|
exec: {
|
|
mode: "auto",
|
|
},
|
|
},
|
|
},
|
|
],
|
|
},
|
|
} as never,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
agentId: "bot-a",
|
|
model: "gpt-5.4-mini",
|
|
}),
|
|
).rejects.toThrow(
|
|
"OpenClaw native Codex conversation binding cannot route interactive approvals yet",
|
|
);
|
|
expect(request).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("rejects binding when configured exec ask mode needs unrouted user approvals", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
requests.push({ method, params: requestParams });
|
|
return {
|
|
thread: { id: "thread-new", sessionId: "session-1", cwd: tempDir },
|
|
model: "gpt-5.4-mini",
|
|
};
|
|
}),
|
|
});
|
|
|
|
await expect(
|
|
startCodexConversationThread({
|
|
config: {
|
|
tools: {
|
|
exec: {
|
|
mode: "ask",
|
|
},
|
|
},
|
|
} as never,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
model: "gpt-5.4-mini",
|
|
}),
|
|
).rejects.toThrow(
|
|
"OpenClaw native Codex conversation binding cannot route interactive approvals yet",
|
|
);
|
|
expect(requests).toEqual([]);
|
|
});
|
|
|
|
it("applies host exec approval floors to configless native bind threads", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
|
|
execApprovalsRuntimeMocks.loadExecApprovals.mockReturnValue({
|
|
version: 1,
|
|
defaults: {
|
|
security: "deny",
|
|
ask: "off",
|
|
},
|
|
agents: {},
|
|
});
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
requests.push({ method, params: requestParams });
|
|
return {
|
|
thread: { id: "thread-new", sessionId: "session-1", cwd: tempDir },
|
|
model: "gpt-5.4-mini",
|
|
};
|
|
}),
|
|
});
|
|
|
|
await expect(
|
|
startCodexConversationThread({
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
model: "gpt-5.4-mini",
|
|
}),
|
|
).rejects.toThrow("tools.exec.mode=deny");
|
|
expect(execApprovalsRuntimeMocks.loadExecApprovals).toHaveBeenCalled();
|
|
expect(requests).toEqual([]);
|
|
});
|
|
|
|
it("clears the Codex app-server binding when a pending bind is denied", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, { threadId: "thread-1", cwd: tempDir });
|
|
|
|
await handleCodexConversationBindingResolved({
|
|
status: "denied",
|
|
decision: "deny",
|
|
request: {
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
},
|
|
conversation: {
|
|
channel: "discord",
|
|
accountId: "default",
|
|
conversationId: "channel:1",
|
|
},
|
|
},
|
|
});
|
|
|
|
await expect(readTestConversationBinding(sessionFile)).resolves.toBeUndefined();
|
|
});
|
|
|
|
it("retires an already-evicted conversation without disturbing its live client siblings", async () => {
|
|
const identity = { kind: "conversation" as const, bindingId: "binding-evicted" };
|
|
const client = {
|
|
request: vi.fn(async () => ({})),
|
|
addCloseHandler: vi.fn(),
|
|
addNotificationHandler: vi.fn(),
|
|
addRequestHandler: vi.fn(),
|
|
} as unknown as CodexAppServerClient;
|
|
ensureCodexAppServerClientRuntime(client, { agentDir: tempDir });
|
|
const releaseEvicted = vi.fn(async (_threadId: string) => undefined);
|
|
await retainCodexAppServerLiveThread(client, "thread-evicted", releaseEvicted);
|
|
for (let index = 0; index < 64; index += 1) {
|
|
await retainCodexAppServerLiveThread(
|
|
client,
|
|
`thread-sibling-${index}`,
|
|
async () => undefined,
|
|
);
|
|
}
|
|
expect(releaseEvicted).toHaveBeenCalledExactlyOnceWith("thread-evicted");
|
|
sharedClientMocks.retainSharedCodexAppServerClientByInstanceId.mockReturnValue({
|
|
client,
|
|
release: vi.fn(),
|
|
});
|
|
await testCodexAppServerBindingStore.mutate(identity, {
|
|
kind: "set",
|
|
binding: {
|
|
threadId: "thread-evicted",
|
|
clientId: "client-with-siblings",
|
|
cwd: tempDir,
|
|
conversationStartId: "start-evicted",
|
|
},
|
|
});
|
|
|
|
await handleCodexConversationBindingResolved({
|
|
status: "denied",
|
|
decision: "deny",
|
|
request: {
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 2,
|
|
bindingId: identity.bindingId,
|
|
workspaceDir: tempDir,
|
|
start: { id: "start-evicted" },
|
|
},
|
|
conversation: {
|
|
channel: "discord",
|
|
accountId: "default",
|
|
conversationId: "channel:evicted",
|
|
},
|
|
},
|
|
});
|
|
|
|
await expect(testCodexAppServerBindingStore.read(identity)).resolves.toBeUndefined();
|
|
await expect(consumeCodexAppServerLiveThread(client, "thread-sibling-0")).resolves.toEqual(
|
|
expect.objectContaining({ release: expect.any(Function) }),
|
|
);
|
|
});
|
|
|
|
it("unsubscribes an untracked incognito conversation when its binding is denied", async () => {
|
|
const identity = { kind: "conversation" as const, bindingId: "binding-incognito" };
|
|
const request = vi.fn(async () => ({}));
|
|
const release = vi.fn();
|
|
sharedClientMocks.retainSharedCodexAppServerClientByInstanceId.mockReturnValue({
|
|
client: { request },
|
|
release,
|
|
});
|
|
await testCodexAppServerBindingStore.mutate(identity, {
|
|
kind: "set",
|
|
binding: {
|
|
threadId: "thread-incognito",
|
|
clientId: "client-incognito",
|
|
cwd: tempDir,
|
|
conversationStartId: "start-incognito",
|
|
},
|
|
});
|
|
|
|
await handleCodexConversationBindingResolved({
|
|
status: "denied",
|
|
decision: "deny",
|
|
request: {
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 2,
|
|
bindingId: identity.bindingId,
|
|
workspaceDir: tempDir,
|
|
source: {
|
|
agentId: "main",
|
|
sessionId: "session-incognito",
|
|
threadId: "thread-source",
|
|
sessionKey: "agent:main:dashboard:incognito-native-bind",
|
|
},
|
|
start: { id: "start-incognito" },
|
|
},
|
|
conversation: {
|
|
channel: "discord",
|
|
accountId: "default",
|
|
conversationId: "channel:incognito",
|
|
},
|
|
},
|
|
});
|
|
|
|
expect(request).toHaveBeenCalledWith(
|
|
"thread/unsubscribe",
|
|
{ threadId: "thread-incognito" },
|
|
expect.objectContaining({ timeoutMs: expect.any(Number) }),
|
|
);
|
|
expect(release).toHaveBeenCalledOnce();
|
|
await expect(testCodexAppServerBindingStore.read(identity)).resolves.toBeUndefined();
|
|
});
|
|
|
|
it("keeps a denied conversation binding when native unsubscribe fails", async () => {
|
|
const identity = { kind: "conversation" as const, bindingId: "binding-failed-denial" };
|
|
const request = vi.fn(async () => {
|
|
throw new Error("native unsubscribe failed");
|
|
});
|
|
const close = vi.fn();
|
|
sharedClientMocks.retainSharedCodexAppServerClientByInstanceId.mockReturnValue({
|
|
client: { request, close },
|
|
release: vi.fn(),
|
|
});
|
|
await testCodexAppServerBindingStore.mutate(identity, {
|
|
kind: "set",
|
|
binding: {
|
|
threadId: "thread-failed-denial",
|
|
clientId: "client-failed-denial",
|
|
cwd: tempDir,
|
|
conversationStartId: "start-failed-denial",
|
|
},
|
|
});
|
|
|
|
await expect(
|
|
handleCodexConversationBindingResolved({
|
|
status: "denied",
|
|
decision: "deny",
|
|
request: {
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 2,
|
|
bindingId: identity.bindingId,
|
|
workspaceDir: tempDir,
|
|
source: {
|
|
agentId: "main",
|
|
sessionId: "session-incognito",
|
|
threadId: "thread-source",
|
|
sessionKey: "agent:main:dashboard:incognito-native-bind",
|
|
},
|
|
start: { id: "start-failed-denial" },
|
|
},
|
|
conversation: {
|
|
channel: "discord",
|
|
accountId: "default",
|
|
conversationId: "channel:incognito",
|
|
},
|
|
},
|
|
}),
|
|
).rejects.toThrow("subscription could not be released");
|
|
|
|
expect(request).toHaveBeenCalledWith(
|
|
"thread/unsubscribe",
|
|
{ threadId: "thread-failed-denial" },
|
|
expect.objectContaining({ timeoutMs: expect.any(Number) }),
|
|
);
|
|
await expect(testCodexAppServerBindingStore.read(identity)).resolves.toMatchObject({
|
|
threadId: "thread-failed-denial",
|
|
});
|
|
});
|
|
|
|
it("preserves the live conversation generation when a replacement bind is denied", async () => {
|
|
const identity = { kind: "conversation" as const, bindingId: "binding-data-1" };
|
|
await testCodexAppServerBindingStore.mutate(identity, {
|
|
kind: "set",
|
|
binding: {
|
|
threadId: "thread-old",
|
|
cwd: tempDir,
|
|
conversationStartId: "start-old",
|
|
},
|
|
});
|
|
|
|
await handleCodexConversationBindingResolved({
|
|
status: "denied",
|
|
decision: "deny",
|
|
request: {
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 2,
|
|
bindingId: "binding-data-1",
|
|
workspaceDir: tempDir,
|
|
start: { id: "start-new", threadId: "thread-new" },
|
|
},
|
|
conversation: {
|
|
channel: "discord",
|
|
accountId: "default",
|
|
conversationId: "channel:1",
|
|
},
|
|
},
|
|
});
|
|
|
|
await expect(testCodexAppServerBindingStore.read(identity)).resolves.toMatchObject({
|
|
threadId: "thread-old",
|
|
conversationStartId: "start-old",
|
|
});
|
|
});
|
|
|
|
it("rejects attaching a conversation to another session's owned thread", async () => {
|
|
const otherIdentity = {
|
|
kind: "session" as const,
|
|
agentId: "main",
|
|
sessionId: "session-other",
|
|
};
|
|
await testCodexAppServerBindingStore.mutate(otherIdentity, {
|
|
kind: "set",
|
|
binding: { threadId: "thread-owned", cwd: tempDir },
|
|
});
|
|
const request = vi.fn(async () => conversationThreadStartResult("thread-owned"));
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({ request });
|
|
|
|
await expect(
|
|
startCodexConversationThread({
|
|
sessionFile: path.join(tempDir, "session.jsonl"),
|
|
workspaceDir: tempDir,
|
|
threadId: "thread-owned",
|
|
}),
|
|
).rejects.toThrow("owned by another OpenClaw session");
|
|
expect(request).not.toHaveBeenCalled();
|
|
await expect(testCodexAppServerBindingStore.read(otherIdentity)).resolves.toMatchObject({
|
|
threadId: "thread-owned",
|
|
});
|
|
});
|
|
|
|
it("consumes inbound bound messages when command authorization is absent", async () => {
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "run this",
|
|
channel: "discord",
|
|
isGroup: true,
|
|
senderIsOwner: false,
|
|
},
|
|
{
|
|
channelId: "discord",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "discord",
|
|
accountId: "default",
|
|
conversationId: "channel-1",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile: path.join(tempDir, "session.jsonl"),
|
|
workspaceDir: tempDir,
|
|
},
|
|
},
|
|
},
|
|
);
|
|
|
|
expect(result).toEqual({ handled: true });
|
|
});
|
|
|
|
it("blocks inbound bound turns without current owner or admin authority", async () => {
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "run this",
|
|
channel: "discord",
|
|
isGroup: true,
|
|
commandAuthorized: true,
|
|
senderIsOwner: false,
|
|
},
|
|
{
|
|
channelId: "discord",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "discord",
|
|
accountId: "default",
|
|
conversationId: "channel-1",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile: path.join(tempDir, "session.jsonl"),
|
|
workspaceDir: tempDir,
|
|
},
|
|
},
|
|
},
|
|
);
|
|
|
|
expect(result).toEqual({
|
|
handled: true,
|
|
reply: { text: "Only an owner or operator.admin can control Codex native execution." },
|
|
});
|
|
expect(sharedClientMocks.getSharedCodexAppServerClient).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("routes a programmatically bound Control UI session through node resume", async () => {
|
|
const resumeCodexCliSessionOnNode = vi.fn(async () => ({
|
|
ok: true as const,
|
|
sessionId: "019e2007-1f7e-7eb1-a42b-8c01f4b9b5cd",
|
|
text: "done",
|
|
}));
|
|
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "continue the task",
|
|
channel: "webchat",
|
|
isGroup: false,
|
|
commandAuthorized: true,
|
|
sessionKey: "node-session",
|
|
},
|
|
{
|
|
channelId: "webchat",
|
|
sessionKey: "node-session",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "webchat",
|
|
accountId: "default",
|
|
conversationId: "node-session",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-cli-node-session",
|
|
version: 1,
|
|
nodeId: "mb-m5",
|
|
sessionId: "019e2007-1f7e-7eb1-a42b-8c01f4b9b5cd",
|
|
cwd: "/repo",
|
|
},
|
|
},
|
|
},
|
|
{
|
|
config: { tools: { exec: { host: "node", node: "mb-m5" } } },
|
|
resumeCodexCliSessionOnNode,
|
|
timeoutMs: 1234,
|
|
},
|
|
);
|
|
|
|
expect(result).toEqual({ handled: true, reply: { text: "done" } });
|
|
expect(resumeCodexCliSessionOnNode).toHaveBeenCalledWith({
|
|
nodeId: "mb-m5",
|
|
sessionId: "019e2007-1f7e-7eb1-a42b-8c01f4b9b5cd",
|
|
prompt: "continue the task",
|
|
cwd: "/repo",
|
|
timeoutMs: 1234,
|
|
});
|
|
});
|
|
|
|
it("blocks bound Codex app-server turns when the current OpenClaw session is sandboxed", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, { threadId: "thread-1", cwd: tempDir });
|
|
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "continue the task",
|
|
channel: "discord",
|
|
isGroup: true,
|
|
commandAuthorized: true,
|
|
sessionKey: "sandboxed-session",
|
|
},
|
|
{
|
|
channelId: "discord",
|
|
sessionKey: "sandboxed-session",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "discord",
|
|
accountId: "default",
|
|
conversationId: "channel-1",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
config: { agents: { defaults: { sandbox: { mode: "all" } } } },
|
|
},
|
|
);
|
|
|
|
expect(result).toEqual({
|
|
handled: true,
|
|
reply: {
|
|
text: expect.stringContaining(
|
|
"Codex-native Codex app-server conversation binding is unavailable because OpenClaw sandboxing is active for this session.",
|
|
),
|
|
},
|
|
});
|
|
expect(sharedClientMocks.getSharedCodexAppServerClient).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("blocks bound Codex app-server turns when exec host=node is active", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, { threadId: "thread-1", cwd: tempDir });
|
|
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "continue the task",
|
|
channel: "discord",
|
|
isGroup: true,
|
|
commandAuthorized: true,
|
|
sessionKey: "node-session",
|
|
},
|
|
{
|
|
channelId: "discord",
|
|
sessionKey: "node-session",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "discord",
|
|
accountId: "default",
|
|
conversationId: "channel-1",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
config: { tools: { exec: { host: "node", node: "worker-1" } } },
|
|
},
|
|
);
|
|
|
|
expect(result).toEqual({
|
|
handled: true,
|
|
reply: {
|
|
text: expect.stringContaining(
|
|
"Codex-native Codex app-server conversation binding is unavailable because OpenClaw exec host=node is active for this session.",
|
|
),
|
|
},
|
|
});
|
|
expect(sharedClientMocks.getSharedCodexAppServerClient).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("blocks bound Codex app-server turns when the binding agent uses node exec without a session key", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, { threadId: "thread-1", cwd: tempDir });
|
|
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "continue the task",
|
|
channel: "discord",
|
|
isGroup: true,
|
|
commandAuthorized: true,
|
|
},
|
|
{
|
|
channelId: "discord",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "discord",
|
|
accountId: "default",
|
|
conversationId: "channel-1",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
agentId: "bot-a",
|
|
},
|
|
},
|
|
},
|
|
{
|
|
config: {
|
|
tools: { exec: { host: "gateway" } },
|
|
agents: {
|
|
list: [
|
|
{
|
|
id: "bot-a",
|
|
tools: { exec: { host: "node", node: "worker-1" } },
|
|
},
|
|
],
|
|
},
|
|
} as never,
|
|
},
|
|
);
|
|
|
|
expect(result?.handled).toBe(true);
|
|
expect(result?.reply?.text).toContain("OpenClaw exec host=node is active");
|
|
expect(sharedClientMocks.getSharedCodexAppServerClient).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("keeps the bound agent node exec block ahead of current-session exec host overrides", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
const storePath = path.join(tempDir, "agents", "main", "sessions", "sessions.json");
|
|
await writeTestConversationBinding(sessionFile, { threadId: "thread-1", cwd: tempDir });
|
|
await upsertSessionEntry({
|
|
storePath,
|
|
sessionKey: "agent:main:session-1",
|
|
entry: {
|
|
sessionId: "session-1",
|
|
updatedAt: Date.now(),
|
|
execHost: "gateway",
|
|
},
|
|
});
|
|
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "continue the task",
|
|
channel: "discord",
|
|
isGroup: true,
|
|
commandAuthorized: true,
|
|
sessionKey: "agent:main:session-1",
|
|
},
|
|
{
|
|
channelId: "discord",
|
|
sessionKey: "agent:main:session-1",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "discord",
|
|
accountId: "default",
|
|
conversationId: "channel-1",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
agentId: "bot-a",
|
|
},
|
|
},
|
|
},
|
|
{
|
|
config: {
|
|
session: {
|
|
store: path.join(tempDir, "agents", "{agentId}", "sessions", "sessions.json"),
|
|
},
|
|
tools: { exec: { host: "gateway" } },
|
|
agents: {
|
|
list: [
|
|
{
|
|
id: "bot-a",
|
|
tools: { exec: { host: "node", node: "worker-1" } },
|
|
},
|
|
],
|
|
},
|
|
} as never,
|
|
},
|
|
);
|
|
|
|
expect(result?.handled).toBe(true);
|
|
expect(result?.reply?.text).toContain("OpenClaw exec host=node is active");
|
|
expect(sharedClientMocks.getSharedCodexAppServerClient).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("rejects bound Codex app-server turns when the binding agent exec auto mode needs approvals", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, { threadId: "thread-1", cwd: tempDir });
|
|
const request = vi.fn(async () => {
|
|
throw new Error("unexpected native turn");
|
|
});
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request,
|
|
addNotificationHandler: vi.fn(() => () => undefined),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
});
|
|
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "continue the task",
|
|
channel: "discord",
|
|
isGroup: true,
|
|
commandAuthorized: true,
|
|
},
|
|
{
|
|
channelId: "discord",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "discord",
|
|
accountId: "default",
|
|
conversationId: "channel-1",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
agentId: "bot-a",
|
|
},
|
|
},
|
|
},
|
|
{
|
|
timeoutMs: 50,
|
|
config: {
|
|
tools: {
|
|
exec: {
|
|
mode: "full",
|
|
},
|
|
},
|
|
agents: {
|
|
list: [
|
|
{
|
|
id: "bot-a",
|
|
tools: {
|
|
exec: {
|
|
mode: "auto",
|
|
},
|
|
},
|
|
},
|
|
],
|
|
},
|
|
} as never,
|
|
},
|
|
);
|
|
|
|
expect(result?.handled).toBe(true);
|
|
expect(result?.reply?.text).toContain(
|
|
"OpenClaw native Codex conversation binding cannot route interactive approvals yet",
|
|
);
|
|
expect(request).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("keeps bound agent approval policy ahead of different-agent session overrides", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
const storePath = path.join(tempDir, "sessions.json");
|
|
await writeTestConversationBinding(sessionFile, { threadId: "thread-1", cwd: tempDir });
|
|
await upsertSessionEntry({
|
|
storePath,
|
|
sessionKey: "agent:main:session-1",
|
|
entry: {
|
|
sessionId: "session-1",
|
|
updatedAt: Date.now(),
|
|
execSecurity: "full",
|
|
execAsk: "off",
|
|
},
|
|
});
|
|
const request = vi.fn(async () => {
|
|
throw new Error("unexpected native turn");
|
|
});
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request,
|
|
addNotificationHandler: vi.fn(() => () => undefined),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
});
|
|
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "continue the task",
|
|
channel: "discord",
|
|
isGroup: true,
|
|
commandAuthorized: true,
|
|
sessionKey: "agent:main:session-1",
|
|
},
|
|
{
|
|
channelId: "discord",
|
|
sessionKey: "agent:main:session-1",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "discord",
|
|
accountId: "default",
|
|
conversationId: "channel-1",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
agentId: "bot-a",
|
|
},
|
|
},
|
|
},
|
|
{
|
|
timeoutMs: 50,
|
|
config: {
|
|
session: { store: storePath },
|
|
tools: {
|
|
exec: {
|
|
mode: "full",
|
|
},
|
|
},
|
|
agents: {
|
|
list: [
|
|
{
|
|
id: "bot-a",
|
|
tools: {
|
|
exec: {
|
|
mode: "auto",
|
|
},
|
|
},
|
|
},
|
|
],
|
|
},
|
|
} as never,
|
|
},
|
|
);
|
|
|
|
expect(result?.handled).toBe(true);
|
|
expect(result?.reply?.text).toContain(
|
|
"OpenClaw native Codex conversation binding cannot route interactive approvals yet",
|
|
);
|
|
expect(request).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("blocks bound Codex CLI node turns when the current OpenClaw session is sandboxed", async () => {
|
|
const resumeCodexCliSessionOnNode = vi.fn();
|
|
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "continue the task",
|
|
channel: "discord",
|
|
isGroup: true,
|
|
commandAuthorized: true,
|
|
sessionKey: "sandboxed-session",
|
|
},
|
|
{
|
|
channelId: "discord",
|
|
sessionKey: "sandboxed-session",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "discord",
|
|
accountId: "default",
|
|
conversationId: "channel-1",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-cli-node-session",
|
|
version: 1,
|
|
nodeId: "mb-m5",
|
|
sessionId: "019e2007-1f7e-7eb1-a42b-8c01f4b9b5cd",
|
|
cwd: "/repo",
|
|
},
|
|
},
|
|
},
|
|
{
|
|
config: { agents: { defaults: { sandbox: { mode: "all" } } } },
|
|
resumeCodexCliSessionOnNode,
|
|
},
|
|
);
|
|
|
|
expect(result).toEqual({
|
|
handled: true,
|
|
reply: {
|
|
text: expect.stringContaining(
|
|
"Codex-native Codex CLI node conversation binding is unavailable because OpenClaw sandboxing is active for this session.",
|
|
),
|
|
},
|
|
});
|
|
expect(resumeCodexCliSessionOnNode).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("recreates a missing bound thread and preserves auth plus turn overrides", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
agentRuntimeMocks.ensureAuthProfileStore.mockReturnValue({
|
|
version: 1,
|
|
profiles: {
|
|
work: {
|
|
type: "oauth",
|
|
provider: "openai",
|
|
access: "access-token",
|
|
},
|
|
},
|
|
});
|
|
await writeTestConversationBinding(sessionFile, {
|
|
threadId: "thread-old",
|
|
cwd: tempDir,
|
|
authProfileId: "work",
|
|
model: "gpt-5.4-mini",
|
|
modelProvider: "openai",
|
|
approvalPolicy: "on-request",
|
|
sandbox: "workspace-write",
|
|
serviceTier: "fast",
|
|
});
|
|
const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
|
|
const notificationHandlers: Array<(notification: Record<string, unknown>) => void> = [];
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
requests.push({ method, params: requestParams });
|
|
if (method === "turn/start" && requestParams.threadId === "thread-old") {
|
|
throw new Error("thread not found: thread-old");
|
|
}
|
|
if (method === "thread/start") {
|
|
return {
|
|
thread: { id: "thread-new", sessionId: "session-1", cwd: tempDir },
|
|
model: "gpt-5.4-mini",
|
|
};
|
|
}
|
|
if (method === "turn/start" && requestParams.threadId === "thread-new") {
|
|
setImmediate(() => {
|
|
for (const handler of notificationHandlers) {
|
|
handler({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "thread-new",
|
|
turn: {
|
|
id: "turn-new",
|
|
status: "completed",
|
|
items: [
|
|
{
|
|
id: "assistant-1",
|
|
type: "agentMessage",
|
|
text: "Recovered",
|
|
},
|
|
],
|
|
},
|
|
},
|
|
});
|
|
}
|
|
});
|
|
return { turn: { id: "turn-new" } };
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}),
|
|
addNotificationHandler: vi.fn((handler) => {
|
|
notificationHandlers.push(handler);
|
|
return () => undefined;
|
|
}),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
});
|
|
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "hi again",
|
|
bodyForAgent: "hi again",
|
|
channel: "telegram",
|
|
isGroup: false,
|
|
commandAuthorized: true,
|
|
},
|
|
{
|
|
channelId: "telegram",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "telegram",
|
|
accountId: "default",
|
|
conversationId: "5185575566",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
},
|
|
},
|
|
},
|
|
{ timeoutMs: 500 },
|
|
);
|
|
|
|
expect(result).toEqual({ handled: true, reply: { text: "Recovered" } });
|
|
expect(requests.map((request) => request.method)).toEqual([
|
|
"turn/start",
|
|
"thread/start",
|
|
"turn/start",
|
|
]);
|
|
const sharedClientParams = mockCallArg(sharedClientMocks.getSharedCodexAppServerClient) as {
|
|
authProfileId?: unknown;
|
|
};
|
|
expect(sharedClientParams?.authProfileId).toBe("work");
|
|
expect(requests[1]?.params.model).toBe("gpt-5.4-mini");
|
|
expect(requests[1]?.params.approvalPolicy).toBe("on-request");
|
|
expect(requests[1]?.params.sandbox).toBe("workspace-write");
|
|
expect(requests[1]?.params.serviceTier).toBe("priority");
|
|
expect(requests[1]?.params).not.toHaveProperty("modelProvider");
|
|
expect(requests[2]?.params.threadId).toBe("thread-new");
|
|
expect(requests[2]?.params.approvalPolicy).toBe("on-request");
|
|
expect(requests[2]?.params.serviceTier).toBe("priority");
|
|
const savedBinding = await readTestConversationBinding(sessionFile);
|
|
expect(savedBinding?.threadId).toBe("thread-new");
|
|
expect(savedBinding?.authProfileId).toBe("work");
|
|
expect(savedBinding?.approvalPolicy).toBe("on-request");
|
|
expect(savedBinding?.sandbox).toBe("workspace-write");
|
|
expect(savedBinding?.serviceTier).toBe("priority");
|
|
expect(savedBinding).not.toHaveProperty("modelProvider");
|
|
});
|
|
|
|
it("applies a new lazy bind generation before running its first turn", async () => {
|
|
const identity = { kind: "conversation" as const, bindingId: "binding-data-1" };
|
|
await testCodexAppServerBindingStore.mutate(identity, {
|
|
kind: "set",
|
|
binding: {
|
|
threadId: "thread-old",
|
|
cwd: "/old-repo",
|
|
conversationStartId: "start-old",
|
|
},
|
|
});
|
|
const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
|
|
let notificationHandler: ((notification: unknown) => void) | undefined;
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
requests.push({ method, params: requestParams });
|
|
if (method === "thread/resume") {
|
|
return conversationThreadStartResult("thread-target");
|
|
}
|
|
if (method === "turn/start") {
|
|
setImmediate(() =>
|
|
notificationHandler?.({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "thread-target",
|
|
turn: {
|
|
id: "turn-1",
|
|
status: "completed",
|
|
items: [{ type: "agentMessage", id: "item-1", text: "rebound" }],
|
|
},
|
|
},
|
|
}),
|
|
);
|
|
return { turn: { id: "turn-1" } };
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}),
|
|
addNotificationHandler: vi.fn((handler: (notification: unknown) => void) => {
|
|
notificationHandler = handler;
|
|
return () => undefined;
|
|
}),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
});
|
|
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "continue",
|
|
channel: "telegram",
|
|
isGroup: false,
|
|
commandAuthorized: true,
|
|
},
|
|
{
|
|
channelId: "telegram",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "telegram",
|
|
accountId: "default",
|
|
conversationId: "5185575566",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 2,
|
|
bindingId: "binding-data-1",
|
|
workspaceDir: "/new-repo",
|
|
start: { id: "start-new", threadId: "thread-target" },
|
|
},
|
|
},
|
|
},
|
|
{ timeoutMs: 500 },
|
|
);
|
|
|
|
expect(result).toEqual({ handled: true, reply: { text: "rebound" } });
|
|
expect(requests.map((request) => request.method)).toEqual(["thread/resume", "turn/start"]);
|
|
expect(requests[0]?.params.threadId).toBe("thread-target");
|
|
expect(requests[1]?.params.cwd).toBe("/new-repo");
|
|
await expect(testCodexAppServerBindingStore.read(identity)).resolves.toMatchObject({
|
|
threadId: "thread-target",
|
|
cwd: "/new-repo",
|
|
conversationStartId: "start-new",
|
|
});
|
|
});
|
|
|
|
it("moves bounded visible session history into an isolated owned conversation thread", async () => {
|
|
const source = {
|
|
agentId: "main",
|
|
sessionId: "source-session",
|
|
sessionKey: "agent:main:source-session",
|
|
threadId: "thread-source",
|
|
};
|
|
const storePath = path.join(tempDir, "source.sqlite");
|
|
await upsertSessionEntry({
|
|
...source,
|
|
storePath,
|
|
entry: { sessionId: source.sessionId, updatedAt: Date.now() },
|
|
});
|
|
await appendSessionTranscriptMessageByIdentity({
|
|
...source,
|
|
storePath,
|
|
message: { role: "user", content: "Earlier question", timestamp: 1 },
|
|
});
|
|
await appendSessionTranscriptMessageByIdentity({
|
|
...source,
|
|
storePath,
|
|
message: { role: "assistant", content: "Earlier answer", timestamp: 2 },
|
|
});
|
|
const sourceIdentity = {
|
|
kind: "session" as const,
|
|
agentId: source.agentId,
|
|
sessionId: source.sessionId,
|
|
sessionKey: source.sessionKey,
|
|
};
|
|
await testCodexAppServerBindingStore.mutate(sourceIdentity, {
|
|
kind: "set",
|
|
binding: {
|
|
threadId: source.threadId,
|
|
clientId: "source-client",
|
|
cwd: tempDir,
|
|
dynamicToolsFingerprint: "harness-only-tools",
|
|
},
|
|
});
|
|
const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
|
|
const notificationHandlers = new Set<(notification: unknown) => void>();
|
|
const releaseSource = vi.fn(async () => undefined);
|
|
const client = {
|
|
getInstanceId: () => "source-client",
|
|
request: vi.fn(async (method: string, params: Record<string, unknown>) => {
|
|
requests.push({ method, params });
|
|
if (method === "thread/start") {
|
|
return conversationThreadStartResult("thread-bound");
|
|
}
|
|
if (method === "thread/inject_items") {
|
|
return {};
|
|
}
|
|
if (method === "turn/start") {
|
|
queueMicrotask(() => {
|
|
for (const handler of notificationHandlers) {
|
|
handler({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "thread-bound",
|
|
turn: {
|
|
id: "turn-bound",
|
|
status: "completed",
|
|
items: [{ type: "agentMessage", id: "answer", text: "Bound reply" }],
|
|
},
|
|
},
|
|
});
|
|
}
|
|
});
|
|
return { turn: { id: "turn-bound" } };
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}),
|
|
addNotificationHandler: vi.fn((handler: (notification: unknown) => void) => {
|
|
notificationHandlers.add(handler);
|
|
return () => notificationHandlers.delete(handler);
|
|
}),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
addCloseHandler: vi.fn(() => () => undefined),
|
|
} as unknown as CodexAppServerClient;
|
|
ensureCodexAppServerClientRuntime(client, { agentDir: tempDir });
|
|
await retainCodexAppServerLiveThread(client, source.threadId, releaseSource);
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue(client);
|
|
sharedClientMocks.retainSharedCodexAppServerClientByInstanceId.mockReturnValue({
|
|
client,
|
|
release: vi.fn(),
|
|
});
|
|
const { event, ctx } = boundConversationClaim(path.join(tempDir, "session.jsonl"));
|
|
ctx.pluginBinding.data = {
|
|
kind: "codex-app-server-session" as const,
|
|
version: 2 as const,
|
|
bindingId: "binding-source-transfer",
|
|
workspaceDir: tempDir,
|
|
source,
|
|
start: { id: "start-source-transfer" },
|
|
};
|
|
|
|
await expect(
|
|
handleCodexConversationInboundClaim(event, ctx, {
|
|
config: { session: { store: storePath } },
|
|
timeoutMs: 500,
|
|
}),
|
|
).resolves.toEqual({ handled: true, reply: { text: "Bound reply" } });
|
|
|
|
expect(requests.map(({ method }) => method)).toEqual([
|
|
"thread/start",
|
|
"thread/inject_items",
|
|
"turn/start",
|
|
]);
|
|
expect(requests[0]?.params).toMatchObject({
|
|
developerInstructions: expect.stringContaining("bound to an OpenClaw conversation"),
|
|
config: { apps: { _default: { enabled: false } }, "features.apps": false },
|
|
});
|
|
expect(requests[0]?.params).not.toHaveProperty("dynamicTools");
|
|
expect(requests[1]?.params.items).toMatchObject([
|
|
{ role: "user", content: [{ text: "Earlier question" }] },
|
|
{ role: "assistant", content: [{ text: "Earlier answer" }] },
|
|
]);
|
|
expect(releaseSource).toHaveBeenCalledExactlyOnceWith("thread-source");
|
|
await expect(testCodexAppServerBindingStore.read(sourceIdentity)).resolves.toBeUndefined();
|
|
await expect(consumeCodexAppServerLiveThread(client, "thread-bound")).resolves.toEqual(
|
|
expect.objectContaining({ release: expect.any(Function) }),
|
|
);
|
|
});
|
|
|
|
it.each([
|
|
{ label: "a registered source run after its client was retired", exposeClient: false },
|
|
{ label: "a live client whose source subscription is claimed", exposeClient: true },
|
|
])("does not transfer $label into a bound conversation", async ({ exposeClient }) => {
|
|
const source = {
|
|
agentId: "main",
|
|
sessionId: "active-source-session",
|
|
sessionKey: "agent:main:active-source-session",
|
|
threadId: "thread-active-source",
|
|
};
|
|
const sourceIdentity = {
|
|
kind: "session" as const,
|
|
agentId: source.agentId,
|
|
sessionId: source.sessionId,
|
|
sessionKey: source.sessionKey,
|
|
};
|
|
await testCodexAppServerBindingStore.mutate(sourceIdentity, {
|
|
kind: "set",
|
|
binding: { threadId: source.threadId, clientId: "active-client", cwd: tempDir },
|
|
});
|
|
const request = vi.fn(async (method: string) => {
|
|
if (method === "thread/start") {
|
|
return conversationThreadStartResult("thread-active-target");
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
});
|
|
const client = {
|
|
getInstanceId: () => "active-client",
|
|
request,
|
|
addNotificationHandler: vi.fn(() => () => undefined),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
addCloseHandler: vi.fn(() => () => undefined),
|
|
} as unknown as CodexAppServerClient;
|
|
ensureCodexAppServerClientRuntime(client, { agentDir: tempDir });
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue(client);
|
|
if (exposeClient) {
|
|
await retainCodexAppServerLiveThread(client, source.threadId, async () => undefined);
|
|
await consumeCodexAppServerLiveThread(client, source.threadId);
|
|
sharedClientMocks.retainSharedCodexAppServerClientByInstanceId.mockReturnValue({
|
|
client,
|
|
release: vi.fn(),
|
|
});
|
|
}
|
|
const activeRun = {
|
|
queueMessage: async () => undefined,
|
|
isStreaming: () => true,
|
|
isCompacting: () => false,
|
|
abort: vi.fn(),
|
|
};
|
|
if (!exposeClient) {
|
|
setActiveEmbeddedRun(source.sessionId, activeRun, source.sessionKey);
|
|
}
|
|
const { event, ctx } = boundConversationClaim(path.join(tempDir, "active-source.jsonl"));
|
|
ctx.pluginBinding.data = {
|
|
kind: "codex-app-server-session" as const,
|
|
version: 2 as const,
|
|
bindingId: "binding-active-source",
|
|
workspaceDir: tempDir,
|
|
source,
|
|
start: { id: "start-active-source" },
|
|
};
|
|
|
|
try {
|
|
const result = await handleCodexConversationInboundClaim(event, ctx, { timeoutMs: 500 });
|
|
|
|
expect(result?.reply?.text).toContain("active run");
|
|
expect(request.mock.calls.map(([method]) => method)).toEqual(["thread/start"]);
|
|
await expect(testCodexAppServerBindingStore.read(sourceIdentity)).resolves.toMatchObject({
|
|
threadId: source.threadId,
|
|
});
|
|
await expect(
|
|
testCodexAppServerBindingStore.read({
|
|
kind: "conversation",
|
|
bindingId: "binding-active-source",
|
|
}),
|
|
).resolves.not.toHaveProperty("conversationSourceTransferComplete", true);
|
|
} finally {
|
|
if (!exposeClient) {
|
|
clearActiveEmbeddedRun(source.sessionId, activeRun, source.sessionKey);
|
|
}
|
|
}
|
|
});
|
|
|
|
it("recreates a missing bound thread with the stored binding agent runtime policy", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, {
|
|
threadId: "thread-old",
|
|
cwd: tempDir,
|
|
approvalPolicy: "on-request",
|
|
sandbox: "workspace-write",
|
|
});
|
|
const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
|
|
const notificationHandlers: Array<(notification: Record<string, unknown>) => void> = [];
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
requests.push({ method, params: requestParams });
|
|
if (method === "turn/start" && requestParams.threadId === "thread-old") {
|
|
throw new Error("thread not found: thread-old");
|
|
}
|
|
if (method === "thread/start") {
|
|
return {
|
|
thread: { id: "thread-new", sessionId: "session-1", cwd: tempDir },
|
|
model: "gpt-5.4-mini",
|
|
};
|
|
}
|
|
if (method === "turn/start" && requestParams.threadId === "thread-new") {
|
|
setImmediate(() => {
|
|
for (const handler of notificationHandlers) {
|
|
handler({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "thread-new",
|
|
turn: {
|
|
id: "turn-new",
|
|
status: "completed",
|
|
items: [{ id: "assistant-1", type: "agentMessage", text: "Recovered" }],
|
|
},
|
|
},
|
|
});
|
|
}
|
|
});
|
|
return { turn: { id: "turn-new" } };
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}),
|
|
addNotificationHandler: vi.fn((handler) => {
|
|
notificationHandlers.push(handler);
|
|
return () => undefined;
|
|
}),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
});
|
|
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "hi again",
|
|
bodyForAgent: "hi again",
|
|
channel: "telegram",
|
|
isGroup: false,
|
|
commandAuthorized: true,
|
|
},
|
|
{
|
|
channelId: "telegram",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "telegram",
|
|
accountId: "default",
|
|
conversationId: "5185575566",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
agentId: "bot-a",
|
|
},
|
|
},
|
|
},
|
|
{
|
|
timeoutMs: 500,
|
|
config: {
|
|
tools: {
|
|
exec: {
|
|
mode: "auto",
|
|
},
|
|
},
|
|
agents: {
|
|
list: [
|
|
{
|
|
id: "bot-a",
|
|
tools: {
|
|
exec: {
|
|
mode: "full",
|
|
},
|
|
},
|
|
},
|
|
],
|
|
},
|
|
} as never,
|
|
},
|
|
);
|
|
|
|
expect(result).toEqual({ handled: true, reply: { text: "Recovered" } });
|
|
expect(requests.map((request) => request.method)).toEqual([
|
|
"turn/start",
|
|
"thread/start",
|
|
"turn/start",
|
|
]);
|
|
expect(requests[0]?.params.approvalPolicy).toBe("never");
|
|
expect(requests[0]?.params.sandboxPolicy).toEqual({ type: "dangerFullAccess" });
|
|
expect(requests[1]?.params.approvalPolicy).toBe("never");
|
|
expect(requests[1]?.params.sandbox).toBe("danger-full-access");
|
|
expect(requests[2]?.params.approvalPolicy).toBe("never");
|
|
expect(requests[2]?.params.sandboxPolicy).toEqual({ type: "dangerFullAccess" });
|
|
});
|
|
|
|
it("does not silently decline auto-mode approvals during missing thread recovery", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, {
|
|
threadId: "thread-old",
|
|
cwd: tempDir,
|
|
approvalPolicy: "never",
|
|
sandbox: "danger-full-access",
|
|
});
|
|
const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
|
|
const notificationHandlers: Array<(notification: Record<string, unknown>) => void> = [];
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
requests.push({ method, params: requestParams });
|
|
if (method === "turn/start" && requestParams.threadId === "thread-old") {
|
|
throw new Error("thread not found: thread-old");
|
|
}
|
|
if (method === "thread/start") {
|
|
return {
|
|
thread: { id: "thread-new", sessionId: "session-1", cwd: tempDir },
|
|
model: "gpt-5.4-mini",
|
|
};
|
|
}
|
|
if (method === "turn/start" && requestParams.threadId === "thread-new") {
|
|
setImmediate(() => {
|
|
for (const handler of notificationHandlers) {
|
|
handler({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "thread-new",
|
|
turn: {
|
|
id: "turn-new",
|
|
status: "completed",
|
|
items: [{ id: "assistant-1", type: "agentMessage", text: "Recovered" }],
|
|
},
|
|
},
|
|
});
|
|
}
|
|
});
|
|
return { turn: { id: "turn-new" } };
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}),
|
|
addNotificationHandler: vi.fn((handler) => {
|
|
notificationHandlers.push(handler);
|
|
return () => undefined;
|
|
}),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
});
|
|
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "hi again",
|
|
bodyForAgent: "hi again",
|
|
channel: "telegram",
|
|
isGroup: false,
|
|
commandAuthorized: true,
|
|
},
|
|
{
|
|
channelId: "telegram",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "telegram",
|
|
accountId: "default",
|
|
conversationId: "5185575566",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
timeoutMs: 500,
|
|
config: {
|
|
tools: {
|
|
exec: {
|
|
mode: "auto",
|
|
},
|
|
},
|
|
} as never,
|
|
},
|
|
);
|
|
|
|
expect(result?.handled).toBe(true);
|
|
expect(result?.reply?.text).toContain(
|
|
"OpenClaw native Codex conversation binding cannot route interactive approvals yet",
|
|
);
|
|
expect(requests).toEqual([]);
|
|
});
|
|
|
|
it("creates a fresh thread when recovery finds the binding already cleared", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
|
|
const notificationHandlers: Array<(notification: Record<string, unknown>) => void> = [];
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
requests.push({ method, params: requestParams });
|
|
if (method === "thread/start") {
|
|
return {
|
|
thread: { id: "thread-new", sessionId: "session-1", cwd: tempDir },
|
|
model: "gpt-5.5-mini",
|
|
};
|
|
}
|
|
if (method === "turn/start" && requestParams.threadId === "thread-new") {
|
|
setImmediate(() => {
|
|
for (const handler of notificationHandlers) {
|
|
handler({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "thread-new",
|
|
turn: {
|
|
id: "turn-new",
|
|
status: "completed",
|
|
items: [{ id: "assistant-1", type: "agentMessage", text: "Recovered fresh" }],
|
|
},
|
|
},
|
|
});
|
|
}
|
|
});
|
|
return { turn: { id: "turn-new" } };
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}),
|
|
addNotificationHandler: vi.fn((handler) => {
|
|
notificationHandlers.push(handler);
|
|
return () => undefined;
|
|
}),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
});
|
|
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "hi again",
|
|
bodyForAgent: "hi again",
|
|
channel: "telegram",
|
|
isGroup: true,
|
|
commandAuthorized: true,
|
|
},
|
|
{
|
|
channelId: "telegram",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "telegram",
|
|
accountId: "default",
|
|
conversationId: "redacted-group",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
},
|
|
},
|
|
},
|
|
{ timeoutMs: 500 },
|
|
);
|
|
|
|
expect(result).toEqual({ handled: true, reply: { text: "Recovered fresh" } });
|
|
expect(requests.map((request) => request.method)).toEqual(["thread/start", "turn/start"]);
|
|
expect(requests[1]?.params.threadId).toBe("thread-new");
|
|
expect(requests[1]?.params.personality).toBe("none");
|
|
const savedBinding = await readTestConversationBinding(sessionFile);
|
|
expect(savedBinding?.threadId).toBe("thread-new");
|
|
});
|
|
|
|
it("passes sandbox state when resolving bound turn policy", async () => {
|
|
codexRequirementsTomlMock.mockReturnValue(
|
|
[
|
|
'allowed_sandbox_modes = ["read-only", "workspace-write"]',
|
|
'allowed_approval_policies = ["never", "on-request"]',
|
|
'allowed_approvals_reviewers = ["user"]',
|
|
].join("\n"),
|
|
);
|
|
resolveSandboxContextMock.mockResolvedValue({ enabled: true });
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, {
|
|
threadId: "thread-1",
|
|
cwd: tempDir,
|
|
approvalPolicy: "never",
|
|
sandbox: "danger-full-access",
|
|
});
|
|
let notificationHandler: ((notification: unknown) => void) | undefined;
|
|
const turnStartParams: Record<string, unknown>[] = [];
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
if (method === "turn/start") {
|
|
turnStartParams.push(requestParams);
|
|
setImmediate(() =>
|
|
notificationHandler?.({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "thread-1",
|
|
turn: {
|
|
id: "turn-1",
|
|
status: "completed",
|
|
items: [{ type: "agentMessage", id: "item-1", text: "done" }],
|
|
},
|
|
},
|
|
}),
|
|
);
|
|
return { turn: { id: "turn-1" } };
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}),
|
|
addNotificationHandler: vi.fn((handler: (notification: unknown) => void) => {
|
|
notificationHandler = handler;
|
|
return () => undefined;
|
|
}),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
});
|
|
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "continue",
|
|
bodyForAgent: "continue",
|
|
channel: "telegram",
|
|
isGroup: false,
|
|
commandAuthorized: true,
|
|
sessionKey: "agent:main:session-1",
|
|
},
|
|
{
|
|
channelId: "telegram",
|
|
sessionKey: "agent:main:session-1",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "telegram",
|
|
accountId: "default",
|
|
conversationId: "5185575566",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
timeoutMs: 50,
|
|
config: {
|
|
tools: {
|
|
exec: {
|
|
security: "full",
|
|
ask: "on-miss",
|
|
},
|
|
},
|
|
} as never,
|
|
},
|
|
);
|
|
|
|
expect(result?.handled).toBe(true);
|
|
expect(result?.reply?.text).toContain(
|
|
"OpenClaw native Codex conversation binding cannot route interactive approvals yet",
|
|
);
|
|
expect(result?.reply?.text).not.toContain(
|
|
"legacy full exec security with ask requires Codex app-server danger-full-access",
|
|
);
|
|
expect(resolveSandboxContextMock).toHaveBeenCalledWith({
|
|
config: {
|
|
tools: {
|
|
exec: {
|
|
security: "full",
|
|
ask: "on-miss",
|
|
},
|
|
},
|
|
},
|
|
sessionKey: "agent:main:session-1",
|
|
workspaceDir: tempDir,
|
|
});
|
|
expect(turnStartParams).toEqual([]);
|
|
});
|
|
|
|
it("returns a clean failure reply when app-server turn start rejects", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
const sessionKey = "agent:main:dashboard:incognito-bound-failure";
|
|
const agentDir = path.join(tempDir, "agents", "bot-b", "agent");
|
|
await writeTestConversationBinding(sessionFile, {
|
|
threadId: "thread-1",
|
|
cwd: tempDir,
|
|
authProfileId: "openai:work",
|
|
});
|
|
const unhandledRejections: unknown[] = [];
|
|
const onUnhandledRejection = (reason: unknown) => {
|
|
unhandledRejections.push(reason);
|
|
};
|
|
process.on("unhandledRejection", onUnhandledRejection);
|
|
const requests: string[] = [];
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string) => {
|
|
requests.push(method);
|
|
if (method === "turn/start") {
|
|
throw new Error(
|
|
"unexpected status 401 Unauthorized: Missing bearer <@U123> [trusted](https://evil) @here",
|
|
);
|
|
}
|
|
if (method === "thread/unsubscribe") {
|
|
return {};
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}),
|
|
addNotificationHandler: vi.fn(() => () => undefined),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
});
|
|
|
|
try {
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "hi",
|
|
bodyForAgent: "hi",
|
|
channel: "telegram",
|
|
isGroup: false,
|
|
commandAuthorized: true,
|
|
sessionKey,
|
|
},
|
|
{
|
|
channelId: "telegram",
|
|
sessionKey,
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "telegram",
|
|
accountId: "default",
|
|
conversationId: "5185575566",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
agentDir,
|
|
},
|
|
},
|
|
},
|
|
{ timeoutMs: 50 },
|
|
);
|
|
await new Promise<void>((resolve) => {
|
|
setImmediate(resolve);
|
|
});
|
|
|
|
expect(result).toEqual({
|
|
handled: true,
|
|
reply: {
|
|
text: "Codex app-server turn failed: unexpected status 401 Unauthorized: Missing bearer <\uff20U123> \uff3btrusted\uff3d\uff08https://evil\uff09 \uff20here",
|
|
},
|
|
});
|
|
const replyText = result?.reply?.text ?? "";
|
|
expect(replyText).not.toContain("<@U123>");
|
|
expect(replyText).not.toContain("[trusted](https://evil)");
|
|
expect(replyText).not.toContain("@here");
|
|
expect(unhandledRejections).toStrictEqual([]);
|
|
expect(requests).toEqual(["turn/start", "thread/unsubscribe"]);
|
|
await expect(readTestConversationBinding(sessionFile)).resolves.toBeUndefined();
|
|
} finally {
|
|
process.off("unhandledRejection", onUnhandledRejection);
|
|
}
|
|
});
|
|
|
|
it("gracefully retires an incognito client when failed turn cleanup cannot unsubscribe", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
const sessionKey = "agent:main:dashboard:incognito-unsubscribe-failure";
|
|
await writeTestConversationBinding(sessionFile, {
|
|
threadId: "thread-1",
|
|
cwd: tempDir,
|
|
});
|
|
const request = vi.fn(async (method: string) => {
|
|
if (method === "turn/start") {
|
|
throw new Error("original bound turn failure");
|
|
}
|
|
if (method === "thread/unsubscribe") {
|
|
throw new Error("thread unsubscribe failed");
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
});
|
|
const closeAndWait = vi.fn(async () => true);
|
|
const client = {
|
|
request,
|
|
closeAndWait,
|
|
addNotificationHandler: vi.fn(() => () => undefined),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
};
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue(client);
|
|
sharedClientMocks.clearSharedCodexAppServerClientIfCurrentAndUnclaimed.mockReturnValue({
|
|
found: true,
|
|
closed: false,
|
|
});
|
|
sharedClientMocks.retireSharedCodexAppServerClientIfCurrent.mockReturnValue({
|
|
activeLeases: 2,
|
|
closed: false,
|
|
});
|
|
const { event, ctx } = boundConversationClaim(sessionFile, sessionKey);
|
|
|
|
await expect(handleCodexConversationInboundClaim(event, ctx)).resolves.toEqual({
|
|
handled: true,
|
|
reply: { text: "Codex app-server turn failed: original bound turn failure" },
|
|
});
|
|
|
|
expect(request.mock.calls.map(([method]) => method)).toEqual([
|
|
"turn/start",
|
|
"thread/unsubscribe",
|
|
]);
|
|
expect(sharedClientMocks.retireSharedCodexAppServerClientIfCurrent).toHaveBeenCalledWith(
|
|
client,
|
|
);
|
|
expect(closeAndWait).not.toHaveBeenCalled();
|
|
await expect(readTestConversationBinding(sessionFile)).resolves.toBeUndefined();
|
|
});
|
|
|
|
it("preserves the original incognito failure if client retirement also rejects", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
const sessionKey = "agent:main:dashboard:incognito-retirement-failure";
|
|
await writeTestConversationBinding(sessionFile, {
|
|
threadId: "thread-1",
|
|
cwd: tempDir,
|
|
});
|
|
const closeAndWait = vi.fn(async () => {
|
|
throw new Error("client retirement failed");
|
|
});
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
closeAndWait,
|
|
request: vi.fn(async (method: string) => {
|
|
if (method === "turn/start") {
|
|
throw new Error("original bound turn failure");
|
|
}
|
|
if (method === "thread/unsubscribe") {
|
|
throw new Error("thread unsubscribe failed");
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}),
|
|
addNotificationHandler: vi.fn(() => () => undefined),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
});
|
|
const { event, ctx } = boundConversationClaim(sessionFile, sessionKey);
|
|
|
|
await expect(handleCodexConversationInboundClaim(event, ctx)).resolves.toEqual({
|
|
handled: true,
|
|
reply: { text: "Codex app-server turn failed: original bound turn failure" },
|
|
});
|
|
|
|
expect(closeAndWait).toHaveBeenCalledOnce();
|
|
await expect(readTestConversationBinding(sessionFile)).resolves.toBeUndefined();
|
|
});
|
|
|
|
it.each([
|
|
{ label: "ordinary", sessionKey: undefined },
|
|
{ label: "incognito", sessionKey: "agent:main:dashboard:incognito-resume-failure" },
|
|
])(
|
|
"retires an indeterminate $label resume once without closing sibling leases",
|
|
async ({ sessionKey }) => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, {
|
|
threadId: "thread-1",
|
|
cwd: tempDir,
|
|
});
|
|
const request = vi.fn(async (method: string) => {
|
|
if (method === "thread/resume") {
|
|
throw new Error("conversation resume response timed out");
|
|
}
|
|
if (method === "thread/unsubscribe") {
|
|
throw new Error("detached client must not receive another cleanup request");
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
});
|
|
const closeAndWait = vi.fn(async () => true);
|
|
const client = {
|
|
request,
|
|
closeAndWait,
|
|
getInstanceId: () => "replacement-client",
|
|
};
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue(client);
|
|
sharedClientMocks.clearSharedCodexAppServerClientIfCurrentAndUnclaimed
|
|
.mockReturnValueOnce({ found: true, closed: false })
|
|
.mockReturnValue({ found: false, closed: false });
|
|
sharedClientMocks.retireSharedCodexAppServerClientIfCurrent
|
|
.mockReturnValueOnce({ activeLeases: 2, closed: false })
|
|
.mockReturnValue(undefined);
|
|
const { event, ctx } = boundConversationClaim(sessionFile, sessionKey);
|
|
|
|
await expect(handleCodexConversationInboundClaim(event, ctx)).resolves.toEqual({
|
|
handled: true,
|
|
reply: { text: "Codex app-server turn failed: conversation resume response timed out" },
|
|
});
|
|
|
|
expect(request.mock.calls.map(([method]) => method)).toEqual(["thread/resume"]);
|
|
expect(sharedClientMocks.retireSharedCodexAppServerClientIfCurrent).toHaveBeenCalledOnce();
|
|
expect(sharedClientMocks.retireSharedCodexAppServerClientIfCurrent).toHaveBeenCalledWith(
|
|
client,
|
|
);
|
|
expect(closeAndWait).not.toHaveBeenCalled();
|
|
if (sessionKey) {
|
|
await expect(readTestConversationBinding(sessionFile)).resolves.toBeUndefined();
|
|
} else {
|
|
await expect(readTestConversationBinding(sessionFile)).resolves.toMatchObject({
|
|
threadId: "thread-1",
|
|
clientId: "test-client",
|
|
});
|
|
}
|
|
},
|
|
);
|
|
|
|
it("retains a pre-start final after a saturated commentary stream", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, {
|
|
threadId: "thread-1",
|
|
cwd: tempDir,
|
|
});
|
|
let notificationHandler: ((notification: unknown) => void) | undefined;
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string) => {
|
|
if (method !== "turn/start") {
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}
|
|
for (let index = 0; index < 100; index += 1) {
|
|
notificationHandler?.({
|
|
method: "item/completed",
|
|
params: {
|
|
threadId: "thread-1",
|
|
turnId: "turn-1",
|
|
item: {
|
|
type: "agentMessage",
|
|
id: `commentary-${index}`,
|
|
text: `progress ${index}`,
|
|
phase: "commentary",
|
|
},
|
|
},
|
|
});
|
|
}
|
|
notificationHandler?.({
|
|
method: "item/completed",
|
|
params: {
|
|
threadId: "thread-1",
|
|
turnId: "turn-1",
|
|
item: { type: "agentMessage", id: "answer", text: "authoritative answer" },
|
|
},
|
|
});
|
|
notificationHandler?.({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "thread-1",
|
|
turn: { id: "turn-1", status: "completed", error: null, items: [] },
|
|
},
|
|
});
|
|
return { turn: { id: "turn-1" } };
|
|
}),
|
|
addNotificationHandler: vi.fn((handler: (notification: unknown) => void) => {
|
|
notificationHandler = handler;
|
|
return () => undefined;
|
|
}),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
});
|
|
const { event, ctx } = boundConversationClaim(sessionFile);
|
|
|
|
await expect(handleCodexConversationInboundClaim(event, ctx)).resolves.toEqual({
|
|
handled: true,
|
|
reply: { text: "authoritative answer" },
|
|
});
|
|
});
|
|
|
|
it("reports an interrupted bound turn as cancellation instead of a partial reply", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, {
|
|
threadId: "thread-1",
|
|
cwd: tempDir,
|
|
});
|
|
let notificationHandler: ((notification: unknown) => void) | undefined;
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string) => {
|
|
if (method !== "turn/start") {
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}
|
|
queueMicrotask(() => {
|
|
notificationHandler?.({
|
|
method: "item/agentMessage/delta",
|
|
params: {
|
|
threadId: "thread-1",
|
|
turnId: "turn-1",
|
|
itemId: "item-1",
|
|
delta: "unfinished answer",
|
|
},
|
|
});
|
|
notificationHandler?.({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "thread-1",
|
|
turn: { id: "turn-1", status: "interrupted", error: null, items: [] },
|
|
},
|
|
});
|
|
});
|
|
return { turn: { id: "turn-1" } };
|
|
}),
|
|
addNotificationHandler: vi.fn((handler: (notification: unknown) => void) => {
|
|
notificationHandler = handler;
|
|
return () => undefined;
|
|
}),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
});
|
|
const { event, ctx } = boundConversationClaim(sessionFile);
|
|
|
|
await expect(handleCodexConversationInboundClaim(event, ctx)).resolves.toEqual({
|
|
handled: true,
|
|
reply: { text: "Codex app-server turn failed: codex app-server turn interrupted" },
|
|
});
|
|
});
|
|
|
|
it("does not interrupt a provider failure that matches the local timeout message", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, {
|
|
threadId: "thread-1",
|
|
cwd: tempDir,
|
|
});
|
|
let notificationHandler: ((notification: unknown) => void) | undefined;
|
|
const request = vi.fn(async (method: string) => {
|
|
if (method !== "turn/start") {
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}
|
|
queueMicrotask(() => {
|
|
notificationHandler?.({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "thread-1",
|
|
turn: {
|
|
id: "turn-1",
|
|
status: "failed",
|
|
error: { message: "codex app-server bound turn timed out" },
|
|
items: [],
|
|
},
|
|
},
|
|
});
|
|
});
|
|
return { turn: { id: "turn-1" } };
|
|
});
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request,
|
|
addNotificationHandler: vi.fn((handler: (notification: unknown) => void) => {
|
|
notificationHandler = handler;
|
|
return () => undefined;
|
|
}),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
});
|
|
const { event, ctx } = boundConversationClaim(sessionFile);
|
|
|
|
await expect(handleCodexConversationInboundClaim(event, ctx)).resolves.toEqual({
|
|
handled: true,
|
|
reply: { text: "Codex app-server turn failed: codex app-server bound turn timed out" },
|
|
});
|
|
|
|
expect(request.mock.calls.map(([method]) => method)).toEqual(["turn/start"]);
|
|
expect(sharedClientMocks.retireSharedCodexAppServerClientIfCurrent).not.toHaveBeenCalled();
|
|
await expect(readTestConversationBinding(sessionFile)).resolves.toMatchObject({
|
|
threadId: "thread-1",
|
|
});
|
|
});
|
|
|
|
it.each([
|
|
{ label: "ordinary", sessionKey: undefined, interruptFails: false },
|
|
{
|
|
label: "incognito",
|
|
sessionKey: "agent:main:dashboard:incognito-start-timeout",
|
|
interruptFails: false,
|
|
},
|
|
{ label: "ordinary with a failed interrupt", sessionKey: undefined, interruptFails: true },
|
|
{
|
|
label: "incognito with a failed interrupt",
|
|
sessionKey: "agent:main:dashboard:incognito-start-interrupt-failure",
|
|
interruptFails: true,
|
|
},
|
|
])(
|
|
"interrupts an indeterminate $label native turn before cleanup",
|
|
async ({ sessionKey, interruptFails }) => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
const harness = createClientHarness();
|
|
await writeTestConversationBinding(sessionFile, {
|
|
threadId: "thread-1",
|
|
clientId: harness.client.getInstanceId(),
|
|
cwd: tempDir,
|
|
});
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue(harness.client);
|
|
if (interruptFails) {
|
|
sharedClientMocks.retireSharedCodexAppServerClientIfCurrent.mockReturnValueOnce({
|
|
activeLeases: 2,
|
|
closed: false,
|
|
});
|
|
}
|
|
const waitForRequest = async (method: string) =>
|
|
await vi.waitFor(
|
|
() => {
|
|
const request = harness.writes
|
|
.map((write) => JSON.parse(write) as { id: number; method: string; params: unknown })
|
|
.find((message) => message.method === method);
|
|
if (!request) {
|
|
throw new Error(`Codex conversation harness did not write ${method}`);
|
|
}
|
|
return request;
|
|
},
|
|
{ interval: 1, timeout: 5_000 },
|
|
);
|
|
const { event, ctx } = boundConversationClaim(sessionFile, sessionKey);
|
|
const result = handleCodexConversationInboundClaim(event, ctx, {
|
|
pluginConfig: { appServer: { requestTimeoutMs: 100 } },
|
|
});
|
|
const turnStart = await waitForRequest("turn/start");
|
|
const interrupt = await waitForRequest("turn/interrupt");
|
|
expect(interrupt.params).toEqual({ threadId: "thread-1", turnId: "" });
|
|
harness.send({ id: turnStart.id, result: { turn: { id: "turn-1" } } });
|
|
harness.send(
|
|
interruptFails
|
|
? { id: interrupt.id, error: { code: -32_000, message: "startup interrupt failed" } }
|
|
: { id: interrupt.id, result: {} },
|
|
);
|
|
if (sessionKey && !interruptFails) {
|
|
const unsubscribe = await waitForRequest("thread/unsubscribe");
|
|
harness.send({ id: unsubscribe.id, result: {} });
|
|
}
|
|
|
|
await expect(result).resolves.toEqual({
|
|
handled: true,
|
|
reply: { text: "Codex app-server turn failed: turn/start timed out" },
|
|
});
|
|
expect(harness.writes.map((write) => JSON.parse(write).method)).toEqual([
|
|
"turn/start",
|
|
"turn/interrupt",
|
|
...(sessionKey && !interruptFails ? ["thread/unsubscribe"] : []),
|
|
]);
|
|
expect(sharedClientMocks.retireSharedCodexAppServerClientIfCurrent).toHaveBeenCalledTimes(
|
|
interruptFails ? 1 : 0,
|
|
);
|
|
expect(
|
|
readCodexConversationActiveTurn(testConversationIdentity(sessionFile)),
|
|
).toBeUndefined();
|
|
if (sessionKey) {
|
|
await expect(readTestConversationBinding(sessionFile)).resolves.toBeUndefined();
|
|
}
|
|
harness.client.close();
|
|
},
|
|
);
|
|
|
|
it.each([
|
|
{ label: "ordinary", sessionKey: undefined },
|
|
{ label: "incognito", sessionKey: "agent:main:dashboard:incognito-turn-timeout" },
|
|
])(
|
|
"interrupts a timed-out $label turn before removing active tracking and handlers",
|
|
async ({ sessionKey }) => {
|
|
vi.useFakeTimers();
|
|
try {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, { threadId: "thread-1", cwd: tempDir });
|
|
const identity = testConversationIdentity(sessionFile);
|
|
const cleanupEvents: string[] = [];
|
|
const notificationHandlers = new Set<(notification: unknown) => void>();
|
|
const request = vi.fn(async (method: string) => {
|
|
if (method === "turn/start") {
|
|
return { turn: { id: "turn-1" } };
|
|
}
|
|
if (method === "thread/unsubscribe") {
|
|
cleanupEvents.push("unsubscribe");
|
|
return {};
|
|
}
|
|
if (method !== "turn/interrupt") {
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}
|
|
cleanupEvents.push("interrupt");
|
|
expect(readCodexConversationActiveTurn(identity)).toMatchObject({
|
|
threadId: "thread-1",
|
|
turnId: "turn-1",
|
|
});
|
|
queueMicrotask(() => {
|
|
cleanupEvents.push("turn completed");
|
|
expect(readCodexConversationActiveTurn(identity)).toMatchObject({
|
|
threadId: "thread-1",
|
|
turnId: "turn-1",
|
|
});
|
|
for (const handler of notificationHandlers) {
|
|
handler({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "thread-1",
|
|
turn: { id: "turn-1", status: "interrupted", error: null, items: [] },
|
|
},
|
|
});
|
|
}
|
|
});
|
|
return {};
|
|
});
|
|
const client = {
|
|
request,
|
|
addNotificationHandler: vi.fn((handler: (notification: unknown) => void) => {
|
|
notificationHandlers.add(handler);
|
|
return () => {
|
|
notificationHandlers.delete(handler);
|
|
cleanupEvents.push("notification cleanup");
|
|
};
|
|
}),
|
|
addRequestHandler: vi.fn(() => () => cleanupEvents.push("request cleanup")),
|
|
addCloseHandler: vi.fn(() => () => undefined),
|
|
};
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue(client);
|
|
const { event, ctx } = boundConversationClaim(sessionFile, sessionKey);
|
|
const result = handleCodexConversationInboundClaim(event, ctx, { timeoutMs: 100 });
|
|
|
|
await vi.advanceTimersByTimeAsync(0);
|
|
expect(request).toHaveBeenCalledWith("turn/start", expect.any(Object), expect.any(Object));
|
|
await vi.advanceTimersByTimeAsync(100);
|
|
|
|
await expect(result).resolves.toEqual({
|
|
handled: true,
|
|
reply: { text: "Codex app-server turn failed: codex app-server bound turn timed out" },
|
|
});
|
|
expect(request).toHaveBeenCalledWith(
|
|
"turn/interrupt",
|
|
{ threadId: "thread-1", turnId: "turn-1" },
|
|
{ timeoutMs: 5_000 },
|
|
);
|
|
expect(request.mock.calls.map(([method]) => method)).toEqual([
|
|
"turn/start",
|
|
"turn/interrupt",
|
|
...(sessionKey ? ["thread/unsubscribe"] : []),
|
|
]);
|
|
expect(cleanupEvents).toEqual([
|
|
"interrupt",
|
|
"turn completed",
|
|
...(sessionKey ? ["unsubscribe"] : []),
|
|
]);
|
|
expect(sharedClientMocks.retireSharedCodexAppServerClientIfCurrent).not.toHaveBeenCalled();
|
|
expect(readCodexConversationActiveTurn(identity)).toBeUndefined();
|
|
if (sessionKey) {
|
|
await expect(readTestConversationBinding(sessionFile)).resolves.toBeUndefined();
|
|
} else {
|
|
await expect(readTestConversationBinding(sessionFile)).resolves.toMatchObject({
|
|
threadId: "thread-1",
|
|
});
|
|
}
|
|
} finally {
|
|
vi.useRealTimers();
|
|
}
|
|
},
|
|
);
|
|
|
|
it.each([
|
|
{ label: "a failed ordinary", sessionKey: undefined, acknowledged: false },
|
|
{ label: "an unconfirmed ordinary", sessionKey: undefined, acknowledged: true },
|
|
{
|
|
label: "a failed incognito",
|
|
sessionKey: "agent:main:dashboard:incognito-interrupt-failure",
|
|
acknowledged: false,
|
|
},
|
|
])(
|
|
"retires $label timeout interrupt once without closing sibling leases",
|
|
async ({ sessionKey, acknowledged }) => {
|
|
vi.useFakeTimers();
|
|
try {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, { threadId: "thread-1", cwd: tempDir });
|
|
const identity = testConversationIdentity(sessionFile);
|
|
const closeAndWait = vi.fn(async () => true);
|
|
const request = vi.fn(async (method: string) => {
|
|
if (method === "turn/start") {
|
|
return { turn: { id: "turn-1" } };
|
|
}
|
|
if (method === "turn/interrupt") {
|
|
if (acknowledged) {
|
|
return {};
|
|
}
|
|
throw new Error("turn interrupt could not be confirmed");
|
|
}
|
|
if (method === "thread/unsubscribe") {
|
|
throw new Error("detached client must not receive another cleanup request");
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
});
|
|
const client = {
|
|
request,
|
|
closeAndWait,
|
|
addNotificationHandler: vi.fn(() => () => undefined),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
addCloseHandler: vi.fn(() => () => undefined),
|
|
};
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue(client);
|
|
sharedClientMocks.clearSharedCodexAppServerClientIfCurrentAndUnclaimed
|
|
.mockReturnValueOnce({ found: true, closed: false })
|
|
.mockReturnValue({ found: false, closed: false });
|
|
sharedClientMocks.retireSharedCodexAppServerClientIfCurrent
|
|
.mockReturnValueOnce({ activeLeases: 2, closed: false })
|
|
.mockReturnValue(undefined);
|
|
const { event, ctx } = boundConversationClaim(sessionFile, sessionKey);
|
|
const result = handleCodexConversationInboundClaim(event, ctx, { timeoutMs: 100 });
|
|
|
|
await vi.advanceTimersByTimeAsync(0);
|
|
expect(request).toHaveBeenCalledWith("turn/start", expect.any(Object), expect.any(Object));
|
|
await vi.advanceTimersByTimeAsync(100);
|
|
expect(request).toHaveBeenCalledWith(
|
|
"turn/interrupt",
|
|
{ threadId: "thread-1", turnId: "turn-1" },
|
|
{ timeoutMs: 5_000 },
|
|
);
|
|
if (acknowledged) {
|
|
expect(readCodexConversationActiveTurn(identity)).toMatchObject({
|
|
threadId: "thread-1",
|
|
turnId: "turn-1",
|
|
});
|
|
expect(
|
|
sharedClientMocks.retireSharedCodexAppServerClientIfCurrent,
|
|
).not.toHaveBeenCalled();
|
|
await vi.advanceTimersByTimeAsync(5_000);
|
|
}
|
|
|
|
await expect(result).resolves.toEqual({
|
|
handled: true,
|
|
reply: { text: "Codex app-server turn failed: codex app-server bound turn timed out" },
|
|
});
|
|
expect(request.mock.calls.map(([method]) => method)).toEqual([
|
|
"turn/start",
|
|
"turn/interrupt",
|
|
]);
|
|
expect(sharedClientMocks.retireSharedCodexAppServerClientIfCurrent).toHaveBeenCalledOnce();
|
|
expect(sharedClientMocks.retireSharedCodexAppServerClientIfCurrent).toHaveBeenCalledWith(
|
|
client,
|
|
);
|
|
expect(closeAndWait).not.toHaveBeenCalled();
|
|
expect(readCodexConversationActiveTurn(identity)).toBeUndefined();
|
|
if (sessionKey) {
|
|
await expect(readTestConversationBinding(sessionFile)).resolves.toBeUndefined();
|
|
} else {
|
|
await expect(readTestConversationBinding(sessionFile)).resolves.toMatchObject({
|
|
threadId: "thread-1",
|
|
});
|
|
}
|
|
} finally {
|
|
vi.useRealTimers();
|
|
}
|
|
},
|
|
);
|
|
|
|
it("falls back to content when the channel body for agent is blank", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
const agentDir = path.join(tempDir, "agents", "bot-b", "agent");
|
|
await writeTestConversationBinding(sessionFile, {
|
|
threadId: "thread-1",
|
|
cwd: tempDir,
|
|
});
|
|
let notificationHandler: ((notification: unknown) => void) | undefined;
|
|
const turnStartParams: Record<string, unknown>[] = [];
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
if (method === "turn/start") {
|
|
turnStartParams.push(requestParams);
|
|
setImmediate(() =>
|
|
notificationHandler?.({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "thread-1",
|
|
turn: {
|
|
id: "turn-1",
|
|
status: "completed",
|
|
items: [{ type: "agentMessage", id: "item-1", text: "done" }],
|
|
},
|
|
},
|
|
}),
|
|
);
|
|
return { turn: { id: "turn-1" } };
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}),
|
|
addNotificationHandler: vi.fn((handler: (notification: unknown) => void) => {
|
|
notificationHandler = handler;
|
|
return () => undefined;
|
|
}),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
});
|
|
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "use the fallback prompt",
|
|
bodyForAgent: "",
|
|
channel: "telegram",
|
|
isGroup: false,
|
|
commandAuthorized: true,
|
|
},
|
|
{
|
|
channelId: "telegram",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "telegram",
|
|
accountId: "default",
|
|
conversationId: "5185575566",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
agentDir,
|
|
},
|
|
},
|
|
},
|
|
{ timeoutMs: 50 },
|
|
);
|
|
|
|
expect(result).toEqual({ handled: true, reply: { text: "done" } });
|
|
const sharedClientParams = mockCallArg(sharedClientMocks.getSharedCodexAppServerClient) as {
|
|
agentDir?: unknown;
|
|
};
|
|
expect(sharedClientParams?.agentDir).toBe(agentDir);
|
|
expect(turnStartParams[0]?.input).toEqual([
|
|
{ type: "text", text: "use the fallback prompt", text_elements: [] },
|
|
]);
|
|
expect(turnStartParams[0]?.approvalPolicy).toBe("never");
|
|
expect(turnStartParams[0]?.approvalsReviewer).toBe("user");
|
|
expect(turnStartParams[0]?.sandboxPolicy).toEqual({
|
|
type: "dangerFullAccess",
|
|
});
|
|
});
|
|
|
|
it("keeps network-proxy bound app-server turns on their thread permissions profile", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, {
|
|
threadId: "thread-1",
|
|
cwd: tempDir,
|
|
networkProxyProfileName: NETWORK_PROXY_PROFILE_NAME,
|
|
networkProxyConfigFingerprint: NETWORK_PROXY_CONFIG_FINGERPRINT,
|
|
});
|
|
let notificationHandler: ((notification: unknown) => void) | undefined;
|
|
const turnStartParams: Record<string, unknown>[] = [];
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
if (method === "turn/start") {
|
|
turnStartParams.push(requestParams);
|
|
setImmediate(() =>
|
|
notificationHandler?.({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "thread-1",
|
|
turn: {
|
|
id: "turn-1",
|
|
status: "completed",
|
|
items: [{ type: "agentMessage", id: "item-1", text: "done" }],
|
|
},
|
|
},
|
|
}),
|
|
);
|
|
return { turn: { id: "turn-1" } };
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}),
|
|
addNotificationHandler: vi.fn((handler: (notification: unknown) => void) => {
|
|
notificationHandler = handler;
|
|
return () => undefined;
|
|
}),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
});
|
|
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "hello",
|
|
channel: "telegram",
|
|
isGroup: false,
|
|
commandAuthorized: true,
|
|
},
|
|
{
|
|
channelId: "telegram",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "telegram",
|
|
accountId: "default",
|
|
conversationId: "5185575566",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
pluginConfig: {
|
|
appServer: {
|
|
networkProxy: {
|
|
enabled: true,
|
|
domains: { "api.openai.com": "allow" },
|
|
allowUpstreamProxy: true,
|
|
proxyUrl: "http://127.0.0.1:3128",
|
|
},
|
|
},
|
|
},
|
|
timeoutMs: 50,
|
|
},
|
|
);
|
|
|
|
expect(result).toEqual({ handled: true, reply: { text: "done" } });
|
|
expect(turnStartParams[0]).not.toHaveProperty("permissions");
|
|
expect(turnStartParams[0]).not.toHaveProperty("sandboxPolicy");
|
|
});
|
|
|
|
it("refreshes stale network-proxy bound app-server threads before the turn", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, {
|
|
threadId: "thread-old",
|
|
cwd: tempDir,
|
|
networkProxyProfileName: "openclaw-network-stale",
|
|
networkProxyConfigFingerprint: "stale-proxy-config",
|
|
conversationStartId: "start-1",
|
|
conversationSourceTransferComplete: true,
|
|
});
|
|
let notificationHandler: ((notification: unknown) => void) | undefined;
|
|
const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
requests.push({ method, params: requestParams });
|
|
if (method === "thread/start") {
|
|
return conversationThreadStartResult("thread-new");
|
|
}
|
|
if (method === "turn/start") {
|
|
setImmediate(() =>
|
|
notificationHandler?.({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "thread-new",
|
|
turn: {
|
|
id: "turn-1",
|
|
status: "completed",
|
|
items: [{ type: "agentMessage", id: "item-1", text: "done" }],
|
|
},
|
|
},
|
|
}),
|
|
);
|
|
return { turn: { id: "turn-1" } };
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}),
|
|
addNotificationHandler: vi.fn((handler: (notification: unknown) => void) => {
|
|
notificationHandler = handler;
|
|
return () => undefined;
|
|
}),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
});
|
|
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "hello",
|
|
channel: "telegram",
|
|
isGroup: false,
|
|
commandAuthorized: true,
|
|
},
|
|
{
|
|
channelId: "telegram",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "telegram",
|
|
accountId: "default",
|
|
conversationId: "5185575566",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
pluginConfig: {
|
|
appServer: {
|
|
serviceTier: "priority",
|
|
networkProxy: {
|
|
enabled: true,
|
|
domains: { "api.openai.com": "allow" },
|
|
allowUpstreamProxy: true,
|
|
proxyUrl: "http://127.0.0.1:3128",
|
|
},
|
|
},
|
|
},
|
|
timeoutMs: 50,
|
|
},
|
|
);
|
|
|
|
expect(result).toEqual({ handled: true, reply: { text: "done" } });
|
|
expect(requests.map((request) => request.method)).toEqual(["thread/start", "turn/start"]);
|
|
expect(requests[0]?.params.config).toMatchObject(NETWORK_PROXY_CONFIG_PATCH);
|
|
expect(requests[0]?.params).not.toHaveProperty("sandbox");
|
|
expect(requests[0]?.params.serviceTier).toBe("priority");
|
|
expect(requests[1]?.params.threadId).toBe("thread-new");
|
|
expect(requests[1]?.params).not.toHaveProperty("sandboxPolicy");
|
|
const bindingAfterRefresh = await readTestConversationBinding(sessionFile);
|
|
expect(bindingAfterRefresh?.threadId).toBe("thread-new");
|
|
expect(bindingAfterRefresh?.networkProxyProfileName).toBe(NETWORK_PROXY_PROFILE_NAME);
|
|
expect(bindingAfterRefresh?.networkProxyConfigFingerprint).toBe(
|
|
NETWORK_PROXY_CONFIG_FINGERPRINT,
|
|
);
|
|
expect(bindingAfterRefresh?.conversationStartId).toBe("start-1");
|
|
expect(bindingAfterRefresh?.conversationSourceTransferComplete).toBe(true);
|
|
});
|
|
|
|
it("restores the old owner and rolls back a network replacement when unsubscribe fails", async () => {
|
|
const sessionFile = path.join(tempDir, "network-rotation-release-failure.jsonl");
|
|
await writeTestConversationBinding(sessionFile, {
|
|
threadId: "thread-old",
|
|
clientId: "client-network-rotation",
|
|
cwd: tempDir,
|
|
networkProxyProfileName: "openclaw-network-stale",
|
|
networkProxyConfigFingerprint: "stale-proxy-config",
|
|
conversationStartId: "start-1",
|
|
});
|
|
const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
|
|
const client = {
|
|
getInstanceId: () => "client-network-rotation",
|
|
request: vi.fn(async (method: string, params: Record<string, unknown>) => {
|
|
requests.push({ method, params });
|
|
if (method === "thread/start") {
|
|
return conversationThreadStartResult("thread-new");
|
|
}
|
|
if (method === "thread/unsubscribe" && params.threadId === "thread-old") {
|
|
throw new Error("old thread unsubscribe failed");
|
|
}
|
|
if (method === "thread/unsubscribe" && params.threadId === "thread-new") {
|
|
return {};
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}),
|
|
addNotificationHandler: vi.fn(() => () => undefined),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
addCloseHandler: vi.fn(() => () => undefined),
|
|
} as unknown as CodexAppServerClient;
|
|
ensureCodexAppServerClientRuntime(client, { agentDir: tempDir });
|
|
await expect(
|
|
retainCodexAppServerLiveThread(client, "thread-old", async (threadId) => {
|
|
await client.request("thread/unsubscribe", { threadId });
|
|
}),
|
|
).resolves.toBe(true);
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue(client);
|
|
const { event, ctx } = boundConversationClaim(sessionFile);
|
|
|
|
const result = await handleCodexConversationInboundClaim(event, ctx, {
|
|
pluginConfig: NETWORK_PROXY_PLUGIN_CONFIG,
|
|
});
|
|
|
|
expect(result?.reply?.text).toContain("old thread unsubscribe failed");
|
|
expect(requests).toEqual([
|
|
{ method: "thread/start", params: expect.any(Object) },
|
|
{ method: "thread/unsubscribe", params: { threadId: "thread-old" } },
|
|
{ method: "thread/unsubscribe", params: { threadId: "thread-new" } },
|
|
]);
|
|
await expect(readTestConversationBinding(sessionFile)).resolves.toMatchObject({
|
|
clientId: "client-network-rotation",
|
|
threadId: "thread-old",
|
|
});
|
|
expect(isCodexAppServerLiveThreadClaimed(client, "thread-old")).toBe(false);
|
|
await expect(consumeCodexAppServerLiveThread(client, "thread-old")).resolves.toEqual(
|
|
expect.objectContaining({ release: expect.any(Function) }),
|
|
);
|
|
await expect(consumeCodexAppServerLiveThread(client, "thread-new")).resolves.toBeUndefined();
|
|
});
|
|
|
|
it("blocks Guardian-mode bound turns with stale no-approval policy on custom model providers", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, {
|
|
threadId: "thread-1",
|
|
cwd: tempDir,
|
|
model: "local-model",
|
|
modelProvider: "lmstudio",
|
|
approvalPolicy: "never",
|
|
sandbox: "danger-full-access",
|
|
});
|
|
let notificationHandler: ((notification: unknown) => void) | undefined;
|
|
const turnStartParams: Record<string, unknown>[] = [];
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
if (method === "turn/start") {
|
|
turnStartParams.push(requestParams);
|
|
setImmediate(() =>
|
|
notificationHandler?.({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "thread-1",
|
|
turn: {
|
|
id: "turn-1",
|
|
status: "completed",
|
|
items: [{ type: "agentMessage", id: "item-1", text: "done" }],
|
|
},
|
|
},
|
|
}),
|
|
);
|
|
return { turn: { id: "turn-1" } };
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}),
|
|
addNotificationHandler: vi.fn((handler: (notification: unknown) => void) => {
|
|
notificationHandler = handler;
|
|
return () => undefined;
|
|
}),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
});
|
|
|
|
const result = await handleCodexConversationInboundClaim(
|
|
{
|
|
content: "hello",
|
|
channel: "telegram",
|
|
isGroup: false,
|
|
commandAuthorized: true,
|
|
},
|
|
{
|
|
channelId: "telegram",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "telegram",
|
|
accountId: "default",
|
|
conversationId: "5185575566",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
timeoutMs: 50,
|
|
pluginConfig: {
|
|
appServer: {
|
|
mode: "guardian",
|
|
},
|
|
},
|
|
},
|
|
);
|
|
|
|
expect(result?.handled).toBe(true);
|
|
expect(result?.reply?.text).toContain(
|
|
"OpenClaw native Codex conversation binding cannot route interactive approvals yet",
|
|
);
|
|
expect(turnStartParams).toEqual([]);
|
|
expect(sharedClientMocks.getSharedCodexAppServerClient).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("infers custom model providers for legacy bound turns without stored modelProvider", async () => {
|
|
const sessionFile = path.join(tempDir, "session.jsonl");
|
|
await writeTestConversationBinding(sessionFile, {
|
|
threadId: "thread-1",
|
|
cwd: tempDir,
|
|
model: "lmstudio/local-model",
|
|
approvalPolicy: "on-request",
|
|
sandbox: "workspace-write",
|
|
});
|
|
let notificationHandler: ((notification: unknown) => void) | undefined;
|
|
const turnStartParams: Record<string, unknown>[] = [];
|
|
sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue({
|
|
request: vi.fn(async (method: string, requestParams: Record<string, unknown>) => {
|
|
if (method === "turn/start") {
|
|
turnStartParams.push(requestParams);
|
|
setImmediate(() =>
|
|
notificationHandler?.({
|
|
method: "turn/completed",
|
|
params: {
|
|
threadId: "thread-1",
|
|
turn: {
|
|
id: "turn-1",
|
|
status: "completed",
|
|
items: [{ type: "agentMessage", id: "item-1", text: "done" }],
|
|
},
|
|
},
|
|
}),
|
|
);
|
|
return { turn: { id: "turn-1" } };
|
|
}
|
|
throw new Error(`unexpected method: ${method}`);
|
|
}),
|
|
addNotificationHandler: vi.fn((handler: (notification: unknown) => void) => {
|
|
notificationHandler = handler;
|
|
return () => undefined;
|
|
}),
|
|
addRequestHandler: vi.fn(() => () => undefined),
|
|
});
|
|
|
|
await expect(
|
|
handleCodexConversationInboundClaim(
|
|
{
|
|
content: "hello",
|
|
channel: "telegram",
|
|
isGroup: false,
|
|
commandAuthorized: true,
|
|
},
|
|
{
|
|
channelId: "telegram",
|
|
pluginBinding: {
|
|
bindingId: "binding-1",
|
|
pluginId: "codex",
|
|
pluginRoot: tempDir,
|
|
channel: "telegram",
|
|
accountId: "default",
|
|
conversationId: "5185575566",
|
|
boundAt: Date.now(),
|
|
data: {
|
|
kind: "codex-app-server-session",
|
|
version: 1,
|
|
sessionFile,
|
|
workspaceDir: tempDir,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
timeoutMs: 50,
|
|
pluginConfig: {
|
|
appServer: {
|
|
mode: "guardian",
|
|
},
|
|
},
|
|
},
|
|
),
|
|
).resolves.toMatchObject({
|
|
handled: true,
|
|
reply: {
|
|
text: expect.stringContaining(
|
|
"OpenClaw native Codex conversation binding cannot route interactive approvals yet",
|
|
),
|
|
},
|
|
});
|
|
|
|
expect(turnStartParams).toEqual([]);
|
|
expect(sharedClientMocks.getSharedCodexAppServerClient).not.toHaveBeenCalled();
|
|
});
|
|
});
|
|
/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */
|