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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 21 additions & 2 deletions src/permission/reactor-authorize.ts
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ import {
type WorkerGrantStore,
} from "./worker-grant.js";
import { canonicalToolName } from "../agent/canonical-tool-name.js";
import { prepareDispatchedToolCall } from "../agent/tool-aliases.js";
import { getSubAgentIdentity } from "../subagent/identity-context.js";
import {
FLEET_VERBS,
Expand Down Expand Up @@ -100,6 +101,23 @@ function resolveWorkerCwd(cwd: string | undefined): string {
return cwd ?? getSubAgentIdentity()?.cwd ?? process.cwd();
}

/**
* The call as DynamicToolRunner dispatches it: canonical engine name plus
* hidden-alias argument coercion (gpt `shell` -> run_shell via
* coerceShellArgs). Authorize sees the raw wire call; the middleware that
* consumes an authorized retry sees the dispatched call. Keying both on the
* dispatched shape keeps one fingerprint per executed call. Invalid hidden
* arguments never dispatch, so they keep their raw shape.
*/
function dispatchedCall(call: ToolCallType): ToolCallType {
const engine = canonicalToolName(call.name);
try {
return prepareDispatchedToolCall(call, engine);
} catch {
return { ...call, name: engine };
}
}

function workerCallIdentity(
sessionId: string,
call: ToolCallType,
Expand All @@ -110,10 +128,11 @@ function workerCallIdentity(
args: Record<string, unknown>;
cwd: string;
} {
const dispatched = dispatchedCall(call);
return {
sessionId,
canonicalTool: canonicalToolName(call.name),
args: (call.arguments ?? {}) as Record<string, unknown>,
canonicalTool: dispatched.name,
args: (dispatched.arguments ?? {}) as Record<string, unknown>,
cwd,
};
}
Expand Down
147 changes: 147 additions & 0 deletions src/permission/worker-grant-dispatch-identity.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,147 @@
// gpt-family workers call run_shell as `shell`. Reactor authz sees that raw
// wire call; the permission middleware sees DynamicToolRunner's
// prepareDispatchedToolCall output, whose coerceShellArgs rebuilds the
// arguments. The authorized retry must match the same envelope at both
// stages.
import { describe, expect, test } from "bun:test";
import type { ToolCall } from "@intx/types/runtime";

import { prepareDispatchedToolCall } from "../agent/tool-aliases.js";
import { gateToolCall } from "../plugins/permission-plugin.js";
import { createPermissionGate } from "./gate.js";
import { workerPermissionGate } from "./reactor-authorize.js";
import { WorkerGrantStore } from "./worker-grant.js";

const CWD = process.cwd();

// The call shape a gpt-family worker sends for run_shell.
const wireCall = (id: string): ToolCall => ({
id,
name: "shell",
arguments: {
command: "git status --short",
timeout: 120000,
cwd: CWD,
background: false,
stop: "",
},
});

function extractRequestId(reason: string): string {
const match = /grant request ([0-9a-f-]{36})/.exec(reason);
if (match?.[1] === undefined) throw new Error(`no grant id: ${reason}`);
return match[1];
}

describe("gpt-family worker authorized retry identity", () => {
test("approved retry of a `shell` call executes once instead of minting a new grant request", async () => {
// Parent: reactor-gated interactive gate, nothing pre-approved, no auto.
const parentGate = createPermissionGate({
approvals: [],
cwd: CWD,
skipPermissions: false,
interactive: true,
reactorGated: true,
auto: false,
requestApproval: async () => ({ allow: false }),
});
const store = new WorkerGrantStore();
const workerGate = workerPermissionGate(parentGate, {
sessionId: "worker-gpt",
store,
});

// 1. First attempt: reactor authz denies and records envelope A.
const denied = await workerGate.authorizeCall(wireCall("call-1"));
if (denied.effect !== "deny") throw new Error("expected worker deny");
const requestId = extractRequestId(denied.reason);
const envelope = store.peek(requestId);
expect(envelope?.canonicalTool).toBe("run_shell");

// 2. approve_worker_call: parent approved the frozen call (allow-once,
// no session grant) and flipped the envelope to authorized.
expect(store.authorizeRetry(requestId)).toBe(true);

// 3. Retry, stage 1: reactor authz on the raw wire call allows via
// hasAuthorizedRetry (does not consume).
const retryAuthz = await workerGate.authorizeCall(wireCall("call-2"));
expect(retryAuthz).toEqual({ effect: "allow" });
expect(store.peek(requestId)?.status).toBe("authorized");

// 4. Retry, stage 2: DynamicToolRunner coerces the `shell` call onto
// run_shell, then the posix middleware runs gateToolCall.
const dispatched = prepareDispatchedToolCall(
wireCall("call-2"),
"run_shell",
);
expect(dispatched.arguments).not.toEqual(wireCall("x").arguments);

let executed = 0;
const result = await gateToolCall(
workerGate,
dispatched,
new AbortController().signal,
async (call) => {
executed++;
return { callId: call.id, content: "ran" };
},
);

expect(result.isError).not.toBe(true);
expect(executed).toBe(1);
expect(store.peek(requestId)?.status).toBe("consumed");

// One grant, one execution: a replay of the same call fails closed at
// both stages, and no second envelope was minted.
const replay = await gateToolCall(
workerGate,
prepareDispatchedToolCall(wireCall("call-3"), "run_shell"),
new AbortController().signal,
async (call) => {
executed++;
return { callId: call.id, content: "ran" };
},
);
expect(replay.isError).toBe(true);
expect(String(replay.content)).toContain("already consumed");
expect((await workerGate.authorizeCall(wireCall("call-4"))).effect).toBe(
"deny",
);
expect(executed).toBe(1);
});

test("an altered retry (different command) still fails closed", async () => {
const parentGate = createPermissionGate({
approvals: [],
cwd: CWD,
skipPermissions: false,
interactive: true,
reactorGated: true,
auto: false,
requestApproval: async () => ({ allow: false }),
});
const store = new WorkerGrantStore();
const workerGate = workerPermissionGate(parentGate, {
sessionId: "worker-gpt-alt",
store,
});
const denied = await workerGate.authorizeCall(wireCall("call-1"));
if (denied.effect !== "deny") throw new Error("expected worker deny");
const requestId = extractRequestId(denied.reason);
expect(store.authorizeRetry(requestId)).toBe(true);

const altered: ToolCall = {
...wireCall("call-2"),
arguments: { ...wireCall("x").arguments, command: "git push --force" },
};
expect((await workerGate.authorizeCall(altered)).effect).toBe("deny");
const result = await gateToolCall(
workerGate,
prepareDispatchedToolCall(altered, "run_shell"),
new AbortController().signal,
async (call) => ({ callId: call.id, content: "ran" }),
);
expect(result.isError).toBe(true);
expect(store.peek(requestId)?.status).toBe("authorized");
});
});
157 changes: 147 additions & 10 deletions src/session/approval-resume.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ import type {
InboundMessage,
} from "@intx/types/runtime";

import { APPROVAL_TIMEOUT_RESULT_TEXT } from "../permission/decline-markers.js";
import type { PermissionGate } from "../permission/gate.js";
import { createExtraDeniedPathMatcher } from "../plugins/secret-guard-plugin.js";
import {
Expand All @@ -23,6 +24,7 @@ import {
resolveSuspendedApprovalFromStore,
} from "./approval-resume.js";
import { createSessionOperationQueue } from "../tui/delivery-queue.js";
import { createParkedOverlayAbortBinding } from "../tui/runner/parked-overlay-abort.js";
import { defined } from "../../testkit/defined.js";
import { redactedCorrelationId } from "./redacted-correlation.js";
import {
Expand Down Expand Up @@ -936,6 +938,78 @@ describe("approval identity ordering", () => {
expect(decisionBody(message).outcome).toBe("rejected");
});

test("generation change during catch history drops the parked call once", async () => {
const ctx = deferredResume();
const failure = new Error("gate exploded");
ctx.resolveSuspended.mockImplementation(async () => {
throw failure;
});
ctx.history.mockResolvedValueOnce([]).mockImplementationOnce(async () => {
ctx.invalidate();
return [];
});
const handling = ctx.resume.handle(suspension("corr-A", "echo same"));
ctx.lookup.resolve("call-A");
await expect(handling).rejects.toBe(failure);
expect(ctx.onDropped).toHaveBeenCalledTimes(1);
expect(ctx.onDropped).toHaveBeenCalledWith(APPROVAL_DROPPED_NOTICE);
expect(ctx.deliver).toHaveBeenCalledTimes(1);
const message = ctx.deliver.mock.calls[0]?.[0];
if (message === undefined)
throw new Error("expected cancellation rejection");
expect(deliveredCorrelationId(message)).toBe("corr-A");
expect(decisionBody(message)).toEqual({
outcome: "rejected",
message: APPROVAL_DROPPED_NOTICE,
});
});

test("a gate throw with no generation change rejects once as approval surface failed", async () => {
const ctx = deferredResume();
const failure = new Error("gate exploded");
ctx.resolveSuspended.mockImplementation(async () => {
throw failure;
});
const handling = ctx.resume.handle(suspension("corr-A", "echo same"));
ctx.lookup.resolve("call-A");
await expect(handling).rejects.toBe(failure);
expect(ctx.onDropped).not.toHaveBeenCalled();
expect(ctx.deliver).toHaveBeenCalledTimes(1);
const message = ctx.deliver.mock.calls[0]?.[0];
if (message === undefined) throw new Error("expected surface rejection");
expect(deliveredCorrelationId(message)).toBe("corr-A");
expect(decisionBody(message)).toEqual({
outcome: "rejected",
message: "approval surface failed",
});
});

test("generation change during catch history after a timeout does not reject again", async () => {
const ctx = deferredResume();
const failure = new Error("gate exploded");
ctx.resolveSuspended.mockImplementation(async () => {
throw failure;
});
const reading = Promise.withResolvers<undefined>();
const history = Promise.withResolvers<ConversationTurn[]>();
ctx.history.mockResolvedValueOnce([]).mockImplementationOnce(() => {
reading.resolve(undefined);
return history.promise;
});
const handling = ctx.resume.handle(suspension("corr-A", "echo same"));
ctx.lookup.resolve("call-A");
await reading.promise;
expect(ctx.resolveSuspended).toHaveBeenCalledTimes(1);
expect(ctx.registered()).toBeUndefined();
ctx.invalidate();
history.resolve([timeoutTurn("call-A")]);
await expect(handling).rejects.toBe(failure);
expect(ctx.onDropped).toHaveBeenCalledTimes(1);
expect(ctx.onDropped).toHaveBeenCalledWith(APPROVAL_DROPPED_NOTICE);
expect(ctx.deliver).not.toHaveBeenCalled();
expect(ctx.registered()).toBeUndefined();
});

for (const timedOutCallId of ["call-A", "call-B"]) {
test(`generation change during post-gate history with ${timedOutCallId} timeout preserves exact-call cancellation`, async () => {
const ctx = deferredResume();
Expand Down Expand Up @@ -1192,10 +1266,9 @@ describe("approval resume retry re-await", () => {
expect(delivered).toHaveLength(1);
});

// Concurrent duplicates share one in-flight rejection (one gate, zero
// deliveries), then a retry re-gates and delivers exactly once, and later
// retries memoize.
test("a shared rejection re-gates once, delivers once, then memoizes", async () => {
// Concurrent duplicates share one in-flight gate failure (one gate, one
// rejection delivered), and later retries memoize.
test("a shared gate failure rejects the parked call once, then memoizes", async () => {
const gate = Promise.withResolvers<{ allow: boolean }>();
const { resume, delivered, resolveSuspended } = retryHarness({
allow: true,
Expand All @@ -1210,15 +1283,79 @@ describe("approval resume retry re-await", () => {
await expect(first).rejects.toBe(failure);
await expect(second).rejects.toBe(failure);
expect(resolveSuspended).toHaveBeenCalledTimes(1);
expect(delivered).toHaveLength(0);

expect(await resume.handle(suspension("corr-A", "echo alpha"))).toBe(true);
expect(resolveSuspended).toHaveBeenCalledTimes(2);
// Nothing retries a parked call, so the failure answers it.
expect(delivered).toHaveLength(1);
expect(decisionBody(onlyDecision(delivered)).outcome).toBe("approved");
expect(decisionBody(onlyDecision(delivered)).outcome).toBe("rejected");

expect(await resume.handle(suspension("corr-A", "echo alpha"))).toBe(true);
expect(resolveSuspended).toHaveBeenCalledTimes(2);
expect(resolveSuspended).toHaveBeenCalledTimes(1);
expect(delivered).toHaveLength(1);
});
test("a gate failure after the reactor timed the call out delivers nothing", async () => {
const delivered: InboundMessage[] = [];
let timedOut = false;
const failure = new Error("gate exploded");
const resume = createApprovalResume({
getAgent: () => ({
deliver: (message: InboundMessage) => {
delivered.push(message);
},
history: async () =>
timedOut
? ([
{
role: "user",
content: [
{
type: "tool_result",
callId: "call-A",
content: [
{ type: "text", text: APPROVAL_TIMEOUT_RESULT_TEXT },
],
},
],
},
] as unknown as Awaited<ReturnType<Agent["history"]>>)
: [],
}),
resolveParkedCallId: () => "call-A",
gate: {
resolveSuspended: async () => {
timedOut = true;
throw failure;
},
} as unknown as PermissionGate,
});
await expect(
resume.handle(suspension("corr-A", "echo alpha")),
).rejects.toBe(failure);
expect(delivered).toEqual([]);
});
});

describe("parked overlay registration after the operator answers", () => {
// approve_worker_call raises a nested permission prompt as soon as its
// redispatched call starts, while the decision that resumed it is still being
// delivered. That prompt's abort signal merges the parked overlay controller.
test("a permission prompt raised while the decision is delivered is not pre-aborted", async () => {
const binding = createParkedOverlayAbortBinding();
const identity = new AbortController();
const { agent, gate } = createApprovalResumeHarness();
let promptSignal: AbortSignal | undefined;
const resume = createApprovalResume({
getAgent: () => agent,
gate,
resolveParkedCallId: () => "call-A",
registerOverlayAbort: binding.registerOverlayAbort,
deliver: async (message) => {
promptSignal = binding.identitySignal(identity.signal);
agent.deliver(message);
},
});
expect(
await resume.handle(suspension("corr-A", "git status --short")),
).toBe(true);
expect(promptSignal).toBeDefined();
expect(promptSignal?.aborted).toBe(false);
});
});
Loading
Loading