ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-27 04:47:03 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(codex): restore safe async replies and managed approvals (#127714)

Browse Source 
* fix(codex): secure and deliver asynchronous session replies

* fix(codex): honor managed untrusted approval requirements

* test(codex): keep async transcript fixtures fully typed
This commit is contained in:
Peter Steinberger
2026-08-21 17:47:39 -07:00
committed by GitHub
parent 7987a33bb7
commit a554ef50c1
23 changed files with 530 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
extensions/codex/src/app-server/config-contracts.ts
+3 -2
View File
@@ -29,14 +29,15 @@ export type OpenClawExecPolicy = OpenClawExecPolicyForCodexAppServer;
export type ProviderAuthAliasConfig = NonNullable<ProviderAuthAliasLookupParams>["config"];
export type CodexAppServerDefaultPolicy = {
mode: CodexAppServerPolicyMode;
approvalPolicy?: CodexAppServerApprovalPolicy;
approvalPolicy?: CodexAppServerManagedApprovalPolicy;
approvalsReviewer?: CodexAppServerApprovalsReviewer;
sandbox?: CodexAppServerSandboxMode;
dangerFullAccessAllowed?: boolean;
};
export type CodexAppServerApprovalPolicy = "never" | "on-request";
export type CodexAppServerManagedApprovalPolicy = Extract<CodexApprovalPolicy, string>;
export type CodexAppServerApprovalPolicySource = "config" | "env" | "requirements" | "implicit";
export type CodexAppServerEffectiveApprovalPolicy = Exclude<CodexApprovalPolicy, "untrusted">;
export type CodexAppServerEffectiveApprovalPolicy = CodexApprovalPolicy;
export type CodexAppServerSandboxMode = "read-only" | "workspace-write" | "danger-full-access";
export type CodexAppServerApprovalsReviewer = "user" | "auto_review" | "guardian_subagent";
export type CodexAppServerCommandSource = "managed" | "resolved-managed" | "config" | "env";
extensions/codex/src/app-server/config-requirements.ts
+10 -15
View File
@@ -1,7 +1,7 @@
import { readFileSync } from "node:fs";
import type {
CodexAppServerApprovalPolicy,
CodexAppServerApprovalsReviewer,
CodexAppServerManagedApprovalPolicy,
CodexAppServerSandboxMode,
OpenClawExecMode,
} from "./config-contracts.js";
@@ -59,22 +59,14 @@ export function parseAllowedSandboxModesFromCodexRequirements(
export function parseAllowedApprovalPoliciesFromCodexRequirements(
content: string,
): Set<CodexAppServerApprovalPolicy> | undefined {
): Set<CodexAppServerManagedApprovalPolicy> | undefined {
const values = parseTopLevelRequirementsStringArray(content, "allowed_approval_policies");
if (values === undefined) {
return undefined;
}
const normalizedPolicies = values
.map((entry) => normalizeRequirementsApprovalPolicy(entry))
.filter((entry): entry is CodexAppServerApprovalPolicy => entry !== undefined);
if (
normalizedPolicies.length === 0 &&
values.some((entry) => entry.trim().toLowerCase() === "untrusted")
) {
throw new Error(
'Codex requirements allowed_approval_policies only permits retired "untrusted"; replace it with "on-request".',
);
}
.filter((entry): entry is CodexAppServerManagedApprovalPolicy => entry !== undefined);
return normalizedPolicies.length > 0 ? new Set(normalizedPolicies) : undefined;
}
@@ -303,7 +295,7 @@ function globPatternMatches(value: string, pattern: string): boolean {
function normalizeRequirementsApprovalPolicy(
value: string,
): CodexAppServerApprovalPolicy | undefined {
): CodexAppServerManagedApprovalPolicy | undefined {
const normalized = value.trim().toLowerCase();
// Codex still accepts this alias in persisted requirements, while its
// app-server exposes only the canonical on-request value.
@@ -311,7 +303,7 @@ function normalizeRequirementsApprovalPolicy(
return "on-request";
}
if (normalized === "untrusted") {
return undefined;
return normalized;
}
return resolveApprovalPolicy(normalized);
}
@@ -324,12 +316,15 @@ function normalizeRequirementsApprovalsReviewer(
}
export function selectGuardianApprovalPolicy(
allowedApprovalPolicies: Set<CodexAppServerApprovalPolicy> | undefined,
allowedApprovalPolicies: Set<CodexAppServerManagedApprovalPolicy> | undefined,
execModeRequiringPromptingApprovals?: Extract<OpenClawExecMode, "auto" | "ask">,
): CodexAppServerApprovalPolicy {
): CodexAppServerManagedApprovalPolicy {
if (allowedApprovalPolicies === undefined || allowedApprovalPolicies.has("on-request")) {
return "on-request";
}
if (allowedApprovalPolicies.has("untrusted")) {
return "untrusted";
}
if (execModeRequiringPromptingApprovals) {
throw new Error(
`tools.exec.mode=${execModeRequiringPromptingApprovals} requires Codex app-server prompting approvals`,
extensions/codex/src/app-server/config-security.ts
+5 -1
View File
@@ -138,6 +138,9 @@ function stableStringifyJson(value: JsonValue): string {
export function withMcpElicitationsApprovalPolicy(
policy: CodexAppServerEffectiveApprovalPolicy,
): CodexAppServerEffectiveApprovalPolicy {
if (policy === "untrusted") {
return policy;
}
if (typeof policy !== "string") {
return {
granular: {
@@ -294,7 +297,8 @@ export function resolveDefaultCodexAppServerPolicy(params: {
const yoloSandboxAllowed =
allowedSandboxModes === undefined || allowedSandboxModes.has("danger-full-access");
const yoloApprovalAllowed =
allowedApprovalPolicies === undefined || allowedApprovalPolicies.has("never");
allowedApprovalPolicies === undefined ||
(allowedApprovalPolicies.has("never") && !allowedApprovalPolicies.has("untrusted"));
const yoloReviewerAllowed =
allowedApprovalsReviewers === undefined || allowedApprovalsReviewers.has("user");
if (!params.forceGuardian && yoloSandboxAllowed && yoloApprovalAllowed && yoloReviewerAllowed) {
extensions/codex/src/app-server/config.test.ts
+43 -12
View File
@@ -31,6 +31,10 @@ function resolveRuntimeForTest(params: RuntimeOptionsParams = {}) {
}
describe("withMcpElicitationsApprovalPolicy", () => {
it("preserves managed per-command approvals that already allow MCP elicitation", () => {
expect(withMcpElicitationsApprovalPolicy("untrusted")).toBe("untrusted");
});
it("returns every field required by Codex granular approval policy", () => {
expect(withMcpElicitationsApprovalPolicy("never")).toEqual({
granular: {
@@ -1232,6 +1236,27 @@ describe("Codex app-server config", () => {
});
});
it.each([
{ policies: ["untrusted"], description: "only the managed internal policy" },
{ policies: ["untrusted", "never"], description: "managed and unrestricted policies" },
])("preserves $description without weakening Codex approvals", ({ policies }) => {
const runtime = resolveRuntimeForTest({
pluginConfig: {},
modelProvider: "openai",
requirementsToml: `allowed_approval_policies = [${policies
.map((policy) => `"${policy}"`)
.join(", ")}]\n`,
});
expectRuntimePolicy(runtime, {
approvalPolicy: "untrusted",
sandbox: "workspace-write",
approvalsReviewer: "auto_review",
});
expect(runtime.approvalPolicySource).toBe("requirements");
expect(withMcpElicitationsApprovalPolicy(runtime.approvalPolicy)).toBe("untrusted");
});
it("normalizes the deprecated requirements on-failure alias to on-request", () => {
const runtime = resolveRuntimeForTest({
pluginConfig: {},
@@ -2452,23 +2477,11 @@ allowed_sandbox_modes = ["read-only", "workspace-write"]
policies: ["never"],
error: "tools.exec.mode=auto requires Codex app-server prompting approvals",
},
{
execMode: "auto",
policies: ["untrusted"],
error:
'Codex requirements allowed_approval_policies only permits retired "untrusted"; replace it with "on-request".',
},
{
execMode: "ask",
policies: ["never"],
error: "tools.exec.mode=ask requires Codex app-server prompting approvals",
},
{
execMode: "ask",
policies: ["untrusted"],
error:
'Codex requirements allowed_approval_policies only permits retired "untrusted"; replace it with "on-request".',
},
] as const)(
"fails closed when normalized OpenClaw $execMode mode can only use $policies approvals",
({ execMode, policies, error }) => {
@@ -2484,6 +2497,24 @@ allowed_sandbox_modes = ["read-only", "workspace-write"]
},
);
it.each([
{ execMode: "auto" as const, approvalsReviewer: "auto_review" },
{ execMode: "ask" as const, approvalsReviewer: "user" },
])("honors managed prompting approvals for OpenClaw $execMode mode", (expected) => {
const runtime = resolveRuntimeForTest({
pluginConfig: {},
execMode: expected.execMode,
modelProvider: "openai",
requirementsToml: 'allowed_approval_policies = ["untrusted", "never"]\n',
});
expectRuntimePolicy(runtime, {
approvalPolicy: "untrusted",
sandbox: "workspace-write",
approvalsReviewer: expected.approvalsReviewer,
});
});
it("keeps normalized OpenClaw full exec mode on default Codex yolo", () => {
const runtime = resolveRuntimeForTest({
pluginConfig: {},
extensions/codex/src/app-server/event-projector-options.ts
+1
View File
@@ -9,6 +9,7 @@ export type CodexAsyncDeliverySettlement = "settled" | "retry";
export type CodexAppServerEventProjectorOptions = {
initialContextTokens?: number;
nativePostToolUseRelayEnabled?: boolean;
asyncUserMessageAllowed?: boolean;
onAsyncDelivery?: (delivery: {
itemId: string;
message: AssistantMessage;
extensions/codex/src/app-server/event-projector.async-delivery.test.ts
+39 -2
View File
@@ -16,6 +16,44 @@ import { codexTranscriptMirrorRuntime } from "./transcript-mirror.js";
registerCodexEventProjectorTestLifecycle();
describe("CodexAppServerEventProjector async delivery", () => {
it.each([
{ name: "disabled tools", disableTools: true },
{ name: "an empty tool allowlist", toolsAllow: [] },
{ name: "the ring-zero system tool", toolsAllow: ["openclaw"] },
{ name: "an allowlist without message delivery", toolsAllow: ["read"] },
])("does not expose native async messages through $name", async (restriction) => {
const params = await createParams();
const onAsyncDelivery = vi.fn().mockResolvedValue("settled");
const projector = await createProjector({ ...params, ...restriction }, { onAsyncDelivery });
await projector.handleNotification(
forCurrentTurn("item/completed", {
item: {
type: "agentMessage",
id: "unauthorized-async-update",
phase: "final_answer",
delivery: "async",
text: "This restricted update must not reach a user.",
},
}),
);
await projector.handleNotification(
turnCompleted([
{
type: "agentMessage",
id: "authorized-final",
phase: "final_answer",
text: "Ordinary final reply.",
},
]),
);
expect(onAsyncDelivery).not.toHaveBeenCalled();
expect(
JSON.stringify(projector.buildResult(buildEmptyToolTelemetry()).messagesSnapshot),
).not.toContain("restricted update");
});
it("persists async delivery once without selecting it as the final answer", async () => {
const onAgentEvent = vi.fn();
const onBlockReply = vi.fn();
@@ -195,7 +233,7 @@ describe("CodexAppServerEventProjector async delivery", () => {
]);
});
it("retries unsettled sessionless async delivery from the terminal snapshot", async () => {
it("retries unsettled sessionless async delivery when the real terminal summary contains only the final answer", async () => {
const params = await createParams();
const onBlockReply = vi
.fn()
@@ -222,7 +260,6 @@ describe("CodexAppServerEventProjector async delivery", () => {
text: "Retry this background update.",
};
const completed = turnCompleted([
asyncItem,
{
type: "agentMessage",
id: "terminal-retry",
extensions/codex/src/app-server/event-projector.ts
+45
View File
@@ -66,6 +66,10 @@ export class CodexAppServerEventProjector {
private readonly activeItemIds = new Set<string>();
private readonly completedItemIds = new Set<string>();
private readonly settledAsyncDeliveryItemIds = new Set<string>();
private readonly pendingAsyncDeliveries = new Map<
string,
Parameters<NonNullable<CodexAppServerEventProjectorOptions["onAsyncDelivery"]>>[0]
>();
private readonly activeCompactionItemIds = new Set<string>();
private readonly terminalPresentationClearedItemIds = new Set<string>();
private readonly nativeToolOutcomeOrdinals = new Map<string, number>();
@@ -452,6 +456,18 @@ export class CodexAppServerEventProjector {
this.activeItemIds.delete(itemId);
this.completedItemIds.add(itemId);
}
if (
item?.type === "agentMessage" &&
item.delivery === "async" &&
!this.canDeliverAsyncUserMessage()
) {
embeddedAgentLog.warn("blocked unauthorized codex async user message", {
itemId,
threadId: this.threadId,
turnId: this.turnId,
});
return;
}
const asyncMessage = this.assistantProjection.recordItemCompleted(
item,
itemId,
@@ -530,6 +546,9 @@ export class CodexAppServerEventProjector {
this.promptErrorSource = "prompt";
}
const turnItems = turn.items ?? [];
// Upstream terminal summaries contain only the last assistant item. Keep
// earlier unsettled deliveries at their producer instead of inferring them.
const unsettledAsyncDeliveries = [...this.pendingAsyncDeliveries.values()];
// The final snapshot is authoritative when item notifications were omitted.
// Only its last relevant tool may change the terminal presentation.
for (let index = turnItems.length - 1; index >= 0; index -= 1) {
@@ -546,6 +565,13 @@ export class CodexAppServerEventProjector {
}
}
for (const item of turnItems) {
if (
item.type === "agentMessage" &&
item.delivery === "async" &&
!this.canDeliverAsyncUserMessage()
) {
continue;
}
this.diagnostics.warnUnknownItemStatus(item);
const asyncMessage = this.assistantProjection.recordSnapshotItem(item);
if (asyncMessage) {
@@ -562,6 +588,9 @@ export class CodexAppServerEventProjector {
this.toolProgressProjection.emitToolResultSummary(item);
this.toolProgressProjection.emitToolResultOutput(item);
}
for (const delivery of unsettledAsyncDeliveries) {
await this.deliverAsyncMessage(delivery);
}
this.assistantProjection.finalizeAnswerCandidate(turn);
this.activeCompactionItemIds.clear();
await this.reasoningProjection.maybeEndReasoning();
@@ -580,9 +609,25 @@ export class CodexAppServerEventProjector {
const settlement = await this.options.onAsyncDelivery?.(delivery);
if (settlement === "settled") {
this.settledAsyncDeliveryItemIds.add(delivery.itemId);
this.pendingAsyncDeliveries.delete(delivery.itemId);
} else if (settlement === "retry") {
this.pendingAsyncDeliveries.set(delivery.itemId, delivery);
}
}
private canDeliverAsyncUserMessage(): boolean {
if (this.params.disableTools === true) {
return false;
}
if (this.options.asyncUserMessageAllowed !== undefined) {
return this.options.asyncUserMessageAllowed;
}
return (
this.params.toolsAllow === undefined ||
this.params.toolsAllow.some((name) => name === "*" || name === "message")
);
}
private async emitSnapshotOnlyNativeToolProgress(item: CodexThreadItem): Promise<void> {
if (
!shouldSynthesizeToolProgressForItem(item) ||
extensions/codex/src/app-server/run-attempt-active-turn.ts
+17 -11
View File
@@ -103,6 +103,10 @@ export async function activateCodexAttemptTurn(
nativePostToolUseRelayEnabled:
resourceState.nativeHookRelay?.allowedEvents.includes("post_tool_use") === true &&
resourceState.nativeHookRelay.shouldRelayEvent("post_tool_use"),
asyncUserMessageAllowed:
params.disableTools !== true &&
(params.toolsAllow === undefined ||
toolBridge.availableTools.some((tool) => tool.name === "message")),
onAsyncDelivery: async (delivery) => {
return await codexTranscriptMirrorRuntime.deliverAsyncMessageBestEffort({
params: dynamicToolParams,
@@ -189,6 +193,19 @@ export async function activateCodexAttemptTurn(
for (const failure of pendingNativePreToolUseFailures.splice(0)) {
activeProjector.recordNativeToolPreToolUseFailure(failure);
}
const notifyUserMessagePersisted = createCodexAppServerUserMessagePersistenceNotifier(params);
// Buffered async items can persist immediately when the route opens. Commit
// their owning user admission first so durable history stays chronological.
await mirrorPromptAtTurnStartBestEffort({
params,
agentId: sessionAgentId,
notifyUserMessagePersisted,
sessionKey: sandboxSessionKey,
cwd: effectiveCwd,
threadId: resourceState.thread.threadId,
turnId: activeTurnId,
upstreamUserText: turnState.codexTurnPromptText,
});
// The route buffers early events. Publish full turn context, then release in wire order.
if (resourceState.turnRoute) {
try {
@@ -319,17 +336,6 @@ export async function activateCodexAttemptTurn(
terminalState.terminalOutcomeFrozen = true;
params.abortSignal?.removeEventListener("abort", abortFromUpstream);
};
const notifyUserMessagePersisted = createCodexAppServerUserMessagePersistenceNotifier(params);
void mirrorPromptAtTurnStartBestEffort({
params,
agentId: sessionAgentId,
notifyUserMessagePersisted,
sessionKey: sandboxSessionKey,
cwd: effectiveCwd,
threadId: resourceState.thread.threadId,
turnId: activeTurnId,
upstreamUserText: turnState.codexTurnPromptText,
});
const abortListener = () => {
if (state.timedOut) {
void (async () => {
extensions/codex/src/app-server/run-attempt.context-engine.test.ts
+53
View File
@@ -15,6 +15,7 @@ import { MESSAGE_TOOL_DELIVERY_HINTS } from "openclaw/plugin-sdk/message-tool-de
import { createMockPluginRegistry } from "openclaw/plugin-sdk/plugin-test-runtime";
import { registerSandboxBackend } from "openclaw/plugin-sdk/sandbox";
import { upsertSessionEntry } from "openclaw/plugin-sdk/session-store-runtime";
import { readSessionTranscriptEvents } from "openclaw/plugin-sdk/session-transcript-runtime";
import { formatSqliteSessionFileMarker } from "openclaw/plugin-sdk/sqlite-runtime-testing";
import { readStringValue } from "openclaw/plugin-sdk/string-coerce-runtime";
// Codex tests cover run attempt.context engine plugin behavior.
@@ -1867,6 +1868,58 @@ describe("runCodexAppServerAttempt context-engine lifecycle", () => {
});
});
it("persists the admitted user prompt before an async item buffered during turn startup", async () => {
const workspaceDir = path.join(tempDir, "workspace-early-async");
const params = await createSqliteParams(workspaceDir, "early-async-order");
params.onBlockReply = vi.fn();
const recorder = params.userTurnTranscriptRecorder;
if (!recorder) {
throw new Error("expected user turn transcript recorder");
}
recorder.markRuntimePersistencePending = vi.fn();
const harness = createStartedThreadHarness(async (method) => {
if (method === "turn/start") {
await harness.notify({
method: "item/completed",
params: {
threadId: "thread-1",
turnId: "turn-1",
item: {
type: "agentMessage",
id: "startup-async",
phase: "final_answer",
delivery: "async",
text: "Working on the request.",
},
},
});
}
return undefined;
});
const run = runCodexAppServerAttempt(params);
await harness.waitForMethod("turn/start");
await vi.waitFor(() => expect(params.onBlockReply).toHaveBeenCalledOnce());
await harness.completeTurn();
await run;
const sessionTarget = params.sessionTarget;
if (!sessionTarget?.sessionId || !sessionTarget.sessionKey) {
throw new Error("expected a complete session transcript target");
}
const messages = (
await readSessionTranscriptEvents({
...sessionTarget,
sessionId: sessionTarget.sessionId,
sessionKey: sessionTarget.sessionKey,
})
)
.map((event) => (event as { message?: { role?: string } }).message)
.filter((message) => message !== undefined);
expect(messages.slice(0, 2).map((message) => message.role)).toEqual(["user", "assistant"]);
expect(messages[1]).toMatchObject({ openclawAsyncDelivery: { itemId: "startup-async" } });
});
it("reloads mirrored history after bootstrap mutates the session transcript", async () => {
const sessionFile = path.join(tempDir, "session.jsonl");
const workspaceDir = path.join(tempDir, "workspace");
extensions/codex/src/app-server/session-binding.test.ts
+2 -1
View File
@@ -76,7 +76,7 @@ describe("Codex app-server binding store", () => {
});
});
it("does not normalize untrusted persisted binding policy into live runtime policy", () => {
it("preserves the effective managed approval policy in persisted thread bindings", () => {
expect(
readCodexAppServerThreadBinding({
threadId: "thread-untrusted-policy",
@@ -87,6 +87,7 @@ describe("Codex app-server binding store", () => {
).toEqual({
threadId: "thread-untrusted-policy",
cwd: "/repo",
approvalPolicy: "untrusted",
sandbox: "workspace-write",
});
});
extensions/codex/src/app-server/session-binding.ts
+1 -1
View File
@@ -227,7 +227,7 @@ const threadBindingSchema = z
approvalPolicy: z
.preprocess(
(value) => (value === "on-failure" ? "on-request" : value),
z.enum(["never", "on-request"]).optional(),
z.enum(["never", "on-request", "untrusted"]).optional(),
)
.catch(undefined),
sandbox: z
extensions/codex/src/app-server/session-permission-policy.test.ts
+30
View File
@@ -98,6 +98,36 @@ describe("Codex session permission policy", () => {
});
});
it.each([
{
mode: "guarded" as const,
policies: ["untrusted"],
approvalsReviewer: "user",
},
{
mode: "workspace" as const,
policies: ["untrusted", "never"],
approvalsReviewer: "auto_review",
},
])("preserves managed prompting approval for a $mode session", (expected) => {
const resolved = applyCodexSessionPermissionPolicy({
appServer: appServer(),
permissionMode: expected.mode,
sessionRoot: "/workspace/project",
pluginConfig,
canUseAutoReview: true,
requirementsToml: `allowed_approval_policies = [${expected.policies
.map((policy) => `"${policy}"`)
.join(", ")}]`,
});
expect(resolved).toMatchObject({
sandbox: "workspace-write",
approvalPolicy: "untrusted",
approvalsReviewer: expected.approvalsReviewer,
});
});
it.each([
{
mode: "full" as const,
extensions/codex/src/app-server/session-permission-policy.ts
+3 -3
View File
@@ -2,8 +2,8 @@ import { hostname as readHostName } from "node:os";
import type { EmbeddedRunAttemptParamsV2 } from "openclaw/plugin-sdk/agent-harness-runtime";
import { isPathInside } from "openclaw/plugin-sdk/security-runtime";
import type {
CodexAppServerApprovalPolicy,
CodexAppServerApprovalsReviewer,
CodexAppServerManagedApprovalPolicy,
CodexAppServerRuntimeOptions,
CodexAppServerSandboxMode,
CodexPluginConfig,
@@ -23,7 +23,7 @@ import { resolveCodexAppServerNetworkProxy } from "./config-security.js";
type SessionPermissionMode = NonNullable<EmbeddedRunAttemptParamsV2["permissionMode"]>;
type CodexSessionPermissionTuple = {
approvalPolicy: CodexAppServerApprovalPolicy;
approvalPolicy: CodexAppServerManagedApprovalPolicy;
approvalsReviewer: CodexAppServerApprovalsReviewer;
sandbox: CodexAppServerSandboxMode;
};
@@ -61,7 +61,7 @@ function requirementsAllowTuple(
tuple: CodexSessionPermissionTuple,
allowed: {
sandboxes: Set<CodexAppServerSandboxMode> | undefined;
approvalPolicies: Set<CodexAppServerApprovalPolicy> | undefined;
approvalPolicies: Set<CodexAppServerManagedApprovalPolicy> | undefined;
reviewers: Set<CodexAppServerApprovalsReviewer> | undefined;
},
): boolean {
extensions/codex/src/app-server/transcript-history-projection.ts
+11 -5
View File
@@ -168,6 +168,9 @@ function projectCodexThreadHistory(params: {
if (!text || !role) {
continue;
}
const phase =
item.phase === "commentary" || item.phase === "final_answer" ? item.phase : undefined;
const asyncDelivery = item.delivery === "async";
const message =
role === "assistant"
? attachCodexMirrorIdentity(
@@ -190,13 +193,13 @@ function projectCodexThreadHistory(params: {
...(turn.status === "failed" && turn.error?.message
? { errorMessage: turn.error.message }
: {}),
...(phase ? { phase } : {}),
...(asyncDelivery && itemId ? { openclawAsyncDelivery: { itemId } } : {}),
timestamp,
} satisfies AssistantMessage,
identity,
)
: attachCodexMirrorIdentity({ role, content: text, timestamp } as AgentMessage, identity);
const phase =
item.phase === "commentary" || item.phase === "final_answer" ? item.phase : undefined;
projected.push({
message,
responseItem: {
@@ -262,7 +265,9 @@ export function projectBoundedCodexThreadHistory(params: {
.filter(
({ message }) =>
message.role !== "assistant" ||
(message.stopReason !== "aborted" && message.stopReason !== "error"),
(message.stopReason !== "aborted" &&
message.stopReason !== "error" &&
!("openclawAsyncDelivery" in message)),
)
.map(({ responseItem }) => responseItem),
transcriptMessages: selected.map(({ message }) => message),
@@ -280,8 +285,9 @@ export function projectBoundedCodexVisibleSessionHistory(
}
if (
entry.role === "assistant" &&
"stopReason" in entry.message &&
(entry.message.stopReason === "aborted" || entry.message.stopReason === "error")
(("stopReason" in entry.message &&
(entry.message.stopReason === "aborted" || entry.message.stopReason === "error")) ||
"openclawAsyncDelivery" in entry.message)
) {
continue;
}
extensions/codex/src/app-server/transcript-mirror.test.ts
+67
View File
@@ -24,6 +24,7 @@ import {
import { afterEach, describe, expect, it, vi } from "vitest";
import type { CodexThread } from "./protocol.js";
import { readCodexMirroredSessionHistoryMessages } from "./session-history.js";
import { projectBoundedCodexVisibleSessionHistory } from "./transcript-history-projection.js";
import {
buildCodexUserPromptMessage,
codexTranscriptMirrorRuntime,
@@ -651,6 +652,72 @@ describe("projectBoundedCodexThreadHistory", () => {
expect(JSON.stringify(projection)).not.toContain("failed tail");
});
it("preserves imported async and commentary ownership while keeping async messages out of model history", () => {
const importedThread = {
...thread,
turns: [
{
id: "turn-async-history",
status: "completed",
items: [
{
id: "user-async-history",
type: "userMessage",
content: [{ type: "text", text: "Investigate this" }],
},
{
id: "commentary-history",
type: "agentMessage",
text: "Checking the deployment.",
phase: "commentary",
},
{
id: "async-history",
type: "agentMessage",
text: "Which environment should I use?",
phase: "final_answer",
delivery: "async",
},
{
id: "final-history",
type: "agentMessage",
text: "Deployment complete.",
phase: "final_answer",
},
],
},
],
} as unknown as CodexThread;
const projection = projectBoundedCodexThreadHistory({
thread: importedThread,
throughTurnId: "turn-async-history",
importedAt: 1_800_000_000_000,
});
expect(projection.transcriptMessages).toHaveLength(4);
expect(projection.transcriptMessages[1]).toMatchObject({ phase: "commentary" });
expect(projection.transcriptMessages[2]).toMatchObject({
phase: "final_answer",
openclawAsyncDelivery: { itemId: "async-history" },
});
expect(JSON.stringify(projection.responseItems)).not.toContain(
"Which environment should I use?",
);
expect(projection.responseItems).toHaveLength(3);
const visibleSessionHistory = projectBoundedCodexVisibleSessionHistory(
projection.transcriptMessages.map((message, index) => ({
entryId: `entry-${index}`,
parentId: index === 0 ? null : `entry-${index - 1}`,
seq: index,
role: message.role,
message,
})),
);
expect(JSON.stringify(visibleSessionHistory)).not.toContain("Which environment should I use?");
expect(visibleSessionHistory).toHaveLength(3);
});
it("accepts terminal boundaries", () => {
for (const [status, stopReason] of [
["completed", "stop"],
extensions/codex/src/app-server/transcript-mirror.ts
+17 -2
View File
@@ -1,5 +1,6 @@
import { createHash } from "node:crypto";
import {
deliverAgentHarnessUserInputPrompt,
embeddedAgentLog,
formatErrorMessage,
projectAgentHarnessTranscriptMessageForDisplay,
@@ -578,7 +579,7 @@ async function deliverAsyncMessageBestEffort(params: {
return "retry";
}
try {
await params.params.onBlockReply({ text: params.text }, { deliveryIntentId });
await deliverAsyncBlockReply(params.params.onBlockReply, params.text, deliveryIntentId);
return "settled";
} catch (error) {
embeddedAgentLog.warn("failed to deliver codex async agent message", {
@@ -623,7 +624,7 @@ async function deliverAsyncMessageBestEffort(params: {
);
if (params.params.onBlockReply && persistedText !== undefined) {
try {
await params.params.onBlockReply({ text: persistedText }, { deliveryIntentId });
await deliverAsyncBlockReply(params.params.onBlockReply, persistedText, deliveryIntentId);
} catch (error) {
embeddedAgentLog.warn("failed to deliver persisted codex async agent message", {
error: formatErrorMessage(error),
@@ -644,6 +645,20 @@ export const codexTranscriptMirrorRuntime = {
mirrorBestEffort,
};
async function deliverAsyncBlockReply(
onBlockReply: NonNullable<EmbeddedRunAttemptParams["onBlockReply"]>,
text: string,
deliveryIntentId: string,
): Promise<void> {
// Harness-owned prompts already carry the host's canonical source-delivery
// authorization; an empty question list keeps the upstream message exact.
await deliverAgentHarnessUserInputPrompt(
{ onBlockReply: (payload) => onBlockReply(payload, { deliveryIntentId }) },
[],
{ intro: text },
);
}
function resolveCodexMirrorTranscriptTarget(params: {
agentId?: string;
sessionId: string;
extensions/codex/src/conversation-binding.ts
+2 -1
View File
@@ -46,6 +46,7 @@ import {
isCodexAppServerIndeterminateRequestCancellationError,
type CodexAppServerClient,
} from "./app-server/client.js";
import type { CodexAppServerManagedApprovalPolicy } from "./app-server/config-contracts.js";
import {
canUseCodexModelBackedApprovalsReviewerForModel,
codexSandboxPolicyForTurn,
@@ -461,7 +462,7 @@ type CodexThreadBindingParams = {
model?: string;
modelProvider?: string;
authProfileId?: string;
approvalPolicy?: CodexAppServerApprovalPolicy;
approvalPolicy?: CodexAppServerManagedApprovalPolicy;
sandbox?: CodexAppServerSandboxMode;
serviceTier?: CodexServiceTier;
config?: CodexAppServerAuthProfileLookup["config"];
src/auto-reply/reply/dispatch-from-config.delivery-and-tts.test-utils.ts
+30
View File
@@ -1952,6 +1952,36 @@ describe("dispatchReplyFromConfig", () => {
});
});
it("delivers independent durable updates immediately without mixing them into the final Telegram voice reply", async () => {
setNoAbort();
installCaptionedVoiceTestPlugin("telegram");
ttsMocks.state.synthesizeFinalAudio = true;
const dispatcher = createDispatcher();
const ctx = buildTestCtx({ Provider: "telegram", Surface: "telegram" });
const replyResolver = async (
_ctx: MsgContext,
opts?: GetReplyOptions,
): Promise<ReplyPayload> => {
await opts?.onBlockReply?.(
{ text: "Which environment should I use?" },
{ deliveryIntentId: "block-reply:v1:codex-app-server:thread-1:turn-1:question" },
);
expect(dispatcher.sendBlockReply).toHaveBeenCalledWith({
text: "Which environment should I use?",
});
await opts?.onBlockReply?.({ text: "The selected environment is ready." });
return { text: "Deployment complete." };
};
await dispatchReplyFromConfig({ ctx, cfg: emptyConfig, dispatcher, replyResolver });
expect(firstFinalReplyPayload(dispatcher)).toMatchObject({
text: "The selected environment is ready.\nDeployment complete.",
mediaUrl: "https://example.com/tts-synth.opus",
audioAsVoice: true,
});
});
it("delivers deferred Telegram text when synthesis produces no audio", async () => {
setNoAbort();
installCaptionedVoiceTestPlugin("telegram");
src/auto-reply/reply/dispatch-from-config.execute.ts
+14 -15
View File
@@ -427,6 +427,10 @@ export async function executeDispatch(state: PrepareDispatchExecutionReadyState)
}
// Buffered commentary preceded this block; deliver it first.
await flushPendingCommentaryProgress();
const independentDurableBlock = context?.deliveryIntentId !== undefined;
if (independentDurableBlock && state.suppressAcpChildUserDelivery) {
return;
}
if (
state.suppressDelivery &&
!shouldDeliverDespiteSourceReplySuppression(inputPayload, state)
@@ -458,12 +462,12 @@ export async function executeDispatch(state: PrepareDispatchExecutionReadyState)
// and must not be synthesised into the spoken reply. Display
// lanes stay out too: they are presentation, never final text.
const isStatusNotice = isReplyPayloadStatusNotice(payload);
if (
payload.text &&
const contributesToFinalReply =
!isStatusNotice &&
!independentDurableBlock &&
payload.isReasoning !== true &&
payload.isCommentary !== true
) {
payload.isCommentary !== true;
if (payload.text && contributesToFinalReply) {
const joinsBufferedTtsDirective =
cleanBlockTtsDirectiveText?.hasBufferedDirectiveText() === true;
if (state.progressState.accumulatedBlockText.length > 0) {
@@ -480,11 +484,7 @@ export async function executeDispatch(state: PrepareDispatchExecutionReadyState)
state.progressState.blockCount++;
}
let visiblePayload =
payload.text &&
cleanBlockTtsDirectiveText &&
!isStatusNotice &&
payload.isReasoning !== true &&
payload.isCommentary !== true
payload.text && cleanBlockTtsDirectiveText && contributesToFinalReply
? (() => {
const text = cleanBlockTtsDirectiveText.push(payload.text);
return copyReplyPayloadMetadata(payload, {
@@ -493,11 +493,7 @@ export async function executeDispatch(state: PrepareDispatchExecutionReadyState)
});
})()
: payload;
const deferThisBlock =
deferFinalTtsText &&
!isStatusNotice &&
payload.isReasoning !== true &&
payload.isCommentary !== true;
const deferThisBlock = deferFinalTtsText && contributesToFinalReply;
if (deferThisBlock) {
const hasNonTextContent = Boolean(
visiblePayload.mediaUrl ||
@@ -547,7 +543,10 @@ export async function executeDispatch(state: PrepareDispatchExecutionReadyState)
if (isDispatchOperationAborted()) {
return;
}
if (context?.deliveryIntentId || shouldRouteToOriginating) {
if (
shouldRouteToOriginating ||
(independentDurableBlock && state.canRouteDurableBlockReply)
) {
const result = await sendPayloadAsync(
normalizedPayload,
context?.abortSignal,
src/auto-reply/reply/dispatch-from-config.prepare-delivery.ts
+14 -2
View File
@@ -84,6 +84,15 @@ export async function prepareDispatchDelivery(state: GatherDispatchRequestReadyS
isRoutableChannel: routeReplyRuntime?.isRoutableChannel ?? (() => false),
});
const routeReplyTo = replyRoute.to;
// Durable intent identifies an outbound write; it never authorizes a new
// destination or bypasses private-webchat and parent-owned-session fences.
const canRouteDurableBlockReply = Boolean(
!suppressAcpChildUserDelivery &&
!isInternalWebchatTurn &&
routeReplyChannel &&
routeReplyTo &&
routeReplyChannel === normalizedCurrentSurface,
);
const deliveryChannel = shouldRouteToOriginating ? routeReplyChannel : currentSurface;
const replyContextAccountId = routeReplyChannel
? resolveReplyDeliveryAccountId(cfg, routeReplyChannel, replyRoute.accountId)
@@ -133,10 +142,12 @@ export async function prepareDispatchDelivery(state: GatherDispatchRequestReadyS
deliveryIntentId?: string;
},
) => {
const durableRouteAuthorized =
options?.deliveryIntentId !== undefined && canRouteDurableBlockReply;
const runtime =
routeReplyRuntime ?? (options?.deliveryIntentId ? await loadRouteReplyRuntime() : undefined);
routeReplyRuntime ?? (durableRouteAuthorized ? await loadRouteReplyRuntime() : undefined);
if (
(!shouldRouteToOriginating && !options?.deliveryIntentId) ||
(!shouldRouteToOriginating && !durableRouteAuthorized) ||
!routeReplyChannel ||
!routeReplyTo ||
!runtime
@@ -281,6 +292,7 @@ export async function prepareDispatchDelivery(state: GatherDispatchRequestReadyS
normalizedCurrentSurface,
isInternalWebchatTurn,
routeReplyChannel,
canRouteDurableBlockReply,
shouldRouteToOriginating,
shouldSuppressTyping,
routeReplyTo,
src/auto-reply/reply/dispatch-from-config.routing.test-utils.ts
+65
View File
@@ -303,6 +303,71 @@ describe("dispatchReplyFromConfig", () => {
expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1);
});
it("never lets a durable block intent route a private webchat turn to an inherited external recipient", async () => {
setNoAbort();
mocks.routeReply.mockClear();
installThreadingTestPlugin({ id: "imessage" });
const dispatcher = createDispatcher();
const ctx = buildTestCtx({
Provider: "webchat",
Surface: "webchat",
OriginatingChannel: "imessage",
OriginatingTo: "imessage:+15550001111",
});
const replyResolver = async (_ctx: MsgContext, opts?: GetReplyOptions) => {
await requireBlockReplyHandler(opts?.onBlockReply)(
{ text: "Private dashboard update" },
{ deliveryIntentId: "block-reply:v1:codex-app-server:thread-1:turn-1:private" },
);
return undefined;
};
await dispatchReplyFromConfig({
ctx,
cfg: automaticDirectReplyConfig,
dispatcher,
replyResolver,
});
expect(mocks.routeReply).not.toHaveBeenCalled();
expect(dispatcher.sendBlockReply).toHaveBeenCalledWith({ text: "Private dashboard update" });
});
it("never lets a durable block intent deliver directly from a parent-owned background session", async () => {
setNoAbort();
mocks.routeReply.mockClear();
installThreadingTestPlugin({ id: "telegram" });
sessionStoreMocks.currentEntry = {
sessionId: "background-child",
spawnedBy: "agent:main:parent",
acp: { backend: "codex" },
};
const dispatcher = createDispatcher();
const replyResolver = async (_ctx: MsgContext, opts?: GetReplyOptions) => {
await requireBlockReplyHandler(opts?.onBlockReply)(
{ text: "Private delegated progress" },
{ deliveryIntentId: "block-reply:v1:codex-app-server:thread-1:turn-1:child" },
);
return undefined;
};
await dispatchReplyFromConfig({
ctx: buildTestCtx({
Provider: "telegram",
Surface: "telegram",
SessionKey: "agent:main:background-child",
OriginatingChannel: "telegram",
OriginatingTo: "telegram:999",
}),
cfg: automaticDirectReplyConfig,
dispatcher,
replyResolver,
});
expect(mocks.routeReply).not.toHaveBeenCalled();
expect(dispatcher.sendBlockReply).not.toHaveBeenCalled();
});
it("routes external origin replies for internal webchat turns when explicit delivery is set", async () => {
setNoAbort();
mocks.routeReply.mockClear();
src/auto-reply/reply/dispatch-from-config.send-policy-routing.test-utils.ts
+43
View File
@@ -215,6 +215,49 @@ describe("sendPolicy deny — suppress delivery, not processing (#53328)", () =>
expect(dispatcher.sendBlockReply).toHaveBeenCalledTimes(1);
});
it.each([
{ name: "permits", sendPolicy: "allow", delivered: true, context: {} },
{ name: "rejects", sendPolicy: "deny", delivered: false, context: {} },
{
name: "rejects ambient room events for",
sendPolicy: "allow",
delivered: false,
context: { ChatType: "group", InboundEventKind: "room_event" } as const,
},
])(
"$name authorized durable harness updates in message-tool-only turns",
async ({ sendPolicy, delivered, context }) => {
setNoAbort();
sessionStoreMocks.currentEntry = { sessionId: "s1", updatedAt: 0, sendPolicy };
const dispatcher = createDispatcher();
const payload = setReplyPayloadMetadata(
{ text: "Background agent update." },
{ deliverDespiteSourceReplySuppression: true },
);
const replyResolver = vi.fn(async (_ctx: MsgContext, opts?: GetReplyOptions) => {
await requireBlockReplyHandler(opts?.onBlockReply)(payload, {
deliveryIntentId: "block-reply:v1:codex-app-server:thread-1:turn-1:async-update",
});
return [];
});
await dispatchReplyFromConfig({
ctx: buildTestCtx({
ChatType: "direct",
SessionKey: "test:async-harness-update",
...context,
}),
cfg: emptyConfig,
dispatcher,
replyResolver,
replyOptions: { sourceReplyDeliveryMode: "message_tool_only" },
});
expect(dispatcher.sendBlockReply).toHaveBeenCalledTimes(delivered ? 1 : 0);
expect(dispatcher.sendFinalReply).not.toHaveBeenCalled();
},
);
it("keeps hook-cancelled marked blocks out of delivery and queued callbacks", async () => {
setNoAbort();
sessionStoreMocks.currentEntry = {
src/plugin-sdk/agent-harness-runtime.test.ts
+15
View File
@@ -2,6 +2,7 @@
* Tests agent harness runtime helpers and task dispatch behavior.
*/
import { describe, expect, expectTypeOf, it, vi } from "vitest";
import { getReplyPayloadMetadata } from "../auto-reply/reply-payload.js";
import {
attachModelProviderRequestTransport,
buildAgentHarnessUserInputAnswers,
@@ -222,6 +223,20 @@ describe("agent harness runtime SDK facade", () => {
});
describe("agent harness user input helpers", () => {
it("authorizes host-owned text-only harness updates without altering their visible payload", async () => {
const onBlockReply = vi.fn();
await deliverAgentHarnessUserInputPrompt({ onBlockReply }, [], {
intro: "Which environment should I use?",
});
const payload = onBlockReply.mock.calls[0]?.[0];
expect(payload).toEqual({ text: "Which environment should I use?", presentation: undefined });
expect(getReplyPayloadMetadata(payload)).toMatchObject({
deliverDespiteSourceReplySuppression: true,
});
});
it("formats prompts and delivers through blocking replies first", async () => {
const onBlockReply = vi.fn();