mirror of
https://github.com/openclaw/openclaw.git
synced 2026-11-03 14:12:35 +00:00
fix(codex): preserve native assignments across parent rotation (#160952)
* test(upgrade): isolate native assignment fixture lifecycle * test(upgrade): use explicit fixture control flow * test(upgrade): decode native fixture websocket frames * fix(codex): preserve native assignments across parent rotation * test(codex): cancel native rotation at the commit boundary * test(codex): use fresh MCP fixture threads * test(codex): remove redundant attestation call assertion * test(codex): tighten rotation fixture types * test(codex): return a valid workspace resume response * test(codex): isolate schema lifecycle bindings * test(codex): route assignment lifecycle through the database broker * ci: refresh native assignment checks after main repairs * ci: refresh native validation after parent-audit fixture repair * ci: refresh native checks after duplicate route repair
This commit is contained in:
Peter Steinberger
GitHub
parent
7e9d57f290
commit
5367527c27
17 changed files with 1080 additions and 119 deletions
|
|
@ -3259,18 +3259,17 @@ describe("runCodexAppServerAttempt", () => {
|
|||
async (method) => {
|
||||
if (method === "turn/start" && compactDuringAcceptance) {
|
||||
compactDuringAcceptance = false;
|
||||
await harness.notify({
|
||||
method: "item/completed",
|
||||
params: {
|
||||
threadId: "thread-1",
|
||||
turnId: "turn-1",
|
||||
item: { type: "contextCompaction", id: "compact-during-start" },
|
||||
},
|
||||
});
|
||||
await harness.notify(
|
||||
itemNotification("item/completed", {
|
||||
type: "contextCompaction",
|
||||
id: "compact-during-start",
|
||||
}),
|
||||
);
|
||||
}
|
||||
if (method === "turn/start") {
|
||||
turnRequested.resolve();
|
||||
}
|
||||
return method === "thread/resume" ? threadStartResult("thread-1") : undefined;
|
||||
},
|
||||
{ persistedThreads: [] },
|
||||
);
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
|||
import { threadStartResult as nativeThreadStartResult } from "./codex-app-server.test-fixtures.js";
|
||||
import { createCodexTestHostCapabilities } from "./host-capability.test-support.js";
|
||||
import type { CodexThreadStartParams } from "./protocol.js";
|
||||
import { testCodexAppServerBindingStore } from "./session-binding.test-helpers.js";
|
||||
import { createCodexTestBindingStore } from "./session-binding.test-helpers.js";
|
||||
import { createCodexTestModel } from "./test-support.js";
|
||||
import { startOrResumeThread as startOrResumeThreadImpl } from "./thread-lifecycle.js";
|
||||
import { createAppServerOptions as createBaseAppServerOptions } from "./thread-lifecycle.test-fixtures.js";
|
||||
|
|
@ -20,10 +20,11 @@ import { createAppServerOptions as createBaseAppServerOptions } from "./thread-l
|
|||
function startOrResumeThread(
|
||||
params: Omit<Parameters<typeof startOrResumeThreadImpl>[0], "bindingStore">,
|
||||
) {
|
||||
return startOrResumeThreadImpl({ ...params, bindingStore: testCodexAppServerBindingStore });
|
||||
return startOrResumeThreadImpl({ ...params, bindingStore });
|
||||
}
|
||||
|
||||
let tempDir: string;
|
||||
let bindingStore: ReturnType<typeof createCodexTestBindingStore>;
|
||||
|
||||
function createParams(sessionFile: string, workspaceDir: string): EmbeddedRunAttemptParams {
|
||||
return {
|
||||
|
|
@ -66,6 +67,7 @@ function threadStartResult(threadId = "thread-1", serviceTier: string | null = n
|
|||
|
||||
describe("Codex app-server dynamic tool schema boundary contract", () => {
|
||||
beforeEach(async () => {
|
||||
bindingStore = createCodexTestBindingStore();
|
||||
tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-codex-schema-contract-"));
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -103,7 +103,7 @@ export async function withCodexThreadLifecycleBinding(
|
|||
|
||||
type PendingResumeContext = CodexThreadRequestContext & {
|
||||
binding: CodexAppServerThreadBinding;
|
||||
clearCurrentBinding: (operation: string) => Promise<void>;
|
||||
stageBindingReplacement: (operation: string) => void;
|
||||
releaseRetainedThread: (threadId: string, assertCurrent: () => void) => Promise<boolean>;
|
||||
transientRestriction: boolean;
|
||||
};
|
||||
|
|
|
|||
|
|
@ -105,7 +105,7 @@ export async function resumeExistingCodexThread(
|
|||
lifecycleTiming,
|
||||
normalizeBindingModelProvider,
|
||||
throwIfAborted,
|
||||
clearCurrentBinding,
|
||||
stageBindingReplacement,
|
||||
} = context;
|
||||
let acceptedConfiguration: CodexThreadResumePreparation | undefined;
|
||||
let disposeConfiguration: (() => void) | undefined;
|
||||
|
|
@ -413,7 +413,7 @@ export async function resumeExistingCodexThread(
|
|||
embeddedAgentLog.warn("codex app-server thread resume failed; starting a new thread", {
|
||||
error,
|
||||
});
|
||||
await clearCurrentBinding("rotating a stale thread binding");
|
||||
stageBindingReplacement("rotating a stale thread binding");
|
||||
} finally {
|
||||
disposeConfiguration?.();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -34,7 +34,6 @@ import {
|
|||
prepareCodexThreadResume,
|
||||
withCodexThreadLifecycleBinding,
|
||||
} from "./thread-lifecycle-adoption.js";
|
||||
import { CodexThreadBindingConflictError } from "./thread-lifecycle-errors.js";
|
||||
import { resumeExistingCodexThread, startFreshCodexThread } from "./thread-lifecycle-io.js";
|
||||
import {
|
||||
buildCodexThreadBindingPolicy,
|
||||
|
|
@ -234,26 +233,18 @@ export async function startOrResumeThread(
|
|||
}),
|
||||
);
|
||||
}
|
||||
const clearCurrentBinding = async (operation: string) => {
|
||||
const stageBindingReplacement = (operation: string) => {
|
||||
const current = binding;
|
||||
if (!current?.threadId) {
|
||||
return;
|
||||
}
|
||||
assertCodexBindingMayBeReplaced(current, operation, expectedOwnership);
|
||||
const cleared = await params.bindingStore.mutate(
|
||||
bindingIdentity,
|
||||
{
|
||||
kind: "clear",
|
||||
threadId: current.threadId,
|
||||
},
|
||||
assert,
|
||||
);
|
||||
if (!cleared) {
|
||||
throw new CodexThreadBindingConflictError(current.threadId, operation);
|
||||
}
|
||||
assert();
|
||||
// Replacement CAS needs the predecessor row to preserve same-connection inventory.
|
||||
replacementPredecessor = current;
|
||||
binding = undefined;
|
||||
};
|
||||
const rotateStaleBinding = async (
|
||||
const rotateStaleBinding = (
|
||||
current: CodexAppServerThreadBinding,
|
||||
changed: string,
|
||||
details?: Record<string, unknown>,
|
||||
|
|
@ -262,7 +253,7 @@ export async function startOrResumeThread(
|
|||
threadId: current.threadId,
|
||||
...details,
|
||||
});
|
||||
await clearCurrentBinding("rotating a stale thread binding");
|
||||
stageBindingReplacement("rotating a stale thread binding");
|
||||
};
|
||||
const transientDelegationRestriction = params.params.delegationCapability === "report_only";
|
||||
const persistentWebSearchRestriction =
|
||||
|
|
@ -274,7 +265,7 @@ export async function startOrResumeThread(
|
|||
const resumed = await resumePendingCodexThread(params, {
|
||||
...(await prepareRequestContext()),
|
||||
binding,
|
||||
clearCurrentBinding,
|
||||
stageBindingReplacement,
|
||||
releaseRetainedThread: (threadId, assertCurrent) =>
|
||||
releaseRetainedThread(threadId, initialBoundClientId, assertCurrent),
|
||||
transientRestriction:
|
||||
|
|
@ -290,7 +281,7 @@ export async function startOrResumeThread(
|
|||
!restrictedToolSurface &&
|
||||
binding.nativeToolPolicyRestricted === true
|
||||
) {
|
||||
await clearCurrentBinding("rotating a host-policy-restricted thread binding");
|
||||
stageBindingReplacement("rotating a host-policy-restricted thread binding");
|
||||
}
|
||||
if (
|
||||
binding?.threadId &&
|
||||
|
|
@ -300,7 +291,7 @@ export async function startOrResumeThread(
|
|||
"codex app-server native skill isolation changed; starting a new thread",
|
||||
{ threadId: binding.threadId },
|
||||
);
|
||||
await clearCurrentBinding("rotating stale native skill isolation");
|
||||
stageBindingReplacement("rotating stale native skill isolation");
|
||||
}
|
||||
if (
|
||||
binding?.threadId &&
|
||||
|
|
@ -313,7 +304,7 @@ export async function startOrResumeThread(
|
|||
embeddedAgentLog.debug("codex app-server ring-zero restriction changed; rotating thread", {
|
||||
threadId: binding.threadId,
|
||||
});
|
||||
await clearCurrentBinding("rotating a ring-zero thread binding");
|
||||
stageBindingReplacement("rotating a ring-zero thread binding");
|
||||
}
|
||||
if (
|
||||
binding?.threadId &&
|
||||
|
|
@ -326,7 +317,7 @@ export async function startOrResumeThread(
|
|||
binding: binding.appServerRuntimeFingerprint,
|
||||
})
|
||||
) {
|
||||
await rotateStaleBinding(binding, "runtime identity", {
|
||||
rotateStaleBinding(binding, "runtime identity", {
|
||||
connectionClass: params.appServer.connectionClass,
|
||||
});
|
||||
}
|
||||
|
|
@ -347,7 +338,7 @@ export async function startOrResumeThread(
|
|||
requestedModel: params.params.modelId,
|
||||
},
|
||||
);
|
||||
await clearCurrentBinding("rotating a GPT-5.6 multi-agent thread binding");
|
||||
stageBindingReplacement("rotating a GPT-5.6 multi-agent thread binding");
|
||||
}
|
||||
selectionBinding = binding;
|
||||
// Capability read failures use managed search for this turn but must not
|
||||
|
|
@ -413,17 +404,11 @@ export async function startOrResumeThread(
|
|||
// Scheduled configured MCP moved from Codex-native config to OpenClaw dynamic tools.
|
||||
// A persistent main/named session has one binding: rotate its exact predecessor instead
|
||||
// of retaining native and scheduled variants that could diverge or widen authority.
|
||||
assertCodexBindingMayBeReplaced(
|
||||
predecessorBinding,
|
||||
"changing configured MCP ownership",
|
||||
expectedOwnership,
|
||||
);
|
||||
stageBindingReplacement("changing configured MCP ownership");
|
||||
embeddedAgentLog.debug(
|
||||
"codex app-server configured MCP ownership changed; starting a new thread",
|
||||
{ threadId: predecessorBinding.threadId },
|
||||
);
|
||||
replacementPredecessor = predecessorBinding;
|
||||
binding = undefined;
|
||||
preserveExistingBinding = false;
|
||||
}
|
||||
if (
|
||||
|
|
@ -446,7 +431,7 @@ export async function startOrResumeThread(
|
|||
);
|
||||
preserveExistingBinding = true;
|
||||
} else {
|
||||
await rotateStaleBinding(binding, "MCP config");
|
||||
rotateStaleBinding(binding, "MCP config");
|
||||
}
|
||||
binding = undefined;
|
||||
}
|
||||
|
|
@ -478,7 +463,7 @@ export async function startOrResumeThread(
|
|||
} else {
|
||||
// Codex can ignore resume overrides for a loaded thread, so persistent
|
||||
// search-policy changes and legacy bindings without metadata rotate first.
|
||||
await rotateStaleBinding(binding, "web search config");
|
||||
rotateStaleBinding(binding, "web search config");
|
||||
}
|
||||
binding = undefined;
|
||||
}
|
||||
|
|
@ -516,7 +501,7 @@ export async function startOrResumeThread(
|
|||
!contextEngineBinding ||
|
||||
!isContextEngineBindingCompatible(binding.contextEngine, contextEngineBinding)
|
||||
) {
|
||||
await rotateStaleBinding(binding, "context-engine binding", {
|
||||
rotateStaleBinding(binding, "context-engine binding", {
|
||||
engineId: contextEngineBinding?.engineId,
|
||||
previousEngineId: binding.contextEngine?.engineId,
|
||||
epoch: contextEngineBinding?.projection?.epoch,
|
||||
|
|
@ -537,14 +522,14 @@ export async function startOrResumeThread(
|
|||
nextLegacy: legacyUserMcpServersFingerprint,
|
||||
})
|
||||
) {
|
||||
await rotateStaleBinding(binding, "user MCP config");
|
||||
rotateStaleBinding(binding, "user MCP config");
|
||||
}
|
||||
if (
|
||||
binding?.threadId &&
|
||||
(binding.networkProxyConfigFingerprint !== networkProxyConfigFingerprint ||
|
||||
binding.networkProxyProfileName !== params.appServer.networkProxy?.profileName)
|
||||
) {
|
||||
await rotateStaleBinding(binding, "network proxy config");
|
||||
rotateStaleBinding(binding, "network proxy config");
|
||||
}
|
||||
if (binding?.threadId) {
|
||||
const pluginBindingStale = isCodexPluginThreadBindingStale({
|
||||
|
|
@ -555,7 +540,7 @@ export async function startOrResumeThread(
|
|||
hasBindingPolicyContext: Boolean(binding.pluginAppPolicyContext),
|
||||
});
|
||||
if (pluginBindingStale) {
|
||||
await rotateStaleBinding(binding, "plugin app config");
|
||||
rotateStaleBinding(binding, "plugin app config");
|
||||
}
|
||||
}
|
||||
if (binding?.threadId) {
|
||||
|
|
@ -565,7 +550,7 @@ export async function startOrResumeThread(
|
|||
binding.dynamicToolsContainDeferred !== dynamicToolsContainDeferred &&
|
||||
(binding.dynamicToolsContainDeferred !== undefined || !dynamicToolsContainDeferred)
|
||||
) {
|
||||
await rotateStaleBinding(binding, "dynamic tool loading");
|
||||
rotateStaleBinding(binding, "dynamic tool loading");
|
||||
}
|
||||
}
|
||||
if (binding?.threadId) {
|
||||
|
|
@ -596,7 +581,7 @@ export async function startOrResumeThread(
|
|||
},
|
||||
);
|
||||
} else {
|
||||
await rotateStaleBinding(binding, "dynamic tool catalog");
|
||||
rotateStaleBinding(binding, "dynamic tool catalog");
|
||||
}
|
||||
} else {
|
||||
const requestContext = await prepareRequestContext();
|
||||
|
|
@ -612,7 +597,7 @@ export async function startOrResumeThread(
|
|||
}
|
||||
if (incognito || warmReuse.kind === "rotate") {
|
||||
throwIfAborted();
|
||||
await clearCurrentBinding(
|
||||
stageBindingReplacement(
|
||||
incognito
|
||||
? "rotating an unavailable ephemeral thread binding"
|
||||
: "rotating a stale plugin app binding",
|
||||
|
|
@ -622,7 +607,7 @@ export async function startOrResumeThread(
|
|||
const resumed = await resumeExistingCodexThread(params, {
|
||||
...requestContext,
|
||||
binding: resumeBinding,
|
||||
clearCurrentBinding,
|
||||
stageBindingReplacement,
|
||||
prebuiltFinalConfigPatch: warmReuse.prebuiltFinalConfigPatch,
|
||||
prebuiltPluginThreadConfig,
|
||||
buildLoadedPluginThreadConfig,
|
||||
|
|
@ -644,7 +629,7 @@ export async function startOrResumeThread(
|
|||
|
||||
assertCodexBindingMayBeReplaced(binding, "starting a fresh native thread", expectedOwnership);
|
||||
const requestContext = await prepareRequestContext();
|
||||
if (initialBoundThreadId && !preserveExistingBinding && !replacementPredecessor) {
|
||||
if (initialBoundThreadId && !preserveExistingBinding && !configuredMcpOwnershipChanged) {
|
||||
await releaseRetainedThread(initialBoundThreadId);
|
||||
}
|
||||
const started = await startFreshCodexThread(params, {
|
||||
|
|
@ -654,9 +639,8 @@ export async function startOrResumeThread(
|
|||
rotatedContextEngineBinding,
|
||||
replacementPredecessor,
|
||||
});
|
||||
if (replacementPredecessor) {
|
||||
// The predecessor remains authoritative through thread/start and exact-owner CAS.
|
||||
// Release only that prior subscription after the successor has committed.
|
||||
if (replacementPredecessor && !preserveExistingBinding && configuredMcpOwnershipChanged) {
|
||||
// Configured MCP migration retains the prior subscription until its replacement commits.
|
||||
await releaseRetainedThread(replacementPredecessor.threadId, replacementPredecessor.clientId);
|
||||
}
|
||||
return publishCodexThreadInferenceBinding(params, started);
|
||||
|
|
|
|||
|
|
@ -144,7 +144,7 @@ export type CodexThreadResumePreparation = {
|
|||
|
||||
export type CodexResumeThreadContext = CodexThreadRequestContext & {
|
||||
binding: CodexAppServerThreadBinding;
|
||||
clearCurrentBinding: (operation: string) => Promise<void>;
|
||||
stageBindingReplacement: (operation: string) => void;
|
||||
prebuiltPluginThreadConfig?: CodexPluginThreadConfig;
|
||||
buildLoadedPluginThreadConfig?: (
|
||||
binding: CodexAppServerThreadBinding,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,320 @@
|
|||
import assert from "node:assert/strict";
|
||||
import path from "node:path";
|
||||
import { createDeferred } from "openclaw/plugin-sdk/extension-shared";
|
||||
import { afterEach, describe, expect, it, onTestFinished, vi } from "vitest";
|
||||
import {
|
||||
claimCodexAppServerLiveThread,
|
||||
ensureCodexAppServerClientRuntime,
|
||||
retainCodexAppServerLiveThread,
|
||||
} from "./client-runtime.js";
|
||||
import { createCodexNativeSubagentHistoryOwner } from "./native-subagent-history-owner.js";
|
||||
import type { CodexNativeSubagentPendingAssignment } from "./native-subagent-pending-assignments.js";
|
||||
import {
|
||||
bindProductionHarnessHostCapabilitiesForTest,
|
||||
setupRunAttemptTestHooks,
|
||||
} from "./run-attempt-test-harness.js";
|
||||
import {
|
||||
bindingStoreKey,
|
||||
createCodexAppServerBindingStore,
|
||||
type StoredCodexAppServerBinding,
|
||||
} from "./session-binding.js";
|
||||
import { createCodexTestBindingStateStore } from "./session-binding.test-helpers.js";
|
||||
import { useAutoCleanupTempDirTracker } from "./test-support.js";
|
||||
import type { CodexStartOrResumeThreadParams } from "./thread-lifecycle-types.js";
|
||||
import { startOrResumeThread } from "./thread-lifecycle.js";
|
||||
import {
|
||||
createAppServerOptions,
|
||||
createCodexLifecycleHarness,
|
||||
createParams,
|
||||
resetThreadLifecycleTestFixtures,
|
||||
threadStartResult,
|
||||
} from "./thread-lifecycle.test-fixtures.js";
|
||||
|
||||
setupRunAttemptTestHooks();
|
||||
const tempDirs = useAutoCleanupTempDirTracker(afterEach);
|
||||
afterEach(resetThreadLifecycleTestFixtures);
|
||||
|
||||
async function fixture() {
|
||||
const cwd = tempDirs.make("codex-assignment-rotation-");
|
||||
const params = createParams(path.join(cwd, "session.jsonl"), cwd);
|
||||
params.agentId = "main";
|
||||
const identity = {
|
||||
kind: "session" as const,
|
||||
agentId: "main",
|
||||
sessionId: params.sessionId,
|
||||
sessionKey: params.sessionKey,
|
||||
};
|
||||
const key = bindingStoreKey(identity);
|
||||
const values = new Map<string, StoredCodexAppServerBinding>();
|
||||
const store = createCodexAppServerBindingStore(createCodexTestBindingStateStore(values));
|
||||
let starts = 0;
|
||||
let successor: (() => Promise<ReturnType<typeof threadStartResult>>) | undefined;
|
||||
const wire = createCodexLifecycleHarness({
|
||||
respond: async (method) => {
|
||||
if (method === "config/read") {
|
||||
return { config: {}, origins: {}, layers: [] };
|
||||
}
|
||||
if (method === "configRequirements/read") {
|
||||
return { requirements: null };
|
||||
}
|
||||
if (method === "thread/start") {
|
||||
starts += 1;
|
||||
return starts > 1 && successor ? await successor() : threadStartResult(`parent-${starts}`);
|
||||
}
|
||||
if (method === "thread/delete") {
|
||||
return {};
|
||||
}
|
||||
throw new Error(`Unexpected native lifecycle request: ${method}`);
|
||||
},
|
||||
});
|
||||
ensureCodexAppServerClientRuntime(wire.client, { agentDir: cwd });
|
||||
onTestFinished(async () => {
|
||||
await wire.client.closeAndWait();
|
||||
});
|
||||
const options: CodexStartOrResumeThreadParams = {
|
||||
client: wire.client,
|
||||
bindingStore: store,
|
||||
params,
|
||||
cwd,
|
||||
dynamicTools: [],
|
||||
appServer: createAppServerOptions(),
|
||||
nativeCodeModeEnabled: true,
|
||||
userMcpServersEnabled: false,
|
||||
mcpServersFingerprint: "mcp-before",
|
||||
mcpServersFingerprintEvaluated: true,
|
||||
appServerRuntimeFingerprint: "connection-A",
|
||||
};
|
||||
const parent = await startOrResumeThread(options);
|
||||
const releasePredecessor = vi.fn(async (threadId: string) => {
|
||||
await wire.client.request("thread/unsubscribe", { threadId });
|
||||
});
|
||||
await retainCodexAppServerLiveThread(
|
||||
wire.client,
|
||||
parent.threadId,
|
||||
releasePredecessor,
|
||||
parent.liveThreadConfigFingerprint,
|
||||
);
|
||||
const owner = createCodexNativeSubagentHistoryOwner({
|
||||
parentThreadId: parent.threadId,
|
||||
sessionId: params.sessionId,
|
||||
binding: parent,
|
||||
});
|
||||
assert(owner);
|
||||
const assignments: CodexNativeSubagentPendingAssignment[] = [
|
||||
{
|
||||
runId: "codex-thread:running-child",
|
||||
childThreadId: "running-child",
|
||||
nativeParentThreadId: parent.threadId,
|
||||
owner,
|
||||
},
|
||||
{
|
||||
runId: "codex-thread:finished-child",
|
||||
childThreadId: "finished-child",
|
||||
nativeParentThreadId: parent.threadId,
|
||||
owner,
|
||||
recordedCompletion: {
|
||||
childThreadId: "finished-child",
|
||||
status: "succeeded",
|
||||
statusLabel: "recorded_task_result",
|
||||
result: "Retained child result",
|
||||
completedAt: 200,
|
||||
},
|
||||
},
|
||||
];
|
||||
const stored = values.get(key);
|
||||
assert(stored?.state === "active");
|
||||
values.set(key, {
|
||||
...stored,
|
||||
nativeSubagentAssignments: { version: 1, assignments },
|
||||
nativeSubagentTaskImport: { version: 1, taskIds: ["legacy-running", "legacy-finished"] },
|
||||
});
|
||||
const readAssignments = (threadId: string, binding = parent) => {
|
||||
const currentOwner = createCodexNativeSubagentHistoryOwner({
|
||||
parentThreadId: threadId,
|
||||
sessionId: params.sessionId,
|
||||
binding,
|
||||
});
|
||||
assert(currentOwner);
|
||||
assert(typeof store.readNativeSubagentAssignments === "function");
|
||||
return store.readNativeSubagentAssignments(identity, currentOwner);
|
||||
};
|
||||
const readState = () => {
|
||||
const current = values.get(key);
|
||||
if (!current) {
|
||||
return undefined;
|
||||
}
|
||||
const { lease: _lease, ...durable } = current;
|
||||
return structuredClone(durable);
|
||||
};
|
||||
expect(readAssignments(parent.threadId)).toEqual(assignments);
|
||||
return {
|
||||
...wire,
|
||||
options,
|
||||
identity,
|
||||
readState,
|
||||
store,
|
||||
parent,
|
||||
assignments,
|
||||
releasePredecessor,
|
||||
readAssignments,
|
||||
setSuccessor: (respond: () => Promise<ReturnType<typeof threadStartResult>>) => {
|
||||
successor = respond;
|
||||
},
|
||||
rotate: (patch: Partial<CodexStartOrResumeThreadParams> = {}) =>
|
||||
startOrResumeThread({ ...options, mcpServersFingerprint: "mcp-after", ...patch }),
|
||||
};
|
||||
}
|
||||
|
||||
describe("native assignment custody across ordinary parent rotation", () => {
|
||||
it("keeps imported assignments until the ordinary replacement commits", async () => {
|
||||
const f = await fixture();
|
||||
const before = f.readState();
|
||||
const entered = createDeferred<void>();
|
||||
const proceed = createDeferred<void>();
|
||||
f.setSuccessor(async () => {
|
||||
entered.resolve();
|
||||
await proceed.promise;
|
||||
return threadStartResult("parent-2");
|
||||
});
|
||||
const pending = f.rotate();
|
||||
const settled = pending.then(
|
||||
(value) => ({ value }),
|
||||
(error: unknown) => ({ error }),
|
||||
);
|
||||
try {
|
||||
await Promise.race([
|
||||
entered.promise,
|
||||
settled.then(() => {
|
||||
throw new Error("Replacement settled before reaching native thread/start");
|
||||
}),
|
||||
]);
|
||||
expect(f.readState()).toEqual(before);
|
||||
expect(f.readAssignments(f.parent.threadId)).toEqual(f.assignments);
|
||||
expect(f.releasePredecessor).toHaveBeenCalledExactlyOnceWith(f.parent.threadId);
|
||||
} finally {
|
||||
proceed.resolve();
|
||||
await settled;
|
||||
}
|
||||
const replacement = await pending;
|
||||
expect(f.store.read(f.identity)?.threadId).toBe("parent-2");
|
||||
expect(f.readAssignments(replacement.threadId, replacement)).toEqual(f.assignments);
|
||||
});
|
||||
|
||||
it("does not expose the old assignments through a different connection", async () => {
|
||||
const f = await fixture();
|
||||
const replacement = await f.rotate({ appServerRuntimeFingerprint: "connection-B" });
|
||||
expect(replacement.appServerRuntimeFingerprint).toBe("connection-B");
|
||||
expect(f.readAssignments(replacement.threadId, replacement)).toEqual([]);
|
||||
});
|
||||
|
||||
it.each(["start", "abort", "revoked", "conflict"] as const)(
|
||||
"preserves authoritative assignments when the successor encounters %s failure",
|
||||
async (failure) => {
|
||||
const f = await fixture();
|
||||
const controller = new AbortController();
|
||||
let expected = f.readState();
|
||||
assert(expected?.state === "active");
|
||||
const predecessor = expected.binding;
|
||||
const closeHost =
|
||||
failure === "revoked"
|
||||
? await bindProductionHarnessHostCapabilitiesForTest(f.options.params, {
|
||||
profileId: "rotation-operator",
|
||||
scopes: ["operator.write"],
|
||||
assertCurrent: () => {},
|
||||
})
|
||||
: undefined;
|
||||
if (closeHost) {
|
||||
onTestFinished(closeHost);
|
||||
}
|
||||
if (failure === "abort") {
|
||||
const mutate = f.store.mutate.bind(f.store);
|
||||
vi.spyOn(f.store, "mutate").mockImplementation((identity, mutation, assertCurrent) => {
|
||||
if (mutation.kind === "replace-thread") {
|
||||
controller.abort(new Error("Rotation aborted"));
|
||||
}
|
||||
return mutate(identity, mutation, assertCurrent);
|
||||
});
|
||||
}
|
||||
f.setSuccessor(async () => {
|
||||
if (failure === "start") {
|
||||
throw new Error("Successor start rejected");
|
||||
}
|
||||
if (failure === "revoked") {
|
||||
closeHost?.();
|
||||
}
|
||||
if (failure === "conflict") {
|
||||
expect(
|
||||
await f.store.mutate(f.identity, {
|
||||
kind: "replace-thread",
|
||||
expectedThreadId: f.parent.threadId,
|
||||
binding: { ...predecessor, threadId: "competing-parent" },
|
||||
}),
|
||||
).toBe(true);
|
||||
expected = f.readState();
|
||||
}
|
||||
return threadStartResult("parent-uncommitted");
|
||||
});
|
||||
await expect(f.rotate({ signal: controller.signal })).rejects.toThrow(
|
||||
{
|
||||
start: "Successor start rejected",
|
||||
abort: "Rotation aborted",
|
||||
revoked: "agent harness host capability is no longer active",
|
||||
conflict: "Codex thread binding changed while committing a fresh thread: parent-1",
|
||||
}[failure],
|
||||
);
|
||||
expect(f.readState()).toEqual(expected);
|
||||
assert(expected?.state === "active");
|
||||
expect(f.readAssignments(expected.binding.threadId)).toEqual(f.assignments);
|
||||
expect(f.request.mock.calls.filter(([method]) => method === "thread/start")).toHaveLength(2);
|
||||
expect(
|
||||
f.request.mock.calls
|
||||
.filter(([method]) => method === "thread/delete")
|
||||
.map(([, params]) => params),
|
||||
).toEqual(failure === "start" ? [] : [{ threadId: "parent-uncommitted" }]);
|
||||
expect(f.releasePredecessor).toHaveBeenCalledExactlyOnceWith(f.parent.threadId);
|
||||
},
|
||||
);
|
||||
|
||||
it("refuses a claimed predecessor before starting or committing a successor", async () => {
|
||||
const f = await fixture();
|
||||
const before = f.readState();
|
||||
const claim = await claimCodexAppServerLiveThread(f.client, f.parent.threadId);
|
||||
assert(claim);
|
||||
try {
|
||||
await expect(f.rotate()).rejects.toThrow("claimed by active work; stop it first");
|
||||
expect(f.readState()).toEqual(before);
|
||||
expect(f.readAssignments(f.parent.threadId)).toEqual(f.assignments);
|
||||
expect(f.request.mock.calls.filter(([method]) => method === "thread/start")).toHaveLength(1);
|
||||
expect(f.releasePredecessor).not.toHaveBeenCalled();
|
||||
expect(() => claim.assertCurrent()).not.toThrow();
|
||||
} finally {
|
||||
await claim.release(f.parent.threadId);
|
||||
}
|
||||
});
|
||||
|
||||
it("keeps an early rotation transient until a later durable turn", async () => {
|
||||
const f = await fixture();
|
||||
await f.store.mutate(f.identity, {
|
||||
kind: "patch",
|
||||
threadId: f.parent.threadId,
|
||||
patch: { nativeSkillIsolationFingerprint: "previous-skill-policy" },
|
||||
});
|
||||
const before = f.readState();
|
||||
const transient = await startOrResumeThread({
|
||||
...f.options,
|
||||
params: { ...f.options.params, delegationCapability: "report_only" },
|
||||
});
|
||||
expect(transient.lifecycle.preserveExistingBinding).toBe(true);
|
||||
expect(transient.liveThreadConfigFingerprint).toBeUndefined();
|
||||
expect(f.readState()).toEqual(before);
|
||||
expect(f.readAssignments(f.parent.threadId)).toEqual(f.assignments);
|
||||
expect(f.releasePredecessor).not.toHaveBeenCalled();
|
||||
|
||||
const replacement = await startOrResumeThread(f.options);
|
||||
expect(replacement.threadId).toBe("parent-3");
|
||||
expect(replacement.lifecycle.preserveExistingBinding).toBeUndefined();
|
||||
expect(f.store.read(f.identity)?.threadId).toBe(replacement.threadId);
|
||||
expect(f.readAssignments(replacement.threadId, replacement)).toEqual(f.assignments);
|
||||
expect(f.releasePredecessor).toHaveBeenCalledExactlyOnceWith(f.parent.threadId);
|
||||
});
|
||||
});
|
||||
|
|
@ -2831,6 +2831,7 @@ describe("Codex app-server thread lifecycle bindings", () => {
|
|||
modelProvider: "openai",
|
||||
dynamicToolsFingerprint: "[]",
|
||||
});
|
||||
const predecessor = await readCodexAppServerBinding(sessionFile);
|
||||
const params = createParams(sessionFile, workspaceDir);
|
||||
params.toolsAllow = ["openclaw"];
|
||||
if (ephemeral) {
|
||||
|
|
@ -2880,8 +2881,7 @@ describe("Codex app-server thread lifecycle bindings", () => {
|
|||
"mcpServerStatus/list",
|
||||
ephemeral ? "thread/unsubscribe" : "thread/delete",
|
||||
]);
|
||||
expect(request.mock.calls.some(([method]) => method === "turn/start")).toBe(false);
|
||||
expect(await readCodexAppServerBinding(sessionFile)).toBeUndefined();
|
||||
expect(await readCodexAppServerBinding(sessionFile)).toEqual(predecessor);
|
||||
},
|
||||
);
|
||||
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import assert from "node:assert/strict";
|
||||
import fs from "node:fs/promises";
|
||||
import http from "node:http";
|
||||
import path from "node:path";
|
||||
|
|
@ -25,11 +26,9 @@ import {
|
|||
writePolicyProbeServer,
|
||||
} from "./thread-lifecycle.user-mcp-servers.test-support.js";
|
||||
|
||||
function createRequest(
|
||||
startThreadId: string,
|
||||
options: { readRequirements?: boolean; resumeThreadId?: string } = {},
|
||||
) {
|
||||
return vi.fn(async (method: string, _params: unknown) => {
|
||||
function createRequest(startThreadId: string, options: { readRequirements?: boolean } = {}) {
|
||||
let startCount = 0;
|
||||
return vi.fn(async (method: string, params: unknown) => {
|
||||
if (method === "config/read") {
|
||||
return { config: {}, origins: {}, layers: [] };
|
||||
}
|
||||
|
|
@ -37,10 +36,17 @@ function createRequest(
|
|||
return { requirements: null };
|
||||
}
|
||||
if (method === "thread/start") {
|
||||
return threadStartResult(startThreadId);
|
||||
startCount += 1;
|
||||
return threadStartResult(startCount === 1 ? startThreadId : `${startThreadId}-${startCount}`);
|
||||
}
|
||||
if (method === "thread/resume" && options.resumeThreadId) {
|
||||
return threadResumeResult(options.resumeThreadId);
|
||||
if (method === "thread/resume") {
|
||||
assert(
|
||||
params !== null &&
|
||||
typeof params === "object" &&
|
||||
"threadId" in params &&
|
||||
typeof params.threadId === "string",
|
||||
);
|
||||
return threadResumeResult(params.threadId);
|
||||
}
|
||||
throw new Error(`unexpected method: ${method}`);
|
||||
});
|
||||
|
|
@ -80,7 +86,7 @@ describe("startOrResumeThread — user mcp.servers projection (regression: #8081
|
|||
},
|
||||
},
|
||||
};
|
||||
const request = createRequest("thread-policy", { resumeThreadId: "thread-policy" });
|
||||
const request = createRequest("thread-policy");
|
||||
let wire = await createLeasedCodexLifecycleHarness({
|
||||
agentDir: path.join(tempDir, "agent"),
|
||||
respond: request,
|
||||
|
|
@ -236,7 +242,7 @@ describe("startOrResumeThread — user mcp.servers projection (regression: #8081
|
|||
},
|
||||
},
|
||||
} as unknown as EmbeddedRunAttemptParams["config"];
|
||||
const request = createRequest("thread-beta5", { resumeThreadId: "thread-beta5" });
|
||||
const request = createRequest("thread-beta5");
|
||||
let wire = await createLeasedCodexLifecycleHarness({
|
||||
agentDir: path.join(tempDir, "agent"),
|
||||
respond: request,
|
||||
|
|
@ -250,6 +256,7 @@ describe("startOrResumeThread — user mcp.servers projection (regression: #8081
|
|||
await run();
|
||||
const currentBinding = await readCodexAppServerBinding(sessionFile);
|
||||
expect(currentBinding).toBeDefined();
|
||||
expect(currentBinding?.threadId).toBe("thread-beta5");
|
||||
|
||||
const legacyFingerprint = JSON.stringify({
|
||||
mcp_servers: {
|
||||
|
|
@ -278,6 +285,8 @@ describe("startOrResumeThread — user mcp.servers projection (regression: #8081
|
|||
"thread/start",
|
||||
]);
|
||||
const convergedBinding = await readCodexAppServerBinding(sessionFile);
|
||||
expect(convergedBinding?.threadId).toBe("thread-beta5-2");
|
||||
expect(convergedBinding?.threadId).not.toBe(currentBinding?.threadId);
|
||||
expect(convergedBinding?.userMcpServersFingerprint).toMatch(/^sha256:[a-f0-9]{64}$/);
|
||||
expect(convergedBinding?.userMcpServersFingerprint).not.toContain("beta5-access-token");
|
||||
expect(convergedBinding?.userMcpServersFingerprint).not.toBe(legacyFingerprint);
|
||||
|
|
@ -289,7 +298,7 @@ describe("startOrResumeThread — user mcp.servers projection (regression: #8081
|
|||
wire = await createLeasedCodexLifecycleHarness({
|
||||
agentDir: path.join(tempDir, "agent"),
|
||||
respond: request,
|
||||
persistedThreads: ["thread-beta5"],
|
||||
persistedThreads: ["thread-beta5-2"],
|
||||
});
|
||||
request.mockClear();
|
||||
await run();
|
||||
|
|
@ -305,6 +314,13 @@ describe("startOrResumeThread — user mcp.servers projection (regression: #8081
|
|||
"thread/resume",
|
||||
"thread/inject_items",
|
||||
]);
|
||||
expect(request).toHaveBeenCalledWith(
|
||||
"thread/resume",
|
||||
expect.objectContaining({ threadId: convergedBinding?.threadId }),
|
||||
);
|
||||
expect((await readCodexAppServerBinding(sessionFile))?.threadId).toBe(
|
||||
convergedBinding?.threadId,
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
|
|
@ -371,15 +387,14 @@ describe("startOrResumeThread — user mcp.servers projection (regression: #8081
|
|||
},
|
||||
},
|
||||
}) as unknown as EmbeddedRunAttemptParams["config"];
|
||||
const request = createRequest("thread-with-current-bearer", {
|
||||
resumeThreadId: "thread-with-stale-bearer",
|
||||
});
|
||||
const request = createRequest("thread-with-current-bearer");
|
||||
|
||||
await startOrResumeThread({
|
||||
client: { request } as never,
|
||||
...lifecycleOptions(sessionFile, workspaceDir, createConfig("Bearer access-token-one")),
|
||||
});
|
||||
const firstBinding = await readCodexAppServerBinding(sessionFile);
|
||||
expect(firstBinding?.threadId).toBe("thread-with-current-bearer");
|
||||
expect(firstBinding?.userMcpServersFingerprint).toMatch(/^sha256:[a-f0-9]{64}$/);
|
||||
expect(firstBinding?.userMcpServersFingerprint).not.toContain("access-token-one");
|
||||
|
||||
|
|
@ -402,6 +417,8 @@ describe("startOrResumeThread — user mcp.servers projection (regression: #8081
|
|||
"Bearer access-token-two",
|
||||
);
|
||||
const secondBinding = await readCodexAppServerBinding(sessionFile);
|
||||
expect(secondBinding?.threadId).toBe("thread-with-current-bearer-2");
|
||||
expect(secondBinding?.threadId).not.toBe(firstBinding?.threadId);
|
||||
expect(secondBinding?.userMcpServersFingerprint).toMatch(/^sha256:[a-f0-9]{64}$/);
|
||||
expect(secondBinding?.userMcpServersFingerprint).not.toContain("access-token-two");
|
||||
expect(secondBinding?.userMcpServersFingerprint).not.toBe(
|
||||
|
|
|
|||
|
|
@ -50,6 +50,10 @@ const nativeAssignmentLogs = [
|
|||
"native-assignment-eligibility.json",
|
||||
"native-assignment-baseline.json",
|
||||
"native-assignment-first-hop.json",
|
||||
"native-assignment-inventory-after-first-hop.json",
|
||||
"native-assignment-inventory-before-recovery.json",
|
||||
"native-assignment-inventory-after-recovery.json",
|
||||
"native-assignment-inventory-live-final.json",
|
||||
"native-assignment-proof.json",
|
||||
"native-assignment-messages.jsonl",
|
||||
"native-assignment-server.log",
|
||||
|
|
|
|||
|
|
@ -32,6 +32,7 @@ const completeId = "native-upgrade-complete";
|
|||
const phases = new Set([
|
||||
"seed-history",
|
||||
"seed-assignments",
|
||||
"handoff",
|
||||
"recover",
|
||||
"close-loaded",
|
||||
"close-gone",
|
||||
|
|
@ -44,6 +45,7 @@ const subscriptions = new Map();
|
|||
const loaded = new Set();
|
||||
const closeTurns = new Map();
|
||||
let turnSequence = 0;
|
||||
let parentSequence = 0;
|
||||
let seededAssignments = false;
|
||||
let historyPruned = false;
|
||||
|
||||
|
|
@ -52,7 +54,7 @@ function readPhase() {
|
|||
if (!phases.has(phase)) {
|
||||
throw new Error("Unknown native assignment fixture phase");
|
||||
}
|
||||
if (!phase.startsWith("seed-") && !historyPruned) {
|
||||
if (!phase.startsWith("seed-") && phase !== "handoff" && !historyPruned) {
|
||||
// Recovery must use the imported locator/result, not rediscover old spawn events.
|
||||
for (const turn of threads.get(parentId)?.turns ?? []) {
|
||||
turn.items = turn.items.filter((item) => item.type !== "collabAgentToolCall");
|
||||
|
|
@ -114,29 +116,35 @@ function thread(id) {
|
|||
return value;
|
||||
}
|
||||
|
||||
function startThread(params, id = parentId) {
|
||||
function startThread(params, requestedId) {
|
||||
let id = requestedId;
|
||||
if (id === undefined) {
|
||||
parentSequence += 1;
|
||||
id = parentSequence === 1 ? parentId : `${parentId}-${parentSequence}`;
|
||||
}
|
||||
if (threads.has(id)) {
|
||||
throw new Error("Synthetic thread/start must create a fresh thread");
|
||||
}
|
||||
const response = createFakeThreadStartResponse({
|
||||
params,
|
||||
threadId: id,
|
||||
sessionId: "native-upgrade-provider-session",
|
||||
version,
|
||||
});
|
||||
if (!threads.has(id)) {
|
||||
response.thread.createdAt = Math.floor(Date.now() / 1000);
|
||||
response.thread.updatedAt = response.thread.createdAt;
|
||||
if (id !== parentId) {
|
||||
response.thread.parentThreadId = parentId;
|
||||
response.thread.source = {
|
||||
subAgent: {
|
||||
thread_spawn: { parent_thread_id: parentId, depth: 1, agent_path: `/root/${id}` },
|
||||
},
|
||||
};
|
||||
response.thread.preview = "Synthetic native upgrade assignment";
|
||||
}
|
||||
threads.set(id, response.thread);
|
||||
injectedItemsByThread.set(id, []);
|
||||
threadConfigurations.set(id, { params: structuredClone(params), response });
|
||||
response.thread.createdAt = Math.floor(Date.now() / 1000);
|
||||
response.thread.updatedAt = response.thread.createdAt;
|
||||
if (id === runningId || id === completeId) {
|
||||
response.thread.parentThreadId = parentId;
|
||||
response.thread.source = {
|
||||
subAgent: {
|
||||
thread_spawn: { parent_thread_id: parentId, depth: 1, agent_path: `/root/${id}` },
|
||||
},
|
||||
};
|
||||
response.thread.preview = "Synthetic native upgrade assignment";
|
||||
}
|
||||
threads.set(id, response.thread);
|
||||
injectedItemsByThread.set(id, []);
|
||||
threadConfigurations.set(id, { params: structuredClone(params), response });
|
||||
loaded.add(id);
|
||||
return { ...threadConfigurations.get(id).response, thread: thread(id) };
|
||||
}
|
||||
|
|
@ -255,15 +263,24 @@ function completeTurn(socket, phase, threadId, turn, text, status = "completed")
|
|||
notify(socket, phase, "thread/status/changed", { threadId, status: { type: "idle" } });
|
||||
}
|
||||
|
||||
function collabItem(socket, phase, turn, item, method) {
|
||||
function collabItem(socket, phase, threadId, turn, item, method) {
|
||||
if (method === "item/completed") {
|
||||
turn.items.push(item);
|
||||
}
|
||||
notify(socket, phase, method, { threadId: parentId, turnId: turn.id, item });
|
||||
notify(socket, phase, method, { threadId, turnId: turn.id, item });
|
||||
}
|
||||
|
||||
function runPhase(socket, phase, turn) {
|
||||
if (phase === "seed-assignments") {
|
||||
function runPhase(socket, phase, threadId, turn, input) {
|
||||
if (turn.status !== "inProgress") {
|
||||
return;
|
||||
}
|
||||
if (
|
||||
phase === "seed-assignments" &&
|
||||
threadId === parentId &&
|
||||
input.some(
|
||||
(item) => item.type === "text" && item.text.includes("NATIVE_UPGRADE_SEED_ASSIGNMENTS"),
|
||||
)
|
||||
) {
|
||||
if (seededAssignments) {
|
||||
throw new Error("Native assignments were already seeded");
|
||||
}
|
||||
|
|
@ -277,6 +294,7 @@ function runPhase(socket, phase, turn) {
|
|||
collabItem(
|
||||
socket,
|
||||
phase,
|
||||
threadId,
|
||||
turn,
|
||||
{
|
||||
id: `spawn-${id}`,
|
||||
|
|
@ -297,21 +315,32 @@ function runPhase(socket, phase, turn) {
|
|||
}
|
||||
return;
|
||||
}
|
||||
if (phase === "close-loaded" || phase === "close-gone") {
|
||||
if (
|
||||
(phase === "close-loaded" || phase === "close-gone") &&
|
||||
input.some(
|
||||
(item) =>
|
||||
item.type === "text" &&
|
||||
item.text.includes(`Continue the synthetic native upgrade fixture: ${phase}.`),
|
||||
)
|
||||
) {
|
||||
if (closeTurns.has(socket)) {
|
||||
throw new Error("Native close confirmation is already pending on this connection");
|
||||
}
|
||||
const item = {
|
||||
id: `${phase}-${turn.id}`,
|
||||
type: "collabAgentToolCall",
|
||||
tool: "closeAgent",
|
||||
status: "inProgress",
|
||||
senderThreadId: parentId,
|
||||
senderThreadId: threadId,
|
||||
receiverThreadIds: [runningId],
|
||||
agentsStates: {},
|
||||
};
|
||||
closeTurns.set(socket, { phase, turn });
|
||||
collabItem(socket, phase, turn, item, "item/started");
|
||||
closeTurns.set(socket, { phase, threadId, turn });
|
||||
collabItem(socket, phase, threadId, turn, item, "item/started");
|
||||
collabItem(
|
||||
socket,
|
||||
phase,
|
||||
threadId,
|
||||
turn,
|
||||
{
|
||||
...item,
|
||||
|
|
@ -325,7 +354,7 @@ function runPhase(socket, phase, turn) {
|
|||
completeTurn(
|
||||
socket,
|
||||
phase,
|
||||
parentId,
|
||||
threadId,
|
||||
turn,
|
||||
phase === "seed-history" ? "NATIVE_UPGRADE_RETAINED_HISTORY" : "NATIVE_UPGRADE_PARENT_OK",
|
||||
);
|
||||
|
|
@ -450,32 +479,41 @@ function handle(socket, phase, message) {
|
|||
return result({ status: removed ? "unsubscribed" : "notSubscribed" });
|
||||
}
|
||||
case "turn/start": {
|
||||
if (params.threadId !== parentId) {
|
||||
if ([runningId, completeId].includes(params.threadId)) {
|
||||
throw new Error("Only parent turns may be started through this fixture");
|
||||
}
|
||||
const turnId = `native-parent-turn-${++turnSequence}`;
|
||||
const turn = startTurn(socket, phase, parentId, turnId);
|
||||
const turn = startTurn(socket, phase, params.threadId, turnId);
|
||||
result({ turn });
|
||||
setImmediate(() => {
|
||||
try {
|
||||
runPhase(socket, phase, turn);
|
||||
runPhase(socket, phase, params.threadId, turn, params.input ?? []);
|
||||
} catch (error) {
|
||||
log("fixture-error", phase, { message: error.message });
|
||||
completeTurn(socket, phase, parentId, turn, "NATIVE_UPGRADE_FIXTURE_ERROR", "failed");
|
||||
completeTurn(
|
||||
socket,
|
||||
phase,
|
||||
params.threadId,
|
||||
turn,
|
||||
"NATIVE_UPGRADE_FIXTURE_ERROR",
|
||||
"failed",
|
||||
);
|
||||
}
|
||||
});
|
||||
return;
|
||||
}
|
||||
case "turn/interrupt": {
|
||||
if (params.threadId !== parentId) {
|
||||
if ([runningId, completeId].includes(params.threadId)) {
|
||||
throw new Error("This fixture only interrupts parent turns");
|
||||
}
|
||||
const turn = thread(parentId).turns.find((value) => value.id === params.turnId);
|
||||
if (!turn) {
|
||||
throw new Error("Unknown synthetic parent turn");
|
||||
const turn = thread(params.threadId).turns.find((value) => value.id === params.turnId);
|
||||
if (turn?.status !== "inProgress") {
|
||||
throw new Error("No active synthetic parent turn to interrupt");
|
||||
}
|
||||
result({});
|
||||
completeTurn(socket, phase, parentId, turn, undefined, "interrupted");
|
||||
setImmediate(() =>
|
||||
completeTurn(socket, phase, params.threadId, turn, undefined, "interrupted"),
|
||||
);
|
||||
return;
|
||||
}
|
||||
case "thread/loaded/list": {
|
||||
|
|
@ -495,7 +533,7 @@ function handle(socket, phase, message) {
|
|||
completeTurn(
|
||||
socket,
|
||||
closing.phase,
|
||||
parentId,
|
||||
closing.threadId,
|
||||
closing.turn,
|
||||
"NATIVE_UPGRADE_CLOSE_OBSERVED",
|
||||
),
|
||||
|
|
|
|||
|
|
@ -132,7 +132,7 @@ function runParent(phase, wait = true) {
|
|||
agentId: "native-proof",
|
||||
message:
|
||||
phase === "seed-assignments"
|
||||
? "Spawn two native child agents for this synthetic upgrade fixture."
|
||||
? "NATIVE_UPGRADE_SEED_ASSIGNMENTS: Spawn two native child agents for this synthetic upgrade fixture."
|
||||
: `Continue the synthetic native upgrade fixture: ${phase}.`,
|
||||
deliver: false,
|
||||
},
|
||||
|
|
@ -160,13 +160,19 @@ function snapshot() {
|
|||
)
|
||||
.all(SESSION_KEY)
|
||||
.map((row) => Object.assign({}, row));
|
||||
const baselinePath = artifact("native-assignment-baseline.json");
|
||||
const bindingKey = fs.existsSync(baselinePath) ? readJson(baselinePath).binding.key : undefined;
|
||||
const bindings = db
|
||||
.prepare(
|
||||
"SELECT entry_key, value_json FROM plugin_state_entries WHERE plugin_id = 'codex' AND namespace = 'app-server-thread-bindings'",
|
||||
)
|
||||
.all()
|
||||
.map((row) => ({ key: row.entry_key, value: JSON.parse(row.value_json) }))
|
||||
.filter((row) => row.value.binding?.threadId === "native-upgrade-parent");
|
||||
.filter((row) =>
|
||||
bindingKey
|
||||
? row.key === bindingKey
|
||||
: row.value.binding?.threadId === "native-upgrade-parent",
|
||||
);
|
||||
assert.equal(bindings.length, 1, "native parent binding was missing or ambiguous");
|
||||
return { databasePath, tasks, binding: bindings[0] };
|
||||
} finally {
|
||||
|
|
@ -174,6 +180,243 @@ function snapshot() {
|
|||
}
|
||||
}
|
||||
|
||||
function captureInventory(stage) {
|
||||
const observation = { stage, capturedAt: new Date().toISOString() };
|
||||
const fingerprint = (value) => createHash("sha256").update(value).digest("hex");
|
||||
try {
|
||||
const baseline = readJson(artifact("native-assignment-baseline.json"));
|
||||
const db = new DatabaseSync(
|
||||
path.join(required("OPENCLAW_STATE_DIR"), "state", "openclaw.sqlite"),
|
||||
{
|
||||
readOnly: true,
|
||||
},
|
||||
);
|
||||
try {
|
||||
const row = db
|
||||
.prepare(
|
||||
"SELECT value_json FROM plugin_state_entries WHERE plugin_id = 'codex' AND namespace = 'app-server-thread-bindings' AND entry_key = ?",
|
||||
)
|
||||
.get(baseline.binding.key);
|
||||
const stored = row ? JSON.parse(row.value_json) : undefined;
|
||||
const binding = stored?.binding;
|
||||
observation.binding = stored
|
||||
? {
|
||||
state: stored.state,
|
||||
sessionId: stored.sessionId,
|
||||
threadId: binding?.threadId,
|
||||
connectionFingerprint: binding?.appServerRuntimeFingerprint
|
||||
? fingerprint(
|
||||
JSON.stringify([
|
||||
binding.appServerRuntimeFingerprint,
|
||||
binding.connectionScope ?? null,
|
||||
binding.authProfileId ?? null,
|
||||
]),
|
||||
)
|
||||
: null,
|
||||
import: stored.nativeSubagentTaskImport
|
||||
? {
|
||||
version: stored.nativeSubagentTaskImport.version,
|
||||
taskIds: stored.nativeSubagentTaskImport.taskIds,
|
||||
}
|
||||
: undefined,
|
||||
assignments: (stored.nativeSubagentAssignments?.assignments ?? []).map((entry) => ({
|
||||
runId: entry.runId,
|
||||
childThreadId: entry.childThreadId,
|
||||
nativeParentThreadId: entry.nativeParentThreadId,
|
||||
nativeTurnId: entry.nativeTurnId,
|
||||
owner: {
|
||||
parentThreadId: entry.owner?.parentThreadId,
|
||||
sessionId: entry.owner?.sessionId,
|
||||
lifecycleRevision: entry.owner?.lifecycleRevision,
|
||||
connectionFingerprint: entry.owner?.connectionFingerprint,
|
||||
},
|
||||
recordedCompletion: entry.recordedCompletion
|
||||
? {
|
||||
status: entry.recordedCompletion.status,
|
||||
resultSha256: fingerprint(entry.recordedCompletion.result),
|
||||
hasResultMarker: entry.recordedCompletion.result.includes(RESULT_MARKER),
|
||||
}
|
||||
: undefined,
|
||||
})),
|
||||
}
|
||||
: null;
|
||||
} finally {
|
||||
db.close();
|
||||
}
|
||||
const sessions = new DatabaseSync(
|
||||
path.join(
|
||||
required("OPENCLAW_STATE_DIR"),
|
||||
"agents",
|
||||
"native-proof",
|
||||
"agent",
|
||||
"openclaw-agent.sqlite",
|
||||
),
|
||||
{ readOnly: true },
|
||||
);
|
||||
try {
|
||||
const row = sessions
|
||||
.prepare("SELECT current_session_id, entry_json FROM session_nodes WHERE session_key = ?")
|
||||
.get(SESSION_KEY);
|
||||
const entry = row ? JSON.parse(row.entry_json) : undefined;
|
||||
observation.session = row
|
||||
? {
|
||||
currentSessionId: row.current_session_id,
|
||||
sessionId: entry.sessionId,
|
||||
lifecycleRevision: entry.lifecycleRevision,
|
||||
agentHarnessId: entry.agentHarnessId,
|
||||
}
|
||||
: null;
|
||||
} finally {
|
||||
sessions.close();
|
||||
}
|
||||
} catch (error) {
|
||||
// Read failures remain explicit evidence without copying raw state or error text.
|
||||
observation.unavailable = { name: error.name, code: error.code };
|
||||
}
|
||||
writeJson(`native-assignment-inventory-${stage}.json`, observation);
|
||||
}
|
||||
|
||||
async function handoff() {
|
||||
setPhase("handoff");
|
||||
const seeded = nativeMessages().find(
|
||||
(entry) =>
|
||||
entry.phase === "seed-assignments" &&
|
||||
entry.method === "item/completed" &&
|
||||
entry.params?.threadId === "native-upgrade-parent" &&
|
||||
entry.params?.item?.tool === "spawnAgent",
|
||||
);
|
||||
assert(seeded?.params.turnId, "native handoff has no witnessed seed-parent turn");
|
||||
const socket = new WebSocket(readJson(artifact("native-assignment-ready.json")).url);
|
||||
const deadline = AbortSignal.timeout(30_000);
|
||||
let sequence = 0;
|
||||
const pending = new Map();
|
||||
let interruption;
|
||||
let failure;
|
||||
const fail = (error) => {
|
||||
failure ??= error;
|
||||
for (const request of pending.values()) {
|
||||
request.reject(error);
|
||||
}
|
||||
pending.clear();
|
||||
interruption?.reject(error);
|
||||
interruption = undefined;
|
||||
};
|
||||
socket.addEventListener("message", ({ data }) => {
|
||||
try {
|
||||
const message = JSON.parse(data);
|
||||
if (message.method) {
|
||||
if (
|
||||
interruption &&
|
||||
message.method === "turn/completed" &&
|
||||
message.params?.threadId === "native-upgrade-parent" &&
|
||||
message.params.turn?.id === interruption.turnId
|
||||
) {
|
||||
const observed = interruption;
|
||||
interruption = undefined;
|
||||
if (message.params.turn.status === "interrupted") {
|
||||
observed.resolve();
|
||||
} else {
|
||||
observed.reject(new Error("Native handoff did not interrupt the seed-parent turn"));
|
||||
}
|
||||
}
|
||||
return;
|
||||
}
|
||||
const request = pending.get(message.id);
|
||||
if (request) {
|
||||
pending.delete(message.id);
|
||||
if (message.error) {
|
||||
request.reject(new Error(message.error.message));
|
||||
} else {
|
||||
request.resolve(message.result);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
fail(error);
|
||||
}
|
||||
});
|
||||
socket.addEventListener("error", () => fail(new Error("Native handoff transport failed")));
|
||||
socket.addEventListener("close", () => fail(new Error("Native handoff transport closed")));
|
||||
const abort = () => fail(new Error("Native handoff timed out"));
|
||||
deadline.addEventListener("abort", abort, { once: true });
|
||||
const request = (method, params) =>
|
||||
new Promise((resolve, reject) => {
|
||||
deadline.throwIfAborted();
|
||||
if (failure) {
|
||||
throw failure;
|
||||
}
|
||||
const id = ++sequence;
|
||||
pending.set(id, { resolve, reject });
|
||||
socket.send(JSON.stringify({ id, method, params }));
|
||||
});
|
||||
try {
|
||||
await new Promise((resolve, reject) => {
|
||||
socket.addEventListener("open", resolve, { once: true });
|
||||
socket.addEventListener(
|
||||
"error",
|
||||
() => reject(new Error("Native handoff failed to connect")),
|
||||
{ once: true },
|
||||
);
|
||||
deadline.addEventListener(
|
||||
"abort",
|
||||
() => reject(new Error("Native handoff connect timed out")),
|
||||
{ once: true },
|
||||
);
|
||||
});
|
||||
await request("initialize", { clientInfo: { name: "native-upgrade-handoff", version: "1" } });
|
||||
socket.send(JSON.stringify({ method: "initialized", params: {} }));
|
||||
await request("thread/resume", { threadId: "native-upgrade-parent" });
|
||||
const parent = (
|
||||
await request("thread/read", { threadId: "native-upgrade-parent", includeTurns: true })
|
||||
).thread;
|
||||
assert.deepEqual(
|
||||
parent.turns.filter((turn) => turn.status === "inProgress").map((turn) => turn.id),
|
||||
[seeded.params.turnId],
|
||||
);
|
||||
const children = {};
|
||||
for (const threadId of ["native-upgrade-running", "native-upgrade-complete"]) {
|
||||
children[threadId] = await request("thread/read", { threadId, includeTurns: true });
|
||||
}
|
||||
// Codex acknowledges interruption before publishing turn/completed. Arm the
|
||||
// exact terminal witness before the request and join both protocol events.
|
||||
const interrupted = new Promise((resolve, reject) => {
|
||||
deadline.throwIfAborted();
|
||||
if (failure) {
|
||||
throw failure;
|
||||
}
|
||||
interruption = { turnId: seeded.params.turnId, resolve, reject };
|
||||
});
|
||||
await Promise.all([
|
||||
interrupted,
|
||||
request("turn/interrupt", {
|
||||
threadId: "native-upgrade-parent",
|
||||
turnId: seeded.params.turnId,
|
||||
}),
|
||||
]);
|
||||
const settled = (
|
||||
await request("thread/read", { threadId: "native-upgrade-parent", includeTurns: true })
|
||||
).thread;
|
||||
assert.equal(settled.status.type, "idle");
|
||||
assert(settled.turns.every((turn) => turn.status !== "inProgress"));
|
||||
for (const threadId of Object.keys(children)) {
|
||||
assert.deepEqual(
|
||||
await request("thread/read", { threadId, includeTurns: true }),
|
||||
children[threadId],
|
||||
"parent interruption changed a native child",
|
||||
);
|
||||
}
|
||||
assert.deepEqual(
|
||||
snapshot().tasks,
|
||||
readJson(artifact("native-assignment-baseline.json")).tasks,
|
||||
"native handoff changed published source rows",
|
||||
);
|
||||
// The first candidate Gateway may recover retained work before the explicit live probe.
|
||||
setPhase("recover");
|
||||
} finally {
|
||||
deadline.removeEventListener("abort", abort);
|
||||
socket.close();
|
||||
}
|
||||
}
|
||||
|
||||
function assignments(observation) {
|
||||
return observation.binding.value.nativeSubagentAssignments?.assignments ?? [];
|
||||
}
|
||||
|
|
@ -416,24 +659,34 @@ function assertCloseWitness(phase, childLoaded) {
|
|||
.some(
|
||||
(entry) =>
|
||||
entry.method === "turn/completed" &&
|
||||
entry.params?.threadId === "native-upgrade-parent" &&
|
||||
entry.params?.threadId === messages[started].params.threadId &&
|
||||
entry.params?.turn?.id === messages[started].params.turnId,
|
||||
),
|
||||
"native close parent completed before confirmation",
|
||||
);
|
||||
assert.deepEqual(response.result, {
|
||||
data: childLoaded
|
||||
? ["native-upgrade-parent", "native-upgrade-running"]
|
||||
: ["native-upgrade-parent"],
|
||||
nextCursor: null,
|
||||
});
|
||||
assert.equal(response.result.nextCursor, null, "native closure snapshot was incomplete");
|
||||
assert.equal(
|
||||
response.result.data.filter((id) => id === messages[started].params.threadId).length,
|
||||
1,
|
||||
"closing parent was not loaded exactly once",
|
||||
);
|
||||
assert.equal(
|
||||
response.result.data.filter((id) => id === "native-upgrade-running").length,
|
||||
childLoaded ? 1 : 0,
|
||||
);
|
||||
assert(
|
||||
!response.result.data.includes("native-upgrade-complete"),
|
||||
"finished child entered closure inventory",
|
||||
);
|
||||
}
|
||||
|
||||
async function live(candidateVersion) {
|
||||
captureInventory("before-recovery");
|
||||
const before = readJson(artifact("native-assignment-baseline.json"));
|
||||
const firstHop = readJson(artifact("native-assignment-first-hop.json"));
|
||||
assert.equal(firstHop.candidateVersion, candidateVersion);
|
||||
runParent("recover");
|
||||
captureInventory("after-recovery");
|
||||
// The foreground run can settle before its detached completion continuation. Join the
|
||||
// observable delivery and native terminal event before switching the backend's phase.
|
||||
await waitForWitness(() => {
|
||||
|
|
@ -449,7 +702,7 @@ async function live(candidateVersion) {
|
|||
messages.some(
|
||||
(entry) =>
|
||||
entry.method === "turn/completed" &&
|
||||
entry.params?.threadId === "native-upgrade-parent" &&
|
||||
entry.params?.threadId === delivered[0].params.threadId &&
|
||||
entry.params?.turn?.id === reply.result.turn.id,
|
||||
),
|
||||
"completion continuation did not settle",
|
||||
|
|
@ -545,8 +798,17 @@ if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href)
|
|||
);
|
||||
} else if (operation === "post-update") {
|
||||
postUpdate(candidateVersion);
|
||||
} else if (operation === "handoff") {
|
||||
await handoff();
|
||||
} else if (operation === "inventory") {
|
||||
assert.equal(candidateVersion, "after-first-hop");
|
||||
captureInventory(candidateVersion);
|
||||
} else if (operation === "live") {
|
||||
await live(candidateVersion);
|
||||
try {
|
||||
await live(candidateVersion);
|
||||
} finally {
|
||||
captureInventory("live-final");
|
||||
}
|
||||
} else {
|
||||
throw new Error(`Unknown native assignment operation: ${operation}`);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2544,6 +2544,9 @@ if [ "$SCENARIO" = "legacy-operator-state" ]; then
|
|||
fi
|
||||
phase patch-restored-index node scripts/e2e/lib/upgrade-survivor/legacy-operator-restored-index.mjs patch
|
||||
phase stop-restored-index-baseline stop_gateway
|
||||
if [ "$native_assignment_enabled" = "1" ]; then
|
||||
phase handoff-native-assignments node scripts/e2e/lib/upgrade-survivor/native-assignments.mjs handoff
|
||||
fi
|
||||
phase restore-baseline-index node scripts/e2e/lib/upgrade-survivor/legacy-operator-restored-index.mjs restore
|
||||
# Do not restart the published Gateway after restoring stale metadata over its current SQLite state.
|
||||
fi
|
||||
|
|
@ -2615,6 +2618,9 @@ if [ "$SCENARIO" = "legacy-operator-state" ]; then
|
|||
else
|
||||
phase legacy-operator-first-hop-cron probe_legacy_operator_migration
|
||||
fi
|
||||
if [ "$native_assignment_enabled" = "1" ]; then
|
||||
phase capture-native-assignment-first-hop-runtime node scripts/e2e/lib/upgrade-survivor/native-assignments.mjs inventory after-first-hop
|
||||
fi
|
||||
fi
|
||||
phase mobile-pairing-candidate-first verify_mobile_pairing_once \
|
||||
candidate-first "$MOBILE_PAIRING_CANDIDATE_FIRST_EVIDENCE"
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ prepare_diagnostics_capture() {
|
|||
"$ARTIFACT_DIR"/update-noop.{json,err} \
|
||||
"$ARTIFACT_DIR/candidate-cohort.json" \
|
||||
"$ARTIFACT_DIR"/native-assignment-{eligibility,baseline,first-hop,proof,ready,phase}.json \
|
||||
"$ARTIFACT_DIR"/native-assignment-inventory-{after-first-hop,before-recovery,after-recovery,live-final}.json \
|
||||
"$ARTIFACT_DIR/native-assignment-messages.jsonl" "$ARTIFACT_DIR/native-assignment-server.log" \
|
||||
"$ARTIFACT_DIR"/native-recover{,-wait}.{out,err} \
|
||||
"$ARTIFACT_DIR/backup-rollback.json" \
|
||||
|
|
|
|||
|
|
@ -968,6 +968,10 @@ it("does not reuse sibling or startup observations when an attempt fails before
|
|||
...turnLogs,
|
||||
...cronCliLogs,
|
||||
...nativeRecoveryLogs,
|
||||
"native-assignment-inventory-after-first-hop.json",
|
||||
"native-assignment-inventory-before-recovery.json",
|
||||
"native-assignment-inventory-after-recovery.json",
|
||||
"native-assignment-inventory-live-final.json",
|
||||
...baselineGatewayLogs,
|
||||
"sibling-refusal-update.json",
|
||||
"sibling-refusal-status.json",
|
||||
|
|
|
|||
323
test/scripts/upgrade-survivor-native-app-server.test.ts
Normal file
323
test/scripts/upgrade-survivor-native-app-server.test.ts
Normal file
|
|
@ -0,0 +1,323 @@
|
|||
import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
|
||||
import { once } from "node:events";
|
||||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
import { createInterface } from "node:readline";
|
||||
import { pathToFileURL } from "node:url";
|
||||
import { afterAll, beforeAll, expect, it } from "vitest";
|
||||
import { WebSocket } from "ws";
|
||||
import { rawDataToString } from "../../packages/gateway-client/src/websocket-data.js";
|
||||
import { useAutoCleanupTempDirTracker } from "../helpers/temp-dir.js";
|
||||
|
||||
type Turn = { id: string; status: string; items: unknown[] };
|
||||
type Thread = {
|
||||
id: string;
|
||||
parentThreadId?: string;
|
||||
source: unknown;
|
||||
status: { type: string };
|
||||
turns: Turn[];
|
||||
};
|
||||
type Result = { thread?: Thread; turn?: Turn; data?: string[]; nextCursor?: string | null };
|
||||
type Notification = {
|
||||
method: string;
|
||||
params: {
|
||||
threadId?: string;
|
||||
turnId?: string;
|
||||
turn?: Turn;
|
||||
item?: { tool?: string; status?: string };
|
||||
};
|
||||
};
|
||||
|
||||
let child: ChildProcessWithoutNullStreams;
|
||||
let socket: WebSocket;
|
||||
let phaseFile: string;
|
||||
let logFile: string;
|
||||
let sequence = 0;
|
||||
const pending = new Map<
|
||||
number,
|
||||
{ resolve: (value: Result) => void; reject: (error: Error) => void }
|
||||
>();
|
||||
const notifications: Notification[] = [];
|
||||
const waiting = new Set<() => void>();
|
||||
const dirs = useAutoCleanupTempDirTracker((cleanup) =>
|
||||
afterAll(async () => {
|
||||
socket?.terminate();
|
||||
if (child && child.exitCode === null && child.signalCode === null) {
|
||||
const exited = once(child, "exit");
|
||||
child.kill("SIGTERM");
|
||||
await exited;
|
||||
}
|
||||
cleanup();
|
||||
}),
|
||||
);
|
||||
|
||||
function request(method: string, params: object = {}): Promise<Result> {
|
||||
const id = ++sequence;
|
||||
return new Promise((resolve, reject) => {
|
||||
pending.set(id, { resolve, reject });
|
||||
socket.send(JSON.stringify({ id, method, params }));
|
||||
});
|
||||
}
|
||||
|
||||
async function notificationWhere(matches: (entry: Notification) => boolean): Promise<Notification> {
|
||||
const observed = notifications.find(matches);
|
||||
if (observed) {
|
||||
return observed;
|
||||
}
|
||||
return await new Promise<Notification>((resolve) => {
|
||||
const check = () => {
|
||||
const entry = notifications.find(matches);
|
||||
if (entry) {
|
||||
waiting.delete(check);
|
||||
resolve(entry);
|
||||
}
|
||||
};
|
||||
waiting.add(check);
|
||||
});
|
||||
}
|
||||
|
||||
async function completed(threadId: string, turnId: string) {
|
||||
return await notificationWhere(
|
||||
(entry) =>
|
||||
entry.method === "turn/completed" &&
|
||||
entry.params.threadId === threadId &&
|
||||
entry.params.turn?.id === turnId,
|
||||
);
|
||||
}
|
||||
|
||||
beforeAll(async () => {
|
||||
const root = dirs.make("native-upgrade-protocol-");
|
||||
phaseFile = path.join(root, "phase.json");
|
||||
logFile = path.join(root, "messages.jsonl");
|
||||
const readyFile = path.join(root, "ready.json");
|
||||
const driver = path.join(root, "driver.mjs");
|
||||
const fixture = pathToFileURL(
|
||||
path.resolve("scripts/e2e/lib/upgrade-survivor/native-assignment-app-server.mjs"),
|
||||
);
|
||||
fs.writeFileSync(
|
||||
driver,
|
||||
[
|
||||
'import fs from "node:fs";',
|
||||
`await import(${JSON.stringify(fixture.href)});`,
|
||||
`process.stdout.write(fs.readFileSync(${JSON.stringify(readyFile)}, "utf8"));`,
|
||||
].join("\n"),
|
||||
);
|
||||
fs.writeFileSync(phaseFile, JSON.stringify({ phase: "seed-assignments" }));
|
||||
child = spawn(
|
||||
process.execPath,
|
||||
[
|
||||
driver,
|
||||
"--package-root",
|
||||
process.cwd(),
|
||||
"--ready-file",
|
||||
readyFile,
|
||||
"--phase-file",
|
||||
phaseFile,
|
||||
"--log-file",
|
||||
logFile,
|
||||
],
|
||||
{ stdio: "pipe" },
|
||||
);
|
||||
const lines = createInterface({ input: child.stdout });
|
||||
const [line] = await Promise.race([
|
||||
once(lines, "line"),
|
||||
once(child, "exit").then(([code]) => {
|
||||
throw new Error(`Native fixture exited before ready: ${code}`);
|
||||
}),
|
||||
]);
|
||||
lines.close();
|
||||
const ready = JSON.parse(line);
|
||||
socket = new WebSocket(ready.url);
|
||||
socket.on("message", (data) => {
|
||||
const message = JSON.parse(rawDataToString(data));
|
||||
if (message.method) {
|
||||
notifications.push(message);
|
||||
for (const check of waiting) {
|
||||
check();
|
||||
}
|
||||
return;
|
||||
}
|
||||
const response = pending.get(message.id);
|
||||
pending.delete(message.id);
|
||||
if (message.error) {
|
||||
response?.reject(new Error(message.error.message));
|
||||
} else {
|
||||
response?.resolve(message.result);
|
||||
}
|
||||
});
|
||||
await once(socket, "open");
|
||||
await request("initialize", { clientInfo: { name: "native-fixture-regression", version: "1" } });
|
||||
socket.send(JSON.stringify({ method: "initialized", params: {} }));
|
||||
});
|
||||
|
||||
it("isolates native seeding, interruption, and closure from incidental parent turns", async () => {
|
||||
const parent = (await request("thread/start", { cwd: process.cwd() })).thread!;
|
||||
expect(parent.id).toBe("native-upgrade-parent");
|
||||
const seeded = (
|
||||
await request("turn/start", {
|
||||
threadId: parent.id,
|
||||
input: [{ type: "text", text: "NATIVE_UPGRADE_SEED_ASSIGNMENTS" }],
|
||||
})
|
||||
).turn!;
|
||||
await completed("native-upgrade-complete", "native-complete-turn");
|
||||
const before = await request("thread/read", { threadId: parent.id, includeTurns: true });
|
||||
const running = await request("thread/read", {
|
||||
threadId: "native-upgrade-running",
|
||||
includeTurns: true,
|
||||
});
|
||||
const finished = await request("thread/read", {
|
||||
threadId: "native-upgrade-complete",
|
||||
includeTurns: true,
|
||||
});
|
||||
expect(running.thread?.status.type).toBe("active");
|
||||
expect(running.thread?.turns).toMatchObject([
|
||||
{ id: "native-running-turn", status: "inProgress" },
|
||||
]);
|
||||
expect(finished.thread?.turns).toMatchObject([
|
||||
{
|
||||
id: "native-complete-turn",
|
||||
status: "completed",
|
||||
items: [{ type: "agentMessage", text: "NATIVE_UPGRADE_PENDING_RESULT" }],
|
||||
},
|
||||
]);
|
||||
|
||||
const incidental = (await request("thread/start", { cwd: process.cwd() })).thread!;
|
||||
expect(incidental.id).not.toBe(parent.id);
|
||||
expect(incidental.parentThreadId).toBeUndefined();
|
||||
expect(incidental.source).toBe("unknown");
|
||||
expect(incidental.turns).toEqual([]);
|
||||
const incidentalTurn = (
|
||||
await request("turn/start", {
|
||||
threadId: incidental.id,
|
||||
input: [{ type: "text", text: "An unrelated turn while seeding is still active." }],
|
||||
})
|
||||
).turn!;
|
||||
await completed(incidental.id, incidentalTurn.id);
|
||||
expect(await request("thread/read", { threadId: parent.id, includeTurns: true })).toEqual(before);
|
||||
|
||||
fs.writeFileSync(phaseFile, JSON.stringify({ phase: "handoff" }));
|
||||
await request("turn/interrupt", { threadId: parent.id, turnId: seeded.id });
|
||||
await completed(parent.id, seeded.id);
|
||||
const interrupted = (await request("thread/read", { threadId: parent.id, includeTurns: true }))
|
||||
.thread!;
|
||||
expect(interrupted.status.type).toBe("idle");
|
||||
expect(interrupted.turns.find((turn) => turn.id === seeded.id)?.status).toBe("interrupted");
|
||||
expect(interrupted.turns.some((turn) => turn.status === "inProgress")).toBe(false);
|
||||
expect(
|
||||
await request("thread/read", { threadId: "native-upgrade-running", includeTurns: true }),
|
||||
).toEqual(running);
|
||||
expect(
|
||||
await request("thread/read", { threadId: "native-upgrade-complete", includeTurns: true }),
|
||||
).toEqual(finished);
|
||||
const messages = fs
|
||||
.readFileSync(logFile, "utf8")
|
||||
.trim()
|
||||
.split("\n")
|
||||
.map((line) => JSON.parse(line));
|
||||
expect(messages.filter((entry) => entry.direction === "fixture-error")).toEqual([]);
|
||||
const interruptRequest = messages.find(
|
||||
(entry) => entry.direction === "request" && entry.method === "turn/interrupt",
|
||||
);
|
||||
const interruptResponse = messages.findIndex(
|
||||
(entry) => entry.direction === "response" && entry.id === interruptRequest.id,
|
||||
);
|
||||
const interruptNotification = messages.findIndex(
|
||||
(entry) =>
|
||||
entry.method === "turn/completed" &&
|
||||
entry.params?.threadId === parent.id &&
|
||||
entry.params?.turn?.id === seeded.id,
|
||||
);
|
||||
expect(interruptResponse).toBeGreaterThanOrEqual(0);
|
||||
expect(interruptNotification).toBeGreaterThan(interruptResponse);
|
||||
|
||||
for (const phase of ["close-loaded", "close-gone"]) {
|
||||
fs.writeFileSync(phaseFile, JSON.stringify({ phase }));
|
||||
const prompt = `Continue the synthetic native upgrade fixture: ${phase}.`;
|
||||
const closing = (
|
||||
await request("turn/start", {
|
||||
threadId: parent.id,
|
||||
input: [{ type: "text", text: prompt }],
|
||||
})
|
||||
).turn!;
|
||||
await notificationWhere(
|
||||
(entry) =>
|
||||
entry.method === "item/completed" &&
|
||||
entry.params.threadId === parent.id &&
|
||||
entry.params.turnId === closing.id &&
|
||||
entry.params.item?.tool === "closeAgent" &&
|
||||
entry.params.item.status === "completed",
|
||||
);
|
||||
|
||||
const incidentalClosePhase = (
|
||||
await request("turn/start", {
|
||||
threadId: incidental.id,
|
||||
input: [{ type: "text", text: "An unrelated turn with no close request." }],
|
||||
})
|
||||
).turn!;
|
||||
const incidentalOutcome = await notificationWhere(
|
||||
(entry) =>
|
||||
entry.params.threadId === incidental.id &&
|
||||
((entry.method === "turn/completed" && entry.params.turn?.id === incidentalClosePhase.id) ||
|
||||
(entry.method === "item/completed" &&
|
||||
entry.params.turnId === incidentalClosePhase.id &&
|
||||
entry.params.item?.tool === "closeAgent")),
|
||||
);
|
||||
expect(incidentalOutcome.method, `${phase}: an incidental turn must not request closure`).toBe(
|
||||
"turn/completed",
|
||||
);
|
||||
expect(incidentalOutcome.params.turn?.status).toBe("completed");
|
||||
|
||||
const refused = (
|
||||
await request("turn/start", {
|
||||
threadId: incidental.id,
|
||||
input: [{ type: "text", text: prompt }],
|
||||
})
|
||||
).turn!;
|
||||
expect((await completed(incidental.id, refused.id)).params.turn?.status).toBe("failed");
|
||||
expect(
|
||||
notifications.some(
|
||||
(entry) =>
|
||||
entry.params.threadId === incidental.id &&
|
||||
entry.params.turnId === refused.id &&
|
||||
entry.params.item?.tool === "closeAgent",
|
||||
),
|
||||
).toBe(false);
|
||||
|
||||
const waitingParent = (
|
||||
await request("thread/read", { threadId: parent.id, includeTurns: true })
|
||||
).thread!;
|
||||
expect(waitingParent.turns.find((turn) => turn.id === closing.id)?.status).toBe("inProgress");
|
||||
expect(
|
||||
(await request("thread/read", { threadId: "native-upgrade-running" })).thread?.status.type,
|
||||
).toBe("active");
|
||||
|
||||
const loaded = await request("thread/loaded/list");
|
||||
expect(loaded.nextCursor).toBeNull();
|
||||
expect(loaded.data?.filter((id) => id === parent.id)).toHaveLength(1);
|
||||
expect(loaded.data?.filter((id) => id === "native-upgrade-running")).toHaveLength(
|
||||
phase === "close-loaded" ? 1 : 0,
|
||||
);
|
||||
expect(loaded.data).not.toContain("native-upgrade-complete");
|
||||
expect((await completed(parent.id, closing.id)).params.turn?.status).toBe("completed");
|
||||
const confirmed = (await request("thread/read", { threadId: parent.id, includeTurns: true }))
|
||||
.thread!;
|
||||
expect(confirmed.turns.find((turn) => turn.id === closing.id)?.items).toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({ type: "agentMessage", text: "NATIVE_UPGRADE_CLOSE_OBSERVED" }),
|
||||
]),
|
||||
);
|
||||
}
|
||||
const finalMessages = fs
|
||||
.readFileSync(logFile, "utf8")
|
||||
.trim()
|
||||
.split("\n")
|
||||
.map((line) => JSON.parse(line));
|
||||
expect(
|
||||
finalMessages
|
||||
.filter((entry) => entry.direction === "fixture-error")
|
||||
.map((entry) => entry.message),
|
||||
).toEqual([
|
||||
"Native close confirmation is already pending on this connection",
|
||||
"Native close confirmation is already pending on this connection",
|
||||
]);
|
||||
});
|
||||
|
|
@ -154,6 +154,7 @@ export const databaseWorkerExtensionTestFiles = [
|
|||
"extensions/codex/src/app-server/event-projector.output-fidelity.native.test.ts",
|
||||
"extensions/codex/src/app-server/inference-proxy.native.test.ts",
|
||||
"extensions/codex/src/app-server/settled-turn-finalizer.native.test.ts",
|
||||
"extensions/codex/src/app-server/thread-lifecycle.assignment-rotation.test.ts",
|
||||
"extensions/codex/src/app-server/thread-lifecycle.native.test.ts",
|
||||
"extensions/codex/src/app-server/thread-lifecycle.binding.test.ts",
|
||||
"extensions/codex/src/app-server/thread-lifecycle.native-config.test.ts",
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue