mirrors/pikasTech-agentrun
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 支持运行中 steer command

Browse Source
This commit is contained in:
Codex
2026-06-02 10:04:36 +08:00
parent 1d45d272f1
commit d90e01a91c
15 changed files with 206 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/runner/manager-api.ts
+10 -3
View File
@@ -64,12 +64,19 @@ export class RunnerManagerApi {
}
async pollCommands(runId: string, options: { afterSeq?: number; limit?: number; commandId?: string }): Promise<PollCommandsResult> {
const listOptions: { afterSeq?: number; limit?: number } = {};
if (options.afterSeq !== undefined) listOptions.afterSeq = options.afterSeq;
if (options.limit !== undefined) listOptions.limit = options.limit;
const items = await this.listCommands(runId, listOptions);
const selected = options.commandId ? items.find((item) => item.id === options.commandId && item.state === "pending" && item.type === "turn") ?? null : items.find((item) => item.state === "pending" && item.type === "turn") ?? null;
return { items, selected };
}
async listCommands(runId: string, options: { afterSeq?: number; limit?: number } = {}): Promise<CommandRecord[]> {
const afterSeq = options.afterSeq ?? 0;
const limit = options.limit ?? 20;
const response = await this.client.get(`/api/v1/runs/${encodeURIComponent(runId)}/commands?afterSeq=${afterSeq}&limit=${limit}`) as { items?: CommandRecord[] };
const items = Array.isArray(response.items) ? response.items : [];
const selected = options.commandId ? items.find((item) => item.id === options.commandId && item.state === "pending" && item.type === "turn") ?? null : items.find((item) => item.state === "pending" && item.type === "turn") ?? null;
return { items, selected };
return Array.isArray(response.items) ? response.items : [];
}
async ackCommand(commandId: string): Promise<CommandRecord> {
src/runner/run-once.ts
+83 -1
View File
@@ -1,5 +1,5 @@
import { RunnerManagerApi, failureKindFromError, terminalStatusForFailure, errorMessage } from "./manager-api.js";
import { createBackendSession, runBackendTurn, type BackendAdapterOptions, type BackendSession } from "../backend/adapter.js";
import { createBackendSession, runBackendTurn, type BackendActiveTurnControl, type BackendAdapterOptions, type BackendSession } from "../backend/adapter.js";
import { materializeResourceBundle } from "./resource-bundle.js";
import type { BackendEvent, BackendProfile, CommandRecord, FailureKind, JsonRecord, RunRecord, RunnerRecord, TerminalStatus } from "../common/types.js";
import { AgentRunError } from "../common/errors.js";
@@ -81,6 +81,7 @@ export async function runOnce(options: RunnerOnceOptions): Promise<JsonRecord> {
firstPoll = false;
const command = commandsResponse.selected;
if (!command) {
await failPendingSteerWithoutActiveTurn(api, options.runId, commandsResponse.items, runner.id, attemptId);
if (options.oneShot === true) return noPendingResult(runner, claimed, commandResults, commandsResponse.items.length, "one-shot-no-pending");
if (Date.now() - idleSince >= idleTimeoutMs) return noPendingResult(runner, claimed, commandResults, commandsResponse.items.length, "idle-timeout");
await sleep(pollIntervalMs);
@@ -153,6 +154,7 @@ async function executeCommand(api: RunnerManagerApi, options: RunnerOnceOptions,
const abortController = new AbortController();
const stopCancelWatch = watchCancellation(api, options.runId, command.id, abortController);
const stopBackendProgress = startBackendProgress(api, options.runId, command.id, attemptId, runner.id, options.backendProfile ?? null);
let stopSteerWatch: (() => void) | undefined;
try {
const latestRun = await api.getRun(options.runId);
const backendOptions = {
@@ -162,6 +164,13 @@ async function executeCommand(api: RunnerManagerApi, options: RunnerOnceOptions,
onEvent: async (event: BackendEvent) => {
await appendBestEffort(api, options.runId, annotateCommandEvent(event, command.id, attemptId, runner.id));
},
onActiveTurn: (control: BackendActiveTurnControl) => {
stopSteerWatch = startSteerWatch(api, options.runId, command.id, attemptId, runner.id, control, options.pollIntervalMs);
return () => {
stopSteerWatch?.();
stopSteerWatch = undefined;
};
},
};
const result = backendSession ? await backendSession.runTurn(latestRun, command, backendOptions) : await runBackendTurn(latestRun, command, backendOptions);
for (const event of result.events) {
@@ -174,12 +183,85 @@ async function executeCommand(api: RunnerManagerApi, options: RunnerOnceOptions,
const failure = { failureKind, terminalStatus: terminalStatusForFailure(failureKind), message: errorMessage(error) };
return await reportCommandFailure(api, options.runId, command.id, runner, attemptId, failure, "runner:execute");
} finally {
stopSteerWatch?.();
stopBackendProgress();
await appendBestEffort(api, options.runId, { type: "backend_status", payload: { phase: "backend-turn-finished", commandId: command.id, attemptId, runnerId: runner.id } });
stopCancelWatch();
}
}
function startSteerWatch(api: RunnerManagerApi, runId: string, targetCommandId: string, attemptId: string, runnerId: string, control: BackendActiveTurnControl, pollIntervalMs: number | undefined): () => void {
let stopped = false;
let polling = false;
const seen = new Set<string>();
const intervalMs = normalizePollIntervalMs(pollIntervalMs);
const poll = async (): Promise<void> => {
if (stopped || polling) return;
polling = true;
try {
const commands = await api.listCommands(runId, { afterSeq: 0, limit: 100 });
const pendingSteer = commands.filter((item) => item.type === "steer" && item.state === "pending" && !seen.has(item.id));
for (const steerCommand of pendingSteer) {
seen.add(steerCommand.id);
await handleSteerCommand(api, runId, steerCommand, targetCommandId, attemptId, runnerId, control);
}
} catch {
// The active backend turn remains authoritative; missed steer commands stay pending for the next poll.
} finally {
polling = false;
}
};
const timer = setInterval(() => { void poll(); }, intervalMs);
void poll();
return () => {
stopped = true;
clearInterval(timer);
};
}
async function handleSteerCommand(api: RunnerManagerApi, runId: string, command: CommandRecord, targetCommandId: string, attemptId: string, runnerId: string, control: BackendActiveTurnControl): Promise<void> {
const acked = await api.ackCommand(command.id);
if (acked.state === "cancelled") {
await api.reportCommandStatus(command.id, { terminalStatus: "cancelled", failureKind: "cancelled", failureMessage: "steer command cancelled before delivery", threadId: control.threadId, turnId: control.turnId });
return;
}
const prompt = steerPrompt(command.payload);
if (!prompt) {
await reportNonTerminalCommandFailure(api, runId, command.id, attemptId, runnerId, { terminalStatus: "blocked", failureKind: "schema-invalid", message: "steer command payload requires a non-empty prompt, message, or text" }, "runner:steer:payload", control);
return;
}
await appendBestEffort(api, runId, { type: "backend_status", payload: { phase: "steer-command-acknowledged", commandId: command.id, commandType: "steer", targetCommandId, attemptId, runnerId, threadId: control.threadId, turnId: control.turnId } });
try {
await control.steer(prompt);
await appendBestEffort(api, runId, { type: "backend_status", payload: { phase: "turn/steer:completed", commandId: command.id, commandType: "steer", targetCommandId, attemptId, runnerId, threadId: control.threadId, turnId: control.turnId } });
await api.reportCommandStatus(command.id, { terminalStatus: "completed", failureKind: null, failureMessage: null, threadId: control.threadId, turnId: control.turnId });
} catch (error) {
const failureKind = failureKindFromError(error);
await reportNonTerminalCommandFailure(api, runId, command.id, attemptId, runnerId, { terminalStatus: terminalStatusForFailure(failureKind), failureKind, message: errorMessage(error) }, "runner:steer", control);
}
}
async function failPendingSteerWithoutActiveTurn(api: RunnerManagerApi, runId: string, commands: CommandRecord[], runnerId: string, attemptId: string): Promise<void> {
for (const command of commands.filter((item) => item.type === "steer" && item.state === "pending")) {
const acked = await api.ackCommand(command.id);
if (acked.state === "cancelled") continue;
await reportNonTerminalCommandFailure(api, runId, command.id, attemptId, runnerId, { terminalStatus: "blocked", failureKind: "schema-invalid", message: "steer command requires an active turn" }, "runner:steer:no-active-turn");
}
}
async function reportNonTerminalCommandFailure(api: RunnerManagerApi, runId: string, commandId: string, attemptId: string, runnerId: string, failure: { terminalStatus: TerminalStatus; failureKind: FailureKind; message: string }, phase: string, control?: BackendActiveTurnControl): Promise<void> {
await appendBestEffort(api, runId, { type: "error", payload: { failureKind: failure.failureKind, message: failure.message, phase, commandId, attemptId, runnerId, ...(control ? { threadId: control.threadId, turnId: control.turnId } : {}) } });
await api.reportCommandStatus(commandId, { terminalStatus: failure.terminalStatus, failureKind: failure.failureKind, failureMessage: failure.message, ...(control ? { threadId: control.threadId, turnId: control.turnId } : {}) });
}
function steerPrompt(payload: JsonRecord): string | null {
for (const key of ["prompt", "message", "text"]) {
const value = payload[key];
if (typeof value === "string" && value.trim().length > 0) return value.trim();
}
return null;
}
function isTerminalRun(run: RunRecord): boolean {
return run.status === "completed" || run.status === "failed" || run.status === "blocked" || run.status === "cancelled";
}