feat(codex): wire /goal slash command with mid-turn support (#711)
Adds optional `tryHandleOutOfBand` on AgentSession so providers can handle slash commands as side effects without allocating a turn or tripping the activeForegroundTurnId gate. Codex provider implements it for /goal (set/pause/resume/clear), version-gated to codex >= 0.128.0 with `--enable goals`. Mid-turn /goal pause now works alongside a live turn against the same thread. E2E test against real codex 0.128.0 covers set + mid-turn pause + clear.
This commit is contained in:
@@ -1224,6 +1224,48 @@ export class AgentManager {
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Try to run a prompt out-of-band — i.e. without allocating a foreground turn
|
||||
* and without canceling any active turn. Returns true when the session
|
||||
* accepted the prompt as a side-effect command (e.g. /goal pause). Events
|
||||
* emitted by the handler flow through dispatchStream so they persist and
|
||||
* broadcast like normal timeline events.
|
||||
*/
|
||||
tryRunOutOfBand(agentId: string, prompt: AgentPromptInput): boolean {
|
||||
const agent = this.requireSessionAgent(agentId);
|
||||
const handler = agent.session.tryHandleOutOfBand?.(prompt);
|
||||
if (!handler) {
|
||||
return false;
|
||||
}
|
||||
const dispatch = (event: AgentStreamEvent): void => {
|
||||
// Persist timeline items so they show up in fetchAgentTimeline; broadcast
|
||||
// for live subscribers. Other event types are broadcast only.
|
||||
if (event.type === "timeline") {
|
||||
this.touchUpdatedAt(agent);
|
||||
const row = this.recordTimeline(agent.id, event.item);
|
||||
this.dispatchStream(agent.id, event, {
|
||||
seq: row.seq,
|
||||
epoch: this.timelineStore.getEpoch(agent.id),
|
||||
});
|
||||
return;
|
||||
}
|
||||
this.dispatchStream(agent.id, event);
|
||||
};
|
||||
void (async () => {
|
||||
try {
|
||||
await handler.run({ emit: dispatch });
|
||||
} catch (error) {
|
||||
const text = error instanceof Error ? error.message : "Out-of-band command failed";
|
||||
dispatch({
|
||||
type: "timeline",
|
||||
provider: agent.provider,
|
||||
item: { type: "assistant_message", text: `[Error] ${text}` },
|
||||
});
|
||||
}
|
||||
})();
|
||||
return true;
|
||||
}
|
||||
|
||||
recordUserMessage(
|
||||
agentId: string,
|
||||
text: string,
|
||||
|
||||
@@ -513,6 +513,17 @@ export interface AgentSession {
|
||||
setModel?(modelId: string | null): Promise<void>;
|
||||
setThinkingOption?(thinkingOptionId: string | null): Promise<void>;
|
||||
setFeature?(featureId: string, value: unknown): Promise<void>;
|
||||
/**
|
||||
* Out-of-band prompt handler. When non-null, the manager runs the returned
|
||||
* handler instead of allocating a turn. The handler emits stream events
|
||||
* directly via the provided `emit` callback, which routes through the
|
||||
* manager's persistence + broadcast pipeline. The active foreground turn
|
||||
* (if any) is left untouched, so this is how mid-turn side-effect commands
|
||||
* (e.g. /goal pause) reach the provider without canceling the running turn.
|
||||
*/
|
||||
tryHandleOutOfBand?(prompt: AgentPromptInput): {
|
||||
run(ctx: { emit: (event: AgentStreamEvent) => void }): Promise<void>;
|
||||
} | null;
|
||||
}
|
||||
|
||||
export interface ListModelsOptions {
|
||||
|
||||
@@ -174,7 +174,13 @@ export function startAgentRun(
|
||||
prompt: AgentPromptInput,
|
||||
logger: Logger,
|
||||
options?: StartAgentRunOptions,
|
||||
): void {
|
||||
): { outOfBand: boolean } {
|
||||
// Out-of-band commands (e.g. /goal pause) must run WITHOUT canceling an
|
||||
// in-flight turn — replaceAgentRun would interrupt the running turn. The
|
||||
// intercept lives at this layer so it covers every prompt entrypoint.
|
||||
if (agentManager.tryRunOutOfBand(agentId, prompt)) {
|
||||
return { outOfBand: true };
|
||||
}
|
||||
const shouldReplace = Boolean(options?.replaceRunning && agentManager.hasInFlightRun(agentId));
|
||||
const runOptions = options?.runOptions;
|
||||
const iterator = shouldReplace
|
||||
@@ -189,6 +195,7 @@ export function startAgentRun(
|
||||
logger.error({ err: error, agentId }, "Agent stream failed");
|
||||
}
|
||||
})();
|
||||
return { outOfBand: false };
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -237,7 +244,9 @@ export interface SendPromptToAgentParams {
|
||||
* Every surface that sends a prompt to an agent (Session/WS, MCP, CLI-through-MCP)
|
||||
* MUST go through this so behavior can never drift between them.
|
||||
*/
|
||||
export async function sendPromptToAgent(params: SendPromptToAgentParams): Promise<void> {
|
||||
export async function sendPromptToAgent(
|
||||
params: SendPromptToAgentParams,
|
||||
): Promise<{ outOfBand: boolean }> {
|
||||
const {
|
||||
agentManager,
|
||||
agentStorage,
|
||||
@@ -271,7 +280,7 @@ export async function sendPromptToAgent(params: SendPromptToAgentParams): Promis
|
||||
logger.error({ err: error, agentId }, "Failed to record user message");
|
||||
}
|
||||
|
||||
startAgentRun(agentManager, agentId, prompt, logger, {
|
||||
return startAgentRun(agentManager, agentId, prompt, logger, {
|
||||
replaceRunning: true,
|
||||
runOptions,
|
||||
});
|
||||
|
||||
@@ -88,6 +88,47 @@ const ASSISTANT_MESSAGE_BOUNDARY_MARKDOWN = "\n\n---\n\n";
|
||||
const CODEX_PLAN_IMPLEMENTATION_PROMPT_PREFIX =
|
||||
"The user approved the plan. Implement it now. Do not restate or revise the plan unless blocked.";
|
||||
|
||||
// Codex's experimental `goals` feature ships in 0.128.0+. Older binaries reject
|
||||
// `--enable goals` at launch, so we gate by version and silently skip the flag
|
||||
// (and the /goal slash command) when the binary is too old.
|
||||
const CODEX_GOALS_MIN_VERSION: readonly [number, number, number] = [0, 128, 0];
|
||||
|
||||
function parseCodexVersion(versionOutput: string): [number, number, number] | null {
|
||||
const match = versionOutput.match(/(\d+)\.(\d+)\.(\d+)/);
|
||||
if (!match) return null;
|
||||
return [Number(match[1]), Number(match[2]), Number(match[3])];
|
||||
}
|
||||
|
||||
function codexVersionAtLeast(
|
||||
versionOutput: string,
|
||||
min: readonly [number, number, number],
|
||||
): boolean {
|
||||
const parsed = parseCodexVersion(versionOutput);
|
||||
if (!parsed) return false;
|
||||
for (let i = 0; i < 3; i += 1) {
|
||||
if (parsed[i] > min[i]) return true;
|
||||
if (parsed[i] < min[i]) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
type GoalSubcommand =
|
||||
| { kind: "set"; objective: string }
|
||||
| { kind: "pause" }
|
||||
| { kind: "resume" }
|
||||
| { kind: "clear" }
|
||||
| { kind: "usage" };
|
||||
|
||||
function parseGoalSubcommand(args: string | undefined): GoalSubcommand {
|
||||
const trimmed = (args ?? "").trim();
|
||||
if (!trimmed) return { kind: "usage" };
|
||||
const lower = trimmed.toLowerCase();
|
||||
if (lower === "pause") return { kind: "pause" };
|
||||
if (lower === "resume") return { kind: "resume" };
|
||||
if (lower === "clear") return { kind: "clear" };
|
||||
return { kind: "set", objective: trimmed };
|
||||
}
|
||||
|
||||
const CODEX_APP_SERVER_CAPABILITIES: AgentCapabilityFlags = {
|
||||
supportsStreaming: true,
|
||||
supportsSessionPersistence: true,
|
||||
@@ -2648,6 +2689,7 @@ class CodexAppServerAgentSession implements AgentSession {
|
||||
private readonly spawnAppServer: () => Promise<ChildProcessWithoutNullStreams>,
|
||||
private readonly deps: CodexAppServerAgentDeps = {},
|
||||
private readonly ephemeral: boolean = false,
|
||||
private readonly goalsEnabled: boolean = false,
|
||||
) {
|
||||
this.logger = logger.child({ module: "agent", provider: CODEX_PROVIDER });
|
||||
if (config.modeId === undefined) {
|
||||
@@ -3448,11 +3490,90 @@ class CodexAppServerAgentSession implements AgentSession {
|
||||
appServerSkills.length === 0
|
||||
? await listCodexSkills(this.config.cwd, this.deps.workspaceGitService)
|
||||
: [];
|
||||
return [...appServerSkills, ...fallbackSkills, ...prompts].sort((a, b) =>
|
||||
const builtin: AgentSlashCommand[] = [];
|
||||
if (this.goalsEnabled) {
|
||||
builtin.push({
|
||||
name: "goal",
|
||||
description: "Set, pause, resume, or clear the agent's goal",
|
||||
argumentHint: "[<objective>|pause|resume|clear]",
|
||||
});
|
||||
}
|
||||
return [...builtin, ...appServerSkills, ...fallbackSkills, ...prompts].sort((a, b) =>
|
||||
a.name.localeCompare(b.name),
|
||||
);
|
||||
}
|
||||
|
||||
tryHandleOutOfBand(
|
||||
prompt: AgentPromptInput,
|
||||
): { run(ctx: { emit: (event: AgentStreamEvent) => void }): Promise<void> } | null {
|
||||
if (!this.goalsEnabled) return null;
|
||||
if (typeof prompt !== "string") return null;
|
||||
const parsed = this.parseSlashCommandInput(prompt);
|
||||
if (!parsed || parsed.commandName !== "goal") return null;
|
||||
|
||||
const subcommand = parseGoalSubcommand(parsed.args);
|
||||
return {
|
||||
run: async ({ emit }) => {
|
||||
const text = await this.executeGoalSubcommand(subcommand);
|
||||
emit({
|
||||
type: "timeline",
|
||||
provider: CODEX_PROVIDER,
|
||||
item: { type: "assistant_message", text },
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
private async executeGoalSubcommand(subcommand: GoalSubcommand): Promise<string> {
|
||||
if (subcommand.kind === "usage") {
|
||||
return "Usage: /goal <objective>|pause|resume|clear";
|
||||
}
|
||||
try {
|
||||
await this.connect();
|
||||
if (this.currentThreadId) {
|
||||
await this.ensureThreadLoaded();
|
||||
} else {
|
||||
await this.ensureThread();
|
||||
}
|
||||
if (!this.client || !this.currentThreadId) {
|
||||
throw new Error("Codex thread is not available");
|
||||
}
|
||||
switch (subcommand.kind) {
|
||||
case "set": {
|
||||
await this.client.request("thread/goal/set", {
|
||||
threadId: this.currentThreadId,
|
||||
objective: subcommand.objective,
|
||||
status: "active",
|
||||
});
|
||||
return `Goal set: ${subcommand.objective}`;
|
||||
}
|
||||
case "pause": {
|
||||
await this.client.request("thread/goal/set", {
|
||||
threadId: this.currentThreadId,
|
||||
status: "paused",
|
||||
});
|
||||
return "Goal paused.";
|
||||
}
|
||||
case "resume": {
|
||||
await this.client.request("thread/goal/set", {
|
||||
threadId: this.currentThreadId,
|
||||
status: "active",
|
||||
});
|
||||
return "Goal resumed.";
|
||||
}
|
||||
case "clear": {
|
||||
await this.client.request("thread/goal/clear", {
|
||||
threadId: this.currentThreadId,
|
||||
});
|
||||
return "Goal cleared.";
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : "unknown error";
|
||||
return `Failed to update goal: ${message}`;
|
||||
}
|
||||
}
|
||||
|
||||
private async resolveModelAndThinking(): Promise<{
|
||||
model: string;
|
||||
thinkingOptionId: string | undefined;
|
||||
@@ -4489,6 +4610,7 @@ class CodexAppServerAgentSession implements AgentSession {
|
||||
export class CodexAppServerAgentClient implements AgentClient {
|
||||
readonly provider = CODEX_PROVIDER;
|
||||
readonly capabilities = CODEX_APP_SERVER_CAPABILITIES;
|
||||
private goalsEnabledPromise: Promise<boolean> | null = null;
|
||||
|
||||
constructor(
|
||||
private readonly logger: Logger,
|
||||
@@ -4496,17 +4618,41 @@ export class CodexAppServerAgentClient implements AgentClient {
|
||||
private readonly deps: CodexAppServerAgentDeps = {},
|
||||
) {}
|
||||
|
||||
private resolveGoalsEnabled(): Promise<boolean> {
|
||||
if (!this.goalsEnabledPromise) {
|
||||
this.goalsEnabledPromise = (async () => {
|
||||
try {
|
||||
const launchPrefix = await resolveCodexLaunchPrefix(this.runtimeSettings);
|
||||
const versionOutput = await resolveBinaryVersion(launchPrefix.command);
|
||||
const enabled = codexVersionAtLeast(versionOutput, CODEX_GOALS_MIN_VERSION);
|
||||
this.logger.trace({ versionOutput, enabled }, "Resolved codex goals feature gate");
|
||||
return enabled;
|
||||
} catch (error) {
|
||||
this.logger.warn({ err: error }, "Failed to probe codex version for goals gate");
|
||||
return false;
|
||||
}
|
||||
})();
|
||||
}
|
||||
return this.goalsEnabledPromise;
|
||||
}
|
||||
|
||||
private async spawnAppServer(
|
||||
launchEnv?: Record<string, string>,
|
||||
options?: { goalsEnabled?: boolean },
|
||||
): Promise<ChildProcessWithoutNullStreams> {
|
||||
const launchPrefix = await resolveCodexLaunchPrefix(this.runtimeSettings);
|
||||
const args = [...launchPrefix.args, "app-server"];
|
||||
if (options?.goalsEnabled) {
|
||||
args.push("--enable", "goals");
|
||||
}
|
||||
this.logger.trace(
|
||||
{
|
||||
launchPrefix,
|
||||
goalsEnabled: options?.goalsEnabled === true,
|
||||
},
|
||||
"Spawning Codex app server",
|
||||
);
|
||||
const child = spawnProcess(launchPrefix.command, [...launchPrefix.args, "app-server"], {
|
||||
const child = spawnProcess(launchPrefix.command, args, {
|
||||
detached: process.platform !== "win32",
|
||||
stdio: ["pipe", "pipe", "pipe"],
|
||||
...createProviderEnvSpec({
|
||||
@@ -4524,13 +4670,15 @@ export class CodexAppServerAgentClient implements AgentClient {
|
||||
options?: AgentCreateSessionOptions,
|
||||
): Promise<AgentSession> {
|
||||
const sessionConfig: AgentSessionConfig = { ...config, provider: CODEX_PROVIDER };
|
||||
const goalsEnabled = await this.resolveGoalsEnabled();
|
||||
const session = new CodexAppServerAgentSession(
|
||||
sessionConfig,
|
||||
null,
|
||||
this.logger,
|
||||
() => this.spawnAppServer(launchContext?.env),
|
||||
() => this.spawnAppServer(launchContext?.env, { goalsEnabled }),
|
||||
this.deps,
|
||||
options?.persistSession === false,
|
||||
goalsEnabled,
|
||||
);
|
||||
await session.connect();
|
||||
return session;
|
||||
@@ -4548,12 +4696,15 @@ export class CodexAppServerAgentClient implements AgentClient {
|
||||
provider: CODEX_PROVIDER,
|
||||
cwd: overrides?.cwd ?? storedConfig.cwd ?? process.cwd(),
|
||||
};
|
||||
const goalsEnabled = await this.resolveGoalsEnabled();
|
||||
const session = new CodexAppServerAgentSession(
|
||||
merged,
|
||||
handle,
|
||||
this.logger,
|
||||
() => this.spawnAppServer(launchContext?.env),
|
||||
() => this.spawnAppServer(launchContext?.env, { goalsEnabled }),
|
||||
this.deps,
|
||||
false,
|
||||
goalsEnabled,
|
||||
);
|
||||
await session.connect();
|
||||
return session;
|
||||
|
||||
@@ -0,0 +1,278 @@
|
||||
import { beforeAll, beforeEach, describe, expect, test } from "vitest";
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
import pino from "pino";
|
||||
|
||||
import { CodexAppServerAgentClient } from "../agent/providers/codex-app-server-agent.js";
|
||||
import { resolveBinaryVersion } from "../agent/providers/diagnostic-utils.js";
|
||||
import { DaemonClient } from "../test-utils/daemon-client.js";
|
||||
import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
|
||||
import { getFullAccessConfig, isProviderAvailable } from "./agent-configs.js";
|
||||
|
||||
function tmpCwd(): string {
|
||||
return mkdtempSync(path.join(tmpdir(), "daemon-codex-goal-"));
|
||||
}
|
||||
|
||||
function sleep(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
function generateClientMessageId(): string {
|
||||
return `msg_${Date.now()}_${Math.random().toString(36).slice(2, 11)}`;
|
||||
}
|
||||
|
||||
// /goal lives behind codex's `--enable goals` feature flag, which only exists
|
||||
// on codex-cli >= 0.128.0. Older binaries hard-fail at launch when given the
|
||||
// flag, so the codex provider gates registration on the version probe and
|
||||
// older boxes never see the command. Tests must mirror that gate.
|
||||
const GOAL_MIN_VERSION: readonly [number, number, number] = [0, 128, 0];
|
||||
|
||||
function parseVersion(versionOutput: string): [number, number, number] | null {
|
||||
const match = versionOutput.match(/(\d+)\.(\d+)\.(\d+)/);
|
||||
if (!match) return null;
|
||||
return [Number(match[1]), Number(match[2]), Number(match[3])];
|
||||
}
|
||||
|
||||
function meetsMinVersion(versionOutput: string, min: readonly [number, number, number]): boolean {
|
||||
const parsed = parseVersion(versionOutput);
|
||||
if (!parsed) return false;
|
||||
for (let i = 0; i < 3; i += 1) {
|
||||
if (parsed[i] > min[i]) return true;
|
||||
if (parsed[i] < min[i]) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
async function waitForAssistantWaitingOnSleep(
|
||||
client: DaemonClient,
|
||||
agentId: string,
|
||||
timeoutMs = 75_000,
|
||||
): Promise<void> {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
const timeline = await client.fetchAgentTimeline(agentId, { limit: 100 });
|
||||
const assistantText = timeline.entries
|
||||
.filter((entry) => entry.item.type === "assistant_message")
|
||||
.map((entry) => (entry.item as { text: string }).text)
|
||||
.join("\n");
|
||||
if (/sleep|waiting|running/i.test(assistantText) && !/\bdone\b/.test(assistantText)) {
|
||||
return;
|
||||
}
|
||||
await sleep(500);
|
||||
}
|
||||
throw new Error("Timed out waiting for codex to enter the sleep tool call");
|
||||
}
|
||||
|
||||
async function waitForAssistantText(
|
||||
client: DaemonClient,
|
||||
agentId: string,
|
||||
predicate: (text: string) => boolean,
|
||||
timeoutMs: number,
|
||||
description: string,
|
||||
): Promise<void> {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
const timeline = await client.fetchAgentTimeline(agentId, { limit: 200 });
|
||||
const assistantTexts = timeline.entries
|
||||
.filter((entry) => entry.item.type === "assistant_message")
|
||||
.map((entry) => (entry.item as { text: string }).text);
|
||||
if (assistantTexts.some(predicate)) {
|
||||
return;
|
||||
}
|
||||
await sleep(250);
|
||||
}
|
||||
throw new Error(`Timed out waiting for ${description}`);
|
||||
}
|
||||
|
||||
describe("daemon E2E (real codex) - /goal command", () => {
|
||||
let canRun = false;
|
||||
|
||||
beforeAll(async () => {
|
||||
if (!(await isProviderAvailable("codex"))) {
|
||||
return;
|
||||
}
|
||||
const versionOutput = await resolveBinaryVersion("codex");
|
||||
canRun = meetsMinVersion(versionOutput, GOAL_MIN_VERSION);
|
||||
});
|
||||
|
||||
beforeEach((context) => {
|
||||
if (!canRun) {
|
||||
context.skip();
|
||||
}
|
||||
});
|
||||
|
||||
test("/goal <objective> sets the goal and the model can read it back", async () => {
|
||||
const logger = pino({ level: "silent" });
|
||||
const cwd = tmpCwd();
|
||||
const daemon = await createTestPaseoDaemon({
|
||||
agentClients: { codex: new CodexAppServerAgentClient(logger) },
|
||||
logger,
|
||||
});
|
||||
const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
|
||||
|
||||
try {
|
||||
await client.connect();
|
||||
await client.fetchAgents({ subscribe: { subscriptionId: "primary" } });
|
||||
|
||||
const agent = await client.createAgent({
|
||||
cwd,
|
||||
title: "codex-goal-set",
|
||||
...getFullAccessConfig("codex"),
|
||||
});
|
||||
|
||||
const sentinel = "TX9Q4Z";
|
||||
await client.sendMessage(agent.id, `/goal ship feature ${sentinel}`);
|
||||
// Out-of-band commands don't drive the agent through running→idle, so
|
||||
// wait on the ack text directly instead of waitForFinish.
|
||||
await waitForAssistantText(
|
||||
client,
|
||||
agent.id,
|
||||
(text) => text.startsWith("Goal set: ship feature " + sentinel),
|
||||
30_000,
|
||||
"/goal set acknowledgement",
|
||||
);
|
||||
|
||||
// Model awareness check: ask what the goal is. Per the user-visible
|
||||
// behavior the codex experiment promises, the model should reference
|
||||
// the objective in its response.
|
||||
await client.sendMessage(
|
||||
agent.id,
|
||||
"What is your current goal? Reply with just the objective text, nothing else.",
|
||||
);
|
||||
const askResult = await client.waitForFinish(agent.id, 120_000);
|
||||
expect(askResult.status).toBe("idle");
|
||||
|
||||
const finalTimeline = await client.fetchAgentTimeline(agent.id, { limit: 200 });
|
||||
const finalAssistantTexts = finalTimeline.entries
|
||||
.filter((entry) => entry.item.type === "assistant_message")
|
||||
.map((entry) => (entry.item as { text: string }).text);
|
||||
const lastModelReply = finalAssistantTexts[finalAssistantTexts.length - 1] ?? "";
|
||||
expect(lastModelReply.includes(sentinel)).toBe(true);
|
||||
} finally {
|
||||
await client.close();
|
||||
await daemon.close();
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
}
|
||||
}, 300_000);
|
||||
|
||||
test("/goal pause mid-turn does not cancel the running turn", async () => {
|
||||
const logger = pino({ level: "silent" });
|
||||
const cwd = tmpCwd();
|
||||
const daemon = await createTestPaseoDaemon({
|
||||
agentClients: { codex: new CodexAppServerAgentClient(logger) },
|
||||
logger,
|
||||
});
|
||||
const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
|
||||
|
||||
try {
|
||||
await client.connect();
|
||||
await client.fetchAgents({ subscribe: { subscriptionId: "primary" } });
|
||||
|
||||
const agent = await client.createAgent({
|
||||
cwd,
|
||||
title: "codex-goal-mid-turn",
|
||||
...getFullAccessConfig("codex"),
|
||||
});
|
||||
|
||||
// Set an initial goal so pause has something to act on.
|
||||
await client.sendMessage(agent.id, "/goal pilot the long-turn flow");
|
||||
await waitForAssistantText(
|
||||
client,
|
||||
agent.id,
|
||||
(text) => text.startsWith("Goal set: pilot the long-turn flow"),
|
||||
30_000,
|
||||
"/goal set acknowledgement",
|
||||
);
|
||||
|
||||
// Kick off a long-running tool call. We must NOT await this turn.
|
||||
await client.sendAgentMessage(
|
||||
agent.id,
|
||||
"Use the shell tool to run exactly: sleep 30 && echo done. Then report what it outputs. Be brief.",
|
||||
{ messageId: generateClientMessageId() },
|
||||
);
|
||||
await client.waitForAgentUpsert(
|
||||
agent.id,
|
||||
(snapshot) => snapshot.status === "running",
|
||||
60_000,
|
||||
);
|
||||
await waitForAssistantWaitingOnSleep(client, agent.id);
|
||||
|
||||
// Mid-turn: send /goal pause. This must not cancel the running turn.
|
||||
await client.sendAgentMessage(agent.id, "/goal pause", {
|
||||
messageId: generateClientMessageId(),
|
||||
});
|
||||
|
||||
// The running turn must still finish naturally with the sleep output.
|
||||
const finishResult = await client.waitForFinish(agent.id, 180_000);
|
||||
expect(finishResult.status).toBe("idle");
|
||||
|
||||
// Use canonical projection so each timeline row stays as-is. The default
|
||||
// `projected` view merges contiguous assistant_message rows (a streaming
|
||||
// optimization), which would coalesce the out-of-band ack with codex's
|
||||
// next streamed delta. Canonical view sees the underlying raw rows.
|
||||
const timeline = await client.fetchAgentTimeline(agent.id, {
|
||||
limit: 200,
|
||||
projection: "canonical",
|
||||
});
|
||||
const assistantTexts = timeline.entries
|
||||
.filter((entry) => entry.item.type === "assistant_message")
|
||||
.map((entry) => (entry.item as { text: string }).text);
|
||||
|
||||
// Pause acknowledgement must be present as its own row (out-of-band).
|
||||
expect(assistantTexts.some((text) => text === "Goal paused.")).toBe(true);
|
||||
// Original turn must have completed and reported sleep's output.
|
||||
expect(assistantTexts.some((text) => /\bdone\b/.test(text))).toBe(true);
|
||||
// No system-level error fallback.
|
||||
expect(assistantTexts.some((text) => text.startsWith("[Error]"))).toBe(false);
|
||||
expect(assistantTexts.some((text) => text.includes("Failed to update goal"))).toBe(false);
|
||||
} finally {
|
||||
await client.close();
|
||||
await daemon.close();
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
}
|
||||
}, 360_000);
|
||||
|
||||
test("/goal clear removes the goal", async () => {
|
||||
const logger = pino({ level: "silent" });
|
||||
const cwd = tmpCwd();
|
||||
const daemon = await createTestPaseoDaemon({
|
||||
agentClients: { codex: new CodexAppServerAgentClient(logger) },
|
||||
logger,
|
||||
});
|
||||
const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
|
||||
|
||||
try {
|
||||
await client.connect();
|
||||
await client.fetchAgents({ subscribe: { subscriptionId: "primary" } });
|
||||
|
||||
const agent = await client.createAgent({
|
||||
cwd,
|
||||
title: "codex-goal-clear",
|
||||
...getFullAccessConfig("codex"),
|
||||
});
|
||||
|
||||
await client.sendMessage(agent.id, "/goal something to be cleared");
|
||||
await waitForAssistantText(
|
||||
client,
|
||||
agent.id,
|
||||
(text) => text.startsWith("Goal set: something to be cleared"),
|
||||
30_000,
|
||||
"/goal set acknowledgement",
|
||||
);
|
||||
|
||||
await client.sendMessage(agent.id, "/goal clear");
|
||||
await waitForAssistantText(
|
||||
client,
|
||||
agent.id,
|
||||
(text) => text === "Goal cleared.",
|
||||
30_000,
|
||||
"/goal clear acknowledgement",
|
||||
);
|
||||
} finally {
|
||||
await client.close();
|
||||
await daemon.close();
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
}
|
||||
}, 240_000);
|
||||
});
|
||||
@@ -7310,8 +7310,9 @@ export class Session {
|
||||
{ agentId, messageId: msg.messageId, textPrefix: msg.text.slice(0, 80) },
|
||||
"send_agent_message_request: dispatching shared sendPromptToAgent",
|
||||
);
|
||||
let dispatchResult: { outOfBand: boolean };
|
||||
try {
|
||||
await sendPromptToAgent({
|
||||
dispatchResult = await sendPromptToAgent({
|
||||
agentManager: this.agentManager,
|
||||
agentStorage: this.agentStorage,
|
||||
agentId,
|
||||
@@ -7335,6 +7336,19 @@ export class Session {
|
||||
return;
|
||||
}
|
||||
|
||||
if (dispatchResult.outOfBand) {
|
||||
this.emit({
|
||||
type: "send_agent_message_response",
|
||||
payload: {
|
||||
requestId: msg.requestId,
|
||||
agentId,
|
||||
accepted: true,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const startAbort = new AbortController();
|
||||
const startTimeoutMs = 15_000;
|
||||
const startTimeout = setTimeout(() => startAbort.abort("timeout"), startTimeoutMs);
|
||||
|
||||
Reference in New Issue
Block a user