feat: pending-finish proposals and QQ image input
Proposal semantics: worker results no longer distinguish success/fail; only an explicit finish settles a pending proposal. Pending owner input follows up by resuming the original worker session. Dirty pending blocks start_next globally; clean pending only blocks its owner. QQ adapter now downloads image attachments and passes them as ACP image content blocks; video/file attachments degrade to link text.
This commit is contained in:
zenord
+265
-159
@@ -4,17 +4,22 @@ import path from "node:path";
|
||||
import type { AcpConfig, PermissionPolicy } from "../config.js";
|
||||
import { chatKeyFor, type AssistantBinding, type DurableSessionStore } from "../core/durable-session-store.js";
|
||||
import type { Proposal, ProposalPending, ProposalStore } from "../core/proposal-store.js";
|
||||
import type { IncomingAttachment } from "../core/types.js";
|
||||
import type { ResolvedBot } from "../roles/role-registry.js";
|
||||
import type { ConversationRequest, ConversationResponse, ConversationRuntime, RuntimeEventSink, RuntimeStats } from "./types.js";
|
||||
import type { AcpPromptContent } from "./client.js";
|
||||
import { AcpSessionRestoreError, AcpWorker } from "./worker.js";
|
||||
|
||||
const ASSISTANT_POLICY: PermissionPolicy = { mode: "deny", allowedTools: [], allowedCommandPatterns: [] };
|
||||
const ASSISTANT_ENVELOPE = /<GORI_ASSISTANT_ACTION_V1>(?<json>[\s\S]*?)<\/GORI_ASSISTANT_ACTION_V1>\s*$/;
|
||||
const WORKER_ENVELOPE = /<GORI_WORKER_RESULT_V1>(?<json>[\s\S]*?)<\/GORI_WORKER_RESULT_V1>\s*$/;
|
||||
const ASSISTANT_ENVELOPE = /<GORI_ASSISTANT_ACTION_V2>(?<json>[\s\S]*?)<\/GORI_ASSISTANT_ACTION_V2>\s*$/;
|
||||
const WORKER_ENVELOPE = /<GORI_WORKER_RESULT_V2>(?<json>[\s\S]*?)<\/GORI_WORKER_RESULT_V2>\s*$/;
|
||||
|
||||
export type AssistantAction =
|
||||
| { type: "create_proposal"; title: string; goal: string; steps: string[] }
|
||||
| { type: "confirm"; id?: string; answer?: string }
|
||||
| { type: "confirm"; id?: string }
|
||||
| { type: "adjust_proposal"; id?: string; title?: string; goal?: string; steps?: string[] }
|
||||
| { type: "follow_up"; id?: string; instruction: string }
|
||||
| { type: "finish"; id?: string; note?: string }
|
||||
| { type: "start_next" }
|
||||
| { type: "cancel"; id?: string }
|
||||
| { type: "stop" };
|
||||
@@ -24,14 +29,13 @@ export interface ParsedAssistantEnvelope {
|
||||
actions: AssistantAction[];
|
||||
}
|
||||
|
||||
export type WorkerResultStatus = "SUCCESS" | "NEEDS_CONFIRMATION" | "FAILED";
|
||||
export type WorkerResultStatus = "PENDING";
|
||||
|
||||
export interface ParsedWorkerResult {
|
||||
status: WorkerResultStatus;
|
||||
summary: string;
|
||||
question?: string;
|
||||
nextStep?: string;
|
||||
dirty?: boolean;
|
||||
workspaceDirty?: boolean;
|
||||
}
|
||||
|
||||
interface ActiveWorker {
|
||||
@@ -41,6 +45,14 @@ interface ActiveWorker {
|
||||
run?: Promise<void>;
|
||||
}
|
||||
|
||||
interface WorkerLaunchOptions {
|
||||
resumeSessionId?: string;
|
||||
resumeContext?: string;
|
||||
firstTurn?: string;
|
||||
freshSessionNote?: string;
|
||||
images?: IncomingAttachment[];
|
||||
}
|
||||
|
||||
export interface AssistantManagerOptions {
|
||||
assistantWorkspaceHome?: string;
|
||||
kimiCodeHome?: string;
|
||||
@@ -82,13 +94,12 @@ export class AssistantManager implements ConversationRuntime {
|
||||
console.error(`Persisted worker cleanup failed for proposal ${proposal.id} (${error instanceof Error ? error.name : "unknown error"})`);
|
||||
}
|
||||
}
|
||||
const summary = "worker_lost: the runner restarted while this proposal was working; its worker was terminated and the proposal was marked failed";
|
||||
const summary = "worker_lost: the runner restarted while this proposal was working; its worker was terminated and the workspace may be dirty";
|
||||
await this.proposals.update(proposal.id, {
|
||||
status: "failed",
|
||||
pending: { kind: "failure", summary },
|
||||
status: "pending",
|
||||
pending: { summary, workspaceDirty: true, receivedAt: Date.now() },
|
||||
lastWorkerSummary: summary,
|
||||
workerProcessGroup: undefined,
|
||||
finishedAt: Date.now()
|
||||
workerProcessGroup: undefined
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -114,6 +125,11 @@ export class AssistantManager implements ConversationRuntime {
|
||||
return this.scheduler(() => this.confirmLocked(chatKey, userId));
|
||||
}
|
||||
|
||||
async finish(platform: string, chatId: string, userId: string): Promise<boolean> {
|
||||
const chatKey = chatKeyFor(platform, chatId);
|
||||
return this.scheduler(() => this.finishLocked(chatKey, userId));
|
||||
}
|
||||
|
||||
async stop(platform: string, chatId: string, userId: string): Promise<boolean> {
|
||||
const chatKey = chatKeyFor(platform, chatId);
|
||||
return this.scheduler(() => this.stopActiveLocked(chatKey, userId));
|
||||
@@ -141,19 +157,22 @@ export class AssistantManager implements ConversationRuntime {
|
||||
const mine = proposals.filter(owned);
|
||||
const active = this.active;
|
||||
const activeProposal = active ? this.proposals.get(active.proposalId) : undefined;
|
||||
const awaiting = proposals.filter((proposal) => proposal.status === "awaiting_user_confirmation");
|
||||
const pending = proposals.filter((proposal) => proposal.status === "pending");
|
||||
const working = proposals.filter((proposal) => proposal.status === "working");
|
||||
const myQueued = mine.filter((proposal) => proposal.status === "queued").length;
|
||||
const myPending = mine.filter((proposal) => proposal.status === "awaiting_user_confirmation").length;
|
||||
const myPending = mine.filter((proposal) => proposal.status === "pending").length;
|
||||
const myProposed = mine.filter((proposal) => proposal.status === "proposed").length;
|
||||
const schedulerState = active || working.length > 0 ? "working" : awaiting.length > 0 ? "awaiting_confirmation" : "idle";
|
||||
const blockedReason = awaiting.length > 0
|
||||
? (awaiting.some(owned) ? "your proposal is awaiting your confirmation" : "another proposal is awaiting owner confirmation")
|
||||
: active || working.length > 0
|
||||
? (working.some(owned) || (activeProposal && owned(activeProposal)) ? "your proposal is working" : "another proposal is working")
|
||||
: "none";
|
||||
const dirtyPending = pending.some((proposal) => proposal.pending?.workspaceDirty);
|
||||
const schedulerState = active || working.length > 0 ? "working" : pending.length > 0 ? "pending" : "idle";
|
||||
const blockedReason = active || working.length > 0
|
||||
? (working.some(owned) || (activeProposal && owned(activeProposal)) ? "your proposal is working" : "another proposal is working")
|
||||
: myPending > 0
|
||||
? "your proposal is waiting for your decision"
|
||||
: dirtyPending
|
||||
? "a pending proposal left a dirty workspace"
|
||||
: "none";
|
||||
const nextAction = myPending > 0
|
||||
? "confirm your pending proposal"
|
||||
? "finish your pending proposal or follow up with new instructions"
|
||||
: myProposed > 0
|
||||
? "confirm your proposed proposal to queue it"
|
||||
: myQueued > 0
|
||||
@@ -167,11 +186,11 @@ export class AssistantManager implements ConversationRuntime {
|
||||
proposals: proposals.length,
|
||||
queuedProposals: proposals.filter((proposal) => proposal.status === "queued").length,
|
||||
workingProposals: working.length,
|
||||
awaitingConfirmation: awaiting.length,
|
||||
pendingProposals: pending.length,
|
||||
workerRunning: Boolean(active?.worker.inFlight),
|
||||
owner: Boolean(userId && activeProposal && owned(activeProposal)),
|
||||
myQueuedProposals: myQueued,
|
||||
myPendingConfirmations: myPending,
|
||||
myPendingProposals: myPending,
|
||||
schedulerState,
|
||||
blockedReason,
|
||||
nextAction
|
||||
@@ -210,22 +229,43 @@ export class AssistantManager implements ConversationRuntime {
|
||||
private async runAssistantTurn(conversationKey: string, request: ConversationRequest): Promise<string> {
|
||||
const worker = await this.acquireAssistantWorker(conversationKey);
|
||||
try {
|
||||
const reply = await worker.prompt(this.assistantUserPrompt(request));
|
||||
const reply = await worker.prompt(this.assistantPromptContent(worker, request));
|
||||
const parsed = await this.parseAssistantReply(worker, reply);
|
||||
if (worker.isolationViolated) throw new Error("Assistant session attempted forbidden tool activity");
|
||||
await this.store.touchBinding(conversationKey);
|
||||
const corrections = await this.executeActions(chatKeyFor(request.platform, request.chatId), request, parsed.actions);
|
||||
return corrections.length > 0 ? `${parsed.reply}\n${corrections.join("\n")}` : parsed.reply;
|
||||
const chatKey = chatKeyFor(request.platform, request.chatId);
|
||||
const corrections = await this.executeActions(chatKey, request, parsed.actions);
|
||||
const reminders = this.pendingReminders(chatKey, request.userId, parsed.reply);
|
||||
const extras = [...corrections, ...reminders];
|
||||
return extras.length > 0 ? `${parsed.reply}\n${extras.join("\n")}` : parsed.reply;
|
||||
} catch (error) {
|
||||
if (worker.isolationViolated) await this.store.deleteBinding(conversationKey).catch(() => undefined);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private pendingReminders(chatKey: string, userId: string, reply: string): string[] {
|
||||
return this.proposals.list({ status: "pending" })
|
||||
.filter((proposal) => proposal.ownerChatKey === chatKey && proposal.requesterUserId === userId)
|
||||
.filter((proposal) => !reply.includes(proposal.title))
|
||||
.map((proposal) => `提醒：你还有一个待确认的任务「${proposal.title}」，说 finish 结束，或直接继续说要求继续改。`);
|
||||
}
|
||||
|
||||
private assistantPromptContent(worker: AcpWorker, request: ConversationRequest): string | AcpPromptContent[] {
|
||||
const text = this.assistantUserPrompt(request);
|
||||
const images = request.attachments || [];
|
||||
if (images.length === 0) return text;
|
||||
if (!worker.supportsImages) return `${text}\n\n[Attachments]\n${images.map(() => "[图片，当前 agent 不支持图片输入]").join("\n")}`;
|
||||
return [
|
||||
...images.map((image) => ({ type: "image" as const, data: image.data, mimeType: image.mimeType })),
|
||||
{ type: "text" as const, text }
|
||||
];
|
||||
}
|
||||
|
||||
private async parseAssistantReply(worker: AcpWorker, reply: string): Promise<ParsedAssistantEnvelope> {
|
||||
let parsed = parseAssistantActions(reply);
|
||||
if (!parsed) parsed = parseAssistantActions(await worker.prompt(assistantRepairPrompt()));
|
||||
if (!parsed) throw new Error("Assistant did not return a valid GORI_ASSISTANT_ACTION_V1 envelope after one repair attempt");
|
||||
if (!parsed) throw new Error("Assistant did not return a valid GORI_ASSISTANT_ACTION_V2 envelope after one repair attempt");
|
||||
return parsed;
|
||||
}
|
||||
|
||||
@@ -241,8 +281,17 @@ export class AssistantManager implements ConversationRuntime {
|
||||
requesterUserId: request.userId
|
||||
});
|
||||
} else if (action.type === "confirm") {
|
||||
const ok = await this.scheduler(() => this.confirmLocked(chatKey, request.userId, action.id, action.answer));
|
||||
const ok = await this.scheduler(() => this.confirmLocked(chatKey, request.userId, action.id));
|
||||
if (!ok) corrections.push("没有可确认的提案；你只能操作自己发起的任务。");
|
||||
} else if (action.type === "adjust_proposal") {
|
||||
const ok = await this.scheduler(() => this.adjustProposalLocked(chatKey, request.userId, action));
|
||||
if (!ok) corrections.push("没有可调整的提案；只有还没开始执行的任务可以调整，且你只能操作自己发起的任务。");
|
||||
} else if (action.type === "follow_up") {
|
||||
const ok = await this.scheduler(() => this.followUpLocked(chatKey, request.userId, action.id, action.instruction, request.attachments));
|
||||
if (!ok) corrections.push("现在没法继续：只有待确认的任务可以继续，且你只能操作自己发起的任务；如果有别的任务正在执行，也要先等它落定。");
|
||||
} else if (action.type === "finish") {
|
||||
const ok = await this.scheduler(() => this.finishLocked(chatKey, request.userId, action.id, action.note));
|
||||
if (!ok) corrections.push("没有可结束的任务；只有待确认的任务可以 finish，且你只能操作自己发起的任务。");
|
||||
} else if (action.type === "start_next") {
|
||||
const started = await this.scheduler(() => this.tryStartLocked(chatKey, request.userId));
|
||||
if (!started) corrections.push(this.startNextCorrection(chatKey, request.userId));
|
||||
@@ -258,89 +307,99 @@ export class AssistantManager implements ConversationRuntime {
|
||||
}
|
||||
|
||||
private startNextCorrection(chatKey: string, userId: string): string {
|
||||
const queuedMine = this.proposals.list({ status: "queued" })
|
||||
.some((proposal) => proposal.ownerChatKey === chatKey && proposal.requesterUserId === userId);
|
||||
const owned = (proposal: Proposal) => proposal.ownerChatKey === chatKey && proposal.requesterUserId === userId;
|
||||
const queuedMine = this.proposals.list({ status: "queued" }).some(owned);
|
||||
const kept = queuedMine ? "你的任务已保留在队列里。" : "";
|
||||
if (this.proposals.list({ status: "awaiting_user_confirmation" }).length > 0) {
|
||||
return `还不能开始：还有一个任务在等发起人确认。${kept}`.trim();
|
||||
}
|
||||
if (this.active || this.proposals.list({ status: "working" }).length > 0) {
|
||||
return `还不能开始：还有一个任务正在执行。${kept}`.trim();
|
||||
}
|
||||
if (this.proposals.list({ status: "pending" }).some(owned)) {
|
||||
return `还不能开始：你有任务还在等你确认，先 finish 或直接继续说要求。${kept}`.trim();
|
||||
}
|
||||
if (this.proposals.list({ status: "pending" }).some((proposal) => proposal.pending?.workspaceDirty)) {
|
||||
return `还不能开始：有任务留下了未整理的工作区，需要先处理 dirty 工作区。${kept}`.trim();
|
||||
}
|
||||
return "还不能开始：你没有已确认并排队的任务。";
|
||||
}
|
||||
|
||||
private async confirmLocked(chatKey: string, userId: string, id?: string, answer?: string): Promise<boolean> {
|
||||
const proposal = this.resolveTargetProposal(chatKey, userId, id, ["proposed", "awaiting_user_confirmation"]);
|
||||
private async confirmLocked(chatKey: string, userId: string, id?: string): Promise<boolean> {
|
||||
const proposal = this.resolveTargetProposal(chatKey, userId, id, ["proposed"]);
|
||||
if (!proposal) return false;
|
||||
if (proposal.status === "proposed") {
|
||||
await this.proposals.update(proposal.id, { status: "queued", confirmedAt: Date.now() });
|
||||
await this.tryStartLocked(chatKey, userId);
|
||||
return true;
|
||||
}
|
||||
const pending = proposal.pending;
|
||||
if (!pending) return false;
|
||||
if (pending.kind === "success") {
|
||||
await this.terminateActiveLocked(proposal.id);
|
||||
await this.proposals.update(proposal.id, { status: "completed", pending: undefined, finishedAt: Date.now() });
|
||||
return true;
|
||||
}
|
||||
if (pending.kind === "failure") {
|
||||
await this.terminateActiveLocked(proposal.id);
|
||||
if (answer && answer.trim().toLowerCase() === "retry") {
|
||||
await this.proposals.update(proposal.id, { status: "working", pending: undefined, startedAt: Date.now() });
|
||||
this.launchWorker(proposal.id);
|
||||
} else {
|
||||
await this.proposals.update(proposal.id, { status: "failed", pending: undefined, finishedAt: Date.now() });
|
||||
}
|
||||
return true;
|
||||
}
|
||||
await this.proposals.update(proposal.id, { status: "working", pending: undefined });
|
||||
this.continueWorker(proposal.id, answer || "Confirmed. Please continue.", pending.question);
|
||||
await this.proposals.update(proposal.id, { status: "queued", confirmedAt: Date.now() });
|
||||
await this.tryStartLocked(chatKey, userId);
|
||||
return true;
|
||||
}
|
||||
|
||||
private async adjustProposalLocked(chatKey: string, userId: string, action: { id?: string; title?: string; goal?: string; steps?: string[] }): Promise<boolean> {
|
||||
const proposal = this.resolveTargetProposal(chatKey, userId, action.id, ["proposed", "queued"]);
|
||||
if (!proposal) return false;
|
||||
const patch: { title?: string; goal?: string; steps?: string[] } = {};
|
||||
if (action.title !== undefined) patch.title = action.title;
|
||||
if (action.goal !== undefined) patch.goal = action.goal;
|
||||
if (action.steps !== undefined) patch.steps = action.steps;
|
||||
if (Object.keys(patch).length === 0) return false;
|
||||
await this.proposals.update(proposal.id, patch);
|
||||
return true;
|
||||
}
|
||||
|
||||
private async followUpLocked(chatKey: string, userId: string, id: string | undefined, instruction: string, attachments?: IncomingAttachment[]): Promise<boolean> {
|
||||
const proposal = this.resolveTargetProposal(chatKey, userId, id, ["pending"]);
|
||||
if (!proposal) return false;
|
||||
if (this.active || this.proposals.list({ status: "working" }).length > 0) return false;
|
||||
await this.proposals.update(proposal.id, { status: "working", pending: undefined, startedAt: Date.now() });
|
||||
this.launchWorker(proposal.id, {
|
||||
resumeSessionId: proposal.workerNativeSessionId,
|
||||
firstTurn: workerFollowUpPrompt(instruction, proposal.pending?.question),
|
||||
freshSessionNote: "The previous worker session could not be resumed; this is a fresh session continuing the same proposal.",
|
||||
images: attachments
|
||||
});
|
||||
return true;
|
||||
}
|
||||
|
||||
private async finishLocked(chatKey: string, userId: string, id?: string, note?: string): Promise<boolean> {
|
||||
const proposal = this.resolveTargetProposal(chatKey, userId, id, ["pending"]);
|
||||
if (!proposal) return false;
|
||||
await this.terminateActiveLocked(proposal.id);
|
||||
await this.proposals.update(proposal.id, {
|
||||
status: "finished",
|
||||
finishKind: "done",
|
||||
finishNote: note,
|
||||
pending: undefined,
|
||||
finishedAt: Date.now()
|
||||
});
|
||||
return true;
|
||||
}
|
||||
|
||||
private async cancelLocked(chatKey: string, userId: string, id?: string): Promise<boolean> {
|
||||
const proposal = this.resolveTargetProposal(chatKey, userId, id, ["proposed", "queued"]);
|
||||
const proposal = this.resolveTargetProposal(chatKey, userId, id, ["proposed", "queued", "pending"]);
|
||||
if (!proposal) return false;
|
||||
await this.proposals.update(proposal.id, { status: "cancelled", finishedAt: Date.now() });
|
||||
await this.terminateActiveLocked(proposal.id);
|
||||
await this.proposals.update(proposal.id, {
|
||||
status: "finished",
|
||||
finishKind: "cancelled",
|
||||
pending: undefined,
|
||||
finishedAt: Date.now()
|
||||
});
|
||||
return true;
|
||||
}
|
||||
|
||||
private async stopActiveLocked(chatKey: string, userId: string): Promise<boolean> {
|
||||
const active = this.active;
|
||||
const activeProposal = active ? this.proposals.get(active.proposalId) : undefined;
|
||||
if (active && activeProposal && activeProposal.ownerChatKey === chatKey && activeProposal.requesterUserId === userId) {
|
||||
active.settled = true;
|
||||
this.active = undefined;
|
||||
if (active.worker.inFlight) await active.worker.cancel().catch(() => undefined);
|
||||
active.worker.terminateImmediately();
|
||||
await active.worker.terminate().catch(() => undefined);
|
||||
void active.run?.catch(() => undefined);
|
||||
await this.proposals.update(activeProposal.id, {
|
||||
status: "cancelled",
|
||||
pending: undefined,
|
||||
workerProcessGroup: undefined,
|
||||
lastWorkerSummary: "stopped by user",
|
||||
finishedAt: Date.now()
|
||||
});
|
||||
return true;
|
||||
}
|
||||
const waiting = this.resolveTargetProposal(chatKey, userId, undefined, ["awaiting_user_confirmation"]);
|
||||
if (!waiting?.pending) return false;
|
||||
await this.terminateActiveLocked(waiting.id);
|
||||
if (waiting.pending.kind === "success") {
|
||||
await this.proposals.update(waiting.id, { status: "completed", pending: undefined, finishedAt: Date.now() });
|
||||
} else if (waiting.pending.kind === "failure") {
|
||||
await this.proposals.update(waiting.id, { status: "failed", pending: undefined, finishedAt: Date.now() });
|
||||
} else {
|
||||
await this.proposals.update(waiting.id, {
|
||||
status: "cancelled",
|
||||
pending: undefined,
|
||||
lastWorkerSummary: "stopped by user",
|
||||
finishedAt: Date.now()
|
||||
});
|
||||
}
|
||||
if (!active || !activeProposal || activeProposal.ownerChatKey !== chatKey || activeProposal.requesterUserId !== userId) return false;
|
||||
active.settled = true;
|
||||
this.active = undefined;
|
||||
if (active.worker.inFlight) await active.worker.cancel().catch(() => undefined);
|
||||
active.worker.terminateImmediately();
|
||||
await active.worker.terminate().catch(() => undefined);
|
||||
void active.run?.catch(() => undefined);
|
||||
const summary = "被用户中止";
|
||||
await this.proposals.update(activeProposal.id, {
|
||||
status: "pending",
|
||||
pending: { summary, workspaceDirty: true, receivedAt: Date.now() },
|
||||
workerProcessGroup: undefined,
|
||||
lastWorkerSummary: summary
|
||||
});
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -358,7 +417,9 @@ export class AssistantManager implements ConversationRuntime {
|
||||
private async tryStartLocked(chatKey: string, userId: string): Promise<boolean> {
|
||||
if (this.active) return false;
|
||||
if (this.proposals.list({ status: "working" }).length > 0) return false;
|
||||
if (this.proposals.list({ status: "awaiting_user_confirmation" }).length > 0) return false;
|
||||
const owned = (proposal: Proposal) => proposal.ownerChatKey === chatKey && proposal.requesterUserId === userId;
|
||||
if (this.proposals.list({ status: "pending" }).some(owned)) return false;
|
||||
if (this.proposals.list({ status: "pending" }).some((proposal) => proposal.pending?.workspaceDirty)) return false;
|
||||
const next = this.proposals.list({ status: "queued" })
|
||||
.find((proposal) => proposal.ownerChatKey === chatKey && proposal.requesterUserId === userId);
|
||||
if (!next) return false;
|
||||
@@ -378,10 +439,18 @@ export class AssistantManager implements ConversationRuntime {
|
||||
.sort((left, right) => right.updatedAt - left.updatedAt)[0];
|
||||
}
|
||||
|
||||
private launchWorker(proposalId: string, resumeContext?: string): void {
|
||||
private launchWorker(proposalId: string, options: WorkerLaunchOptions = {}): void {
|
||||
const proposal = this.proposals.get(proposalId);
|
||||
if (!proposal || proposal.status !== "working") return;
|
||||
const worker = new AcpWorker(this.bot, this.config, (crashed, error) => this.onWorkerCrash(proposalId, crashed, error), {
|
||||
const active: ActiveWorker = { proposalId, worker: this.spawnProposalWorker(proposalId), settled: false };
|
||||
this.active = active;
|
||||
const run = this.startWorkerRun(active, options);
|
||||
active.run = run;
|
||||
void run.catch((error) => this.handleWorkerRunFailure(active, error));
|
||||
}
|
||||
|
||||
private spawnProposalWorker(proposalId: string): AcpWorker {
|
||||
return new AcpWorker(this.bot, this.config, (crashed, error) => this.onWorkerCrash(proposalId, crashed, error), {
|
||||
kind: "worker",
|
||||
cwd: this.bot.workspace,
|
||||
onSpawn: async (group) => {
|
||||
@@ -389,42 +458,37 @@ export class AssistantManager implements ConversationRuntime {
|
||||
await this.proposals.update(proposalId, { workerProcessGroup: group });
|
||||
}
|
||||
});
|
||||
const active: ActiveWorker = { proposalId, worker, settled: false };
|
||||
this.active = active;
|
||||
const run = this.startWorkerRun(active, resumeContext);
|
||||
active.run = run;
|
||||
void run.catch((error) => this.handleWorkerRunFailure(active, error));
|
||||
}
|
||||
|
||||
private continueWorker(proposalId: string, answer: string, question?: string): void {
|
||||
const active = this.active;
|
||||
if (active && active.proposalId === proposalId && active.worker.nativeSessionId) {
|
||||
active.settled = false;
|
||||
const run = this.runWorkerTurn(active, workerAnswerPrompt(answer, question));
|
||||
active.run = run;
|
||||
void run.catch((error) => this.handleWorkerRunFailure(active, error));
|
||||
return;
|
||||
private async startWorkerRun(active: ActiveWorker, options: WorkerLaunchOptions): Promise<void> {
|
||||
if (options.resumeSessionId) {
|
||||
try {
|
||||
await active.worker.start(options.resumeSessionId);
|
||||
} catch (error) {
|
||||
if (!(error instanceof AcpSessionRestoreError) || this.active !== active || active.settled) throw error;
|
||||
active.worker = this.spawnProposalWorker(active.proposalId);
|
||||
await active.worker.start();
|
||||
options = {
|
||||
...options,
|
||||
resumeContext: [options.resumeContext, options.freshSessionNote].filter(Boolean).join("\n"),
|
||||
firstTurn: options.firstTurn && options.freshSessionNote ? `${options.firstTurn}\n\n${options.freshSessionNote}` : options.firstTurn
|
||||
};
|
||||
}
|
||||
} else {
|
||||
await active.worker.start();
|
||||
}
|
||||
const proposal = this.proposals.get(proposalId);
|
||||
const context = proposal
|
||||
? `The previous worker was lost. Its last summary: ${proposal.lastWorkerSummary || "unknown"}. Its question: ${question || "unknown"}. The user answered: ${answer}`
|
||||
: answer;
|
||||
this.launchWorker(proposalId, context);
|
||||
}
|
||||
|
||||
private async startWorkerRun(active: ActiveWorker, resumeContext?: string): Promise<void> {
|
||||
const nativeSessionId = await active.worker.start();
|
||||
if (this.active !== active || active.settled) throw new Error("Worker was superseded before its session started");
|
||||
await this.proposals.update(active.proposalId, { workerNativeSessionId: nativeSessionId });
|
||||
await this.proposals.update(active.proposalId, { workerNativeSessionId: active.worker.nativeSessionId });
|
||||
await active.worker.prompt(this.bot.workerBootstrap, "initializing");
|
||||
if (this.active !== active || active.settled) throw new Error("Worker was superseded during bootstrap");
|
||||
const proposal = this.proposals.get(active.proposalId);
|
||||
if (!proposal) throw new Error("Worker proposal no longer exists");
|
||||
await this.runWorkerTurn(active, workerTaskPrompt(proposal, resumeContext));
|
||||
const firstTurn = options.firstTurn || workerTaskPrompt(proposal, options.resumeContext);
|
||||
await this.runWorkerTurn(active, firstTurn, options.images);
|
||||
}
|
||||
|
||||
private async runWorkerTurn(active: ActiveWorker, promptText: string): Promise<void> {
|
||||
const reply = await active.worker.prompt(withWorkerResultProtocol(promptText));
|
||||
private async runWorkerTurn(active: ActiveWorker, promptText: string, images?: IncomingAttachment[]): Promise<void> {
|
||||
const reply = await active.worker.prompt(withWorkerResultProtocol(this.workerPromptContent(active.worker, promptText, images)));
|
||||
if (this.active !== active || active.settled) throw new Error("Worker was superseded during its turn");
|
||||
let result = parseWorkerResult(reply);
|
||||
if (!result) {
|
||||
@@ -432,10 +496,19 @@ export class AssistantManager implements ConversationRuntime {
|
||||
if (this.active !== active || active.settled) throw new Error("Worker was superseded during result repair");
|
||||
result = parseWorkerResult(repaired);
|
||||
}
|
||||
if (!result) throw new Error("Worker did not return a valid GORI_WORKER_RESULT_V1 envelope after one repair attempt");
|
||||
if (!result) throw new Error("Worker did not return a valid GORI_WORKER_RESULT_V2 envelope after one repair attempt");
|
||||
await this.settleWorkerResult(active, result);
|
||||
}
|
||||
|
||||
private workerPromptContent(worker: AcpWorker, text: string, images?: IncomingAttachment[]): string | AcpPromptContent[] {
|
||||
if (!images || images.length === 0) return text;
|
||||
if (!worker.supportsImages) return `${text}\n\n[Attachments]\n${images.map(() => "[图片，当前 agent 不支持图片输入]").join("\n")}`;
|
||||
return [
|
||||
...images.map((image) => ({ type: "image" as const, data: image.data, mimeType: image.mimeType })),
|
||||
{ type: "text" as const, text }
|
||||
];
|
||||
}
|
||||
|
||||
private async settleWorkerResult(active: ActiveWorker, result: ParsedWorkerResult): Promise<void> {
|
||||
if (this.active !== active || active.settled || this.shuttingDown) return;
|
||||
const transitioned = await this.scheduler(async () => {
|
||||
@@ -443,21 +516,20 @@ export class AssistantManager implements ConversationRuntime {
|
||||
const proposal = this.proposals.get(active.proposalId);
|
||||
if (!proposal || proposal.status !== "working") { active.settled = true; return false; }
|
||||
active.settled = true;
|
||||
const pending: ProposalPending = result.status === "SUCCESS"
|
||||
? { kind: "success", summary: result.summary }
|
||||
: result.status === "FAILED"
|
||||
? { kind: "failure", summary: result.summary }
|
||||
: { kind: "step", summary: result.summary, question: result.question, nextStep: result.nextStep };
|
||||
this.active = undefined;
|
||||
const pending: ProposalPending = {
|
||||
summary: result.summary,
|
||||
question: result.question,
|
||||
workspaceDirty: result.workspaceDirty,
|
||||
receivedAt: Date.now()
|
||||
};
|
||||
await this.proposals.update(active.proposalId, {
|
||||
status: "awaiting_user_confirmation",
|
||||
status: "pending",
|
||||
pending,
|
||||
lastWorkerSummary: result.summary
|
||||
});
|
||||
if (result.status !== "NEEDS_CONFIRMATION") {
|
||||
this.active = undefined;
|
||||
await active.worker.terminate().catch(() => undefined);
|
||||
await this.proposals.update(active.proposalId, { workerProcessGroup: undefined });
|
||||
}
|
||||
await active.worker.terminate().catch(() => undefined);
|
||||
await this.proposals.update(active.proposalId, { workerProcessGroup: undefined });
|
||||
return true;
|
||||
});
|
||||
if (transitioned) await this.notifyOwner(active.proposalId);
|
||||
@@ -475,8 +547,8 @@ export class AssistantManager implements ConversationRuntime {
|
||||
this.active = undefined;
|
||||
const summary = `worker_error: ${error instanceof Error ? error.message : String(error)}`;
|
||||
await this.proposals.update(active.proposalId, {
|
||||
status: "awaiting_user_confirmation",
|
||||
pending: { kind: "failure", summary },
|
||||
status: "pending",
|
||||
pending: { summary, workspaceDirty: true, receivedAt: Date.now() },
|
||||
lastWorkerSummary: summary,
|
||||
workerProcessGroup: undefined
|
||||
});
|
||||
@@ -646,6 +718,9 @@ export class AssistantManager implements ConversationRuntime {
|
||||
"[Proposal states]",
|
||||
...this.proposalSummaries(chatKey, request.userId),
|
||||
"",
|
||||
"[Pending proposals]",
|
||||
...this.pendingCards(chatKey, request.userId),
|
||||
"",
|
||||
"[Scheduler state]",
|
||||
this.schedulerStateSummary(chatKey, request.userId),
|
||||
"",
|
||||
@@ -654,15 +729,26 @@ export class AssistantManager implements ConversationRuntime {
|
||||
].join("\n");
|
||||
}
|
||||
|
||||
private pendingCards(chatKey: string, userId: string): string[] {
|
||||
const pending = this.proposals.list({ status: "pending" })
|
||||
.filter((proposal) => proposal.ownerChatKey === chatKey && proposal.requesterUserId === userId);
|
||||
if (pending.length === 0) return ["none"];
|
||||
return pending.map((proposal) => {
|
||||
const card = proposal.pending!;
|
||||
return `- id=${proposal.id} title=${JSON.stringify(proposal.title)} summary=${JSON.stringify(card.summary)}${card.question ? ` question=${JSON.stringify(card.question)}` : ""}${card.workspaceDirty ? " workspaceDirty=true" : ""}`;
|
||||
});
|
||||
}
|
||||
|
||||
private schedulerStateSummary(chatKey: string, userId: string): string {
|
||||
const owned = (proposal: Proposal) => proposal.ownerChatKey === chatKey && proposal.requesterUserId === userId;
|
||||
const awaitingOther = this.proposals.list({ status: "awaiting_user_confirmation" }).some((proposal) => !owned(proposal));
|
||||
if (awaitingOther) return "blocked: another proposal is awaiting owner confirmation";
|
||||
const active = this.active;
|
||||
const activeProposal = active ? this.proposals.get(active.proposalId) : undefined;
|
||||
const workingOther = this.proposals.list({ status: "working" }).some((proposal) => !owned(proposal))
|
||||
|| Boolean(activeProposal && !owned(activeProposal));
|
||||
if (workingOther) return "busy: another proposal is working";
|
||||
const dirtyOther = this.proposals.list({ status: "pending" })
|
||||
.some((proposal) => !owned(proposal) && proposal.pending?.workspaceDirty);
|
||||
if (dirtyOther) return "blocked: another proposal left a dirty workspace that needs attention";
|
||||
return "idle";
|
||||
}
|
||||
|
||||
@@ -672,7 +758,7 @@ export class AssistantManager implements ConversationRuntime {
|
||||
if (proposals.length === 0) return ["none"];
|
||||
return proposals.map((proposal) => {
|
||||
const pending = proposal.pending
|
||||
? ` pending=${proposal.pending.kind} summary=${JSON.stringify(proposal.pending.summary)}${proposal.pending.question ? ` question=${JSON.stringify(proposal.pending.question)}` : ""}`
|
||||
? ` summary=${JSON.stringify(proposal.pending.summary)}${proposal.pending.question ? ` question=${JSON.stringify(proposal.pending.question)}` : ""}`
|
||||
: "";
|
||||
return `- id=${proposal.id} status=${proposal.status} title=${JSON.stringify(proposal.title)}${pending}`;
|
||||
});
|
||||
@@ -733,8 +819,30 @@ function parseAssistantAction(value: unknown): AssistantAction | undefined {
|
||||
return { type: "create_proposal", title: value.title, goal: value.goal, steps: value.steps as string[] };
|
||||
}
|
||||
if (value.type === "confirm") {
|
||||
if ((value.id !== undefined && typeof value.id !== "string") || (value.answer !== undefined && typeof value.answer !== "string")) return undefined;
|
||||
return { type: "confirm", id: value.id as string | undefined, answer: value.answer as string | undefined };
|
||||
if (value.id !== undefined && typeof value.id !== "string") return undefined;
|
||||
return { type: "confirm", id: value.id as string | undefined };
|
||||
}
|
||||
if (value.type === "adjust_proposal") {
|
||||
if ((value.id !== undefined && typeof value.id !== "string")
|
||||
|| (value.title !== undefined && (typeof value.title !== "string" || value.title.length === 0))
|
||||
|| (value.goal !== undefined && (typeof value.goal !== "string" || value.goal.length === 0))
|
||||
|| (value.steps !== undefined && (!Array.isArray(value.steps) || value.steps.some((step) => typeof step !== "string" || step.length === 0)))) return undefined;
|
||||
return {
|
||||
type: "adjust_proposal",
|
||||
id: value.id as string | undefined,
|
||||
title: value.title as string | undefined,
|
||||
goal: value.goal as string | undefined,
|
||||
steps: value.steps as string[] | undefined
|
||||
};
|
||||
}
|
||||
if (value.type === "follow_up") {
|
||||
if (typeof value.instruction !== "string" || value.instruction.length === 0
|
||||
|| (value.id !== undefined && typeof value.id !== "string")) return undefined;
|
||||
return { type: "follow_up", id: value.id as string | undefined, instruction: value.instruction };
|
||||
}
|
||||
if (value.type === "finish") {
|
||||
if ((value.id !== undefined && typeof value.id !== "string") || (value.note !== undefined && typeof value.note !== "string")) return undefined;
|
||||
return { type: "finish", id: value.id as string | undefined, note: value.note as string | undefined };
|
||||
}
|
||||
if (value.type === "start_next") return { type: "start_next" };
|
||||
if (value.type === "cancel") {
|
||||
@@ -751,29 +859,29 @@ export function parseWorkerResult(text: string): ParsedWorkerResult | undefined
|
||||
let value: unknown;
|
||||
try { value = JSON.parse(match.groups.json); } catch { return undefined; }
|
||||
if (!isRecord(value) || typeof value.summary !== "string") return undefined;
|
||||
if (value.status !== "SUCCESS" && value.status !== "NEEDS_CONFIRMATION" && value.status !== "FAILED") return undefined;
|
||||
if (value.status !== "PENDING") return undefined;
|
||||
if ((value.question !== undefined && typeof value.question !== "string")
|
||||
|| (value.nextStep !== undefined && typeof value.nextStep !== "string")
|
||||
|| (value.dirty !== undefined && typeof value.dirty !== "boolean")) return undefined;
|
||||
|| (value.workspaceDirty !== undefined && typeof value.workspaceDirty !== "boolean")) return undefined;
|
||||
return {
|
||||
status: value.status,
|
||||
status: "PENDING",
|
||||
summary: value.summary,
|
||||
question: value.question as string | undefined,
|
||||
nextStep: value.nextStep as string | undefined,
|
||||
dirty: value.dirty as boolean | undefined
|
||||
workspaceDirty: value.workspaceDirty as boolean | undefined
|
||||
};
|
||||
}
|
||||
|
||||
function withWorkerResultProtocol(text: string): string {
|
||||
return `${text}\n\nWhen this turn is finished, end your response with exactly one hidden worker result envelope. Use <GORI_WORKER_RESULT_V1>{"status":"SUCCESS","summary":"..."}</GORI_WORKER_RESULT_V1> only when the proposal is fully complete, status "FAILED" with a summary when it cannot be completed, or status "NEEDS_CONFIRMATION" with a "question" when user confirmation is required before continuing. Optional fields: "nextStep", "dirty". Do not emit any other status value or any text after the envelope.`;
|
||||
function withWorkerResultProtocol(input: string | AcpPromptContent[]): string | AcpPromptContent[] {
|
||||
const instruction = "When this turn is finished, end your response with exactly one hidden worker result envelope: <GORI_WORKER_RESULT_V2>{\"status\":\"PENDING\",\"summary\":\"...\"}</GORI_WORKER_RESULT_V2>. The summary is a short user-readable description of what you did or what is blocking you. Add a \"question\" when you need the user's decision before continuing, and set \"workspaceDirty\": true when you left the workspace modified or are unsure about its state. PENDING is the only status; do not emit any other status value or any text after the envelope.";
|
||||
if (typeof input === "string") return `${input}\n\n${instruction}`;
|
||||
return [...input, { type: "text" as const, text: `\n\n${instruction}` }];
|
||||
}
|
||||
|
||||
function workerRepairPrompt(): string {
|
||||
return "Your previous response did not end with a valid GORI_WORKER_RESULT_V1 envelope. Return exactly one valid envelope with status SUCCESS, NEEDS_CONFIRMATION, or FAILED and a summary. Do not perform more work.";
|
||||
return "Your previous response did not end with a valid GORI_WORKER_RESULT_V2 envelope. Return exactly one valid envelope with status PENDING and a summary. Do not perform more work.";
|
||||
}
|
||||
|
||||
function assistantRepairPrompt(): string {
|
||||
return "Your previous response did not end with a valid GORI_ASSISTANT_ACTION_V1 envelope. Reply again with the user-facing text in the JSON \"reply\" field and an \"actions\" array (empty if no state change is needed).";
|
||||
return "Your previous response did not end with a valid GORI_ASSISTANT_ACTION_V2 envelope. Reply again with the user-facing text in the JSON \"reply\" field and an \"actions\" array (empty if no state change is needed).";
|
||||
}
|
||||
|
||||
function workerTaskPrompt(proposal: Proposal, resumeContext?: string): string {
|
||||
@@ -785,15 +893,15 @@ function workerTaskPrompt(proposal: Proposal, resumeContext?: string): string {
|
||||
...proposal.steps.map((step, index) => `${index + 1}. ${step}`),
|
||||
...(resumeContext ? ["", "Additional context:", resumeContext] : []),
|
||||
"",
|
||||
"If you encounter a dirty or unexpected target, or anything that requires the user's decision, stop and report NEEDS_CONFIRMATION with a clear question instead of forcing the change."
|
||||
"If you encounter a dirty or unexpected target, or anything that requires the user's decision, stop and report PENDING with a clear question instead of forcing the change."
|
||||
].join("\n");
|
||||
}
|
||||
|
||||
function workerAnswerPrompt(answer: string, question?: string): string {
|
||||
function workerFollowUpPrompt(instruction: string, question?: string): string {
|
||||
return [
|
||||
"The user answered your question about the current proposal.",
|
||||
`Your question was: ${question || "unknown"}`,
|
||||
`User answer: ${answer}`,
|
||||
"The user sent a follow-up instruction for the current proposal.",
|
||||
question ? `Your last question was: ${question}` : "You had no open question.",
|
||||
`User instruction: ${instruction}`,
|
||||
"Continue executing the same proposal accordingly."
|
||||
].join("\n");
|
||||
}
|
||||
@@ -802,18 +910,16 @@ function assistantEventPrompt(proposal: Proposal): string {
|
||||
const pending = proposal.pending!;
|
||||
return [
|
||||
"[Internal event from gori-agent. This is not a user message.]",
|
||||
`The worker for proposal "${proposal.title}" (id ${proposal.id}) reported ${pending.kind === "step" ? "NEEDS_CONFIRMATION" : pending.kind === "success" ? "SUCCESS" : "FAILED"}.`,
|
||||
`The worker for proposal "${proposal.title}" (id ${proposal.id}) reported its result and the proposal is now waiting for the user's decision.`,
|
||||
`Summary: ${pending.summary}`,
|
||||
pending.question ? `Question for the user: ${pending.question}` : "Question for the user: none",
|
||||
pending.nextStep ? `Suggested next step: ${pending.nextStep}` : "Suggested next step: none",
|
||||
"Tell the user, in your own words, what happened and what they can do next (confirm, answer the question, retry, or stop). End with the usual GORI_ASSISTANT_ACTION_V1 envelope; its actions array MUST be empty because actions are ignored for internal events."
|
||||
"Tell the user, in your own words, what happened and what they can do next (finish to close the task, follow up with new instructions to continue it, or stop). End with the usual GORI_ASSISTANT_ACTION_V2 envelope; its actions array MUST be empty because actions are ignored for internal events."
|
||||
].join("\n");
|
||||
}
|
||||
|
||||
function fallbackEventText(proposal: Proposal): string {
|
||||
const pending = proposal.pending!;
|
||||
const state = pending.kind === "step" ? "等待你回答一个问题" : pending.kind === "success" ? "执行完成，待你确认" : "执行失败，待你确认或重试";
|
||||
return `任务「${proposal.title}」${state}。摘要：${pending.summary}${pending.question ? ` 问题：${pending.question}` : ""}`;
|
||||
return `任务「${proposal.title}」等你确认。摘要：${pending.summary}${pending.question ? ` 问题：${pending.question}` : ""} 说 finish 结束，或直接说要求继续改。`;
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
|
||||
Reference in New Issue
Block a user