feat(agent): automation-first sessions with step plan, option choice and inline approval
- agent.plan emits plan_updated; system prompt asks for a plan first and updates while executing - agent.propose_options registers a decision and suspends the run (waitForFact, no worker slot); the answer or timeout (recommended/stop) resumes it as the tool result - console-session runs get inline approvals for plan-bound capabilities: approve replays the call with the approval, reject fails it with the reason and records it on the decision - proposeFirst blocks execute capabilities until an option is chosen; E12 covers all paths Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
rootandClaude Opus 5.5 committed
1 parent
74c2ed45aa
commit
cf2c4ff0e6
14 files changed
+1360
-26
No files matched your search
@@ -3,6 +3,7 @@ import type { AuthorizationChecker, AuthorizationDecision } from "@umt/agent";
|
||||
import type { CapabilitySpec } from "@umt/contracts";
|
||||
import { type DB, type Kysely, sql } from "@umt/db";
|
||||
import type { SettingsReader } from "../types.ts";
|
||||
import { inlineApproval, proposeFirstBlock } from "./interaction.ts";
|
||||
import { exceedsAncestors } from "./subagents.ts";
|
||||
|
||||
// 服务端授权(REQ §10.2、§10.3):propose/approve/execute 类能力必须命中
|
||||
@@ -127,6 +128,17 @@ export function createAuthorizer(deps: {
|
||||
? { allowed: true }
|
||||
: { allowed: false, reason: "只有 Agent 会话可以派生子会话" };
|
||||
}
|
||||
// agent.propose_options 只登记一个待用户选择的事项(无外部副作用);能否问用户由处理器校验
|
||||
if (d.id === "agent.propose_options") {
|
||||
return runId && (ctx.source === "agent" || ctx.source === "task")
|
||||
? { allowed: true }
|
||||
: { allowed: false, reason: "只有 Agent 会话可以让用户选方案" };
|
||||
}
|
||||
// 先看方案再执行:用户选定方案前,Agent 的执行类调用一律拒绝
|
||||
if (runId && d.kind === "execute" && (ctx.source === "agent" || ctx.source === "task")) {
|
||||
const block = await proposeFirstBlock(db, runId);
|
||||
if (block) return { allowed: false, reason: block };
|
||||
}
|
||||
|
||||
// 1) 单次审批
|
||||
const ap = APPROVAL_PLAN[d.id];
|
||||
@@ -207,6 +219,12 @@ export function createAuthorizer(deps: {
|
||||
}
|
||||
return { allowed: true, grantId: g.id };
|
||||
}
|
||||
// 3) 会话内审批:控制台管理员会话里,没有持续授权的计划类动作在步骤处发起审批并挂起会话
|
||||
if (ap && target && !approvalId) {
|
||||
const planId = inp[ap.field] as string;
|
||||
const inline = await inlineApproval(db, d, { planType: ap.planType, planId, binding: target }, ctx, now());
|
||||
if (inline) return inline;
|
||||
}
|
||||
return {
|
||||
allowed: false,
|
||||
reason: reasons.length > 0 ? `没有可用的授权:${reasons.join(";")}` : `缺少 ${d.id} 的持续授权或有效的单次审批`,
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import type { ServerContext, ServerModule } from "../types.ts";
|
||||
import { registerTaskCapabilities } from "./capabilities.ts";
|
||||
import { registerDownloadRecoveryCapabilities } from "./capabilities-download.ts";
|
||||
import { registerInteractionCapabilities } from "./interaction.ts";
|
||||
import { registerAgentRoutes } from "./routes.ts";
|
||||
import type { AgentEnv, AgentModuleOptions } from "./runner.ts";
|
||||
import { startRun } from "./service.ts";
|
||||
@@ -38,6 +39,7 @@ export function createAgentModule(opts: AgentModuleOptions = {}): ServerModule &
|
||||
registerCapabilities(registry, ctx) {
|
||||
registerTaskCapabilities(registry, ctx);
|
||||
registerDownloadRecoveryCapabilities(registry, ctx, opts);
|
||||
registerInteractionCapabilities(registry, ctx);
|
||||
registerSubagentCapabilities(registry, ctx, {
|
||||
start: (p) =>
|
||||
startRun(ctx, {
|
||||
|
||||
@@ -0,0 +1,692 @@
|
||||
import {
|
||||
type AuthorizationDecision,
|
||||
CapabilityError,
|
||||
type CapabilityRegistry,
|
||||
type HandlerContext,
|
||||
type InvokeContext,
|
||||
redactString,
|
||||
} from "@umt/agent";
|
||||
import {
|
||||
type AgentOptionItem,
|
||||
type AgentOptionsTimeoutAction,
|
||||
type AgentPlanStep,
|
||||
type AgentProposeOptionsOutput,
|
||||
CAPABILITY_CATALOG,
|
||||
type CapabilitySpec,
|
||||
capabilitySpec,
|
||||
type DecisionOption,
|
||||
type DecisionStatus,
|
||||
type KnownCapabilityId,
|
||||
} from "@umt/contracts";
|
||||
import { type AgentRunRow, type DB, type Kysely, sql } from "@umt/db";
|
||||
import type { ServerContext } from "../types.ts";
|
||||
import type { PlanBinding } from "./authorize.ts";
|
||||
import {
|
||||
AwaitSuspendedError,
|
||||
type AwaitSuspension,
|
||||
ancestors,
|
||||
type LiveRun,
|
||||
liveRun,
|
||||
suspendRun,
|
||||
wakeRun,
|
||||
} from "./subagents.ts";
|
||||
|
||||
// AG-2 Agent 交互改为自动化优先(REQ §10、R29):
|
||||
// - 不再区分 plan_only / execute:每个会话默认在授权范围内执行;"先看方案再执行"是会话选项(scope.proposeFirst),
|
||||
// 选定方案前服务端拒绝有副作用的执行类能力。
|
||||
// - agent.plan:模型维护步骤清单(plan_updated 事件),会话页顶部实时勾选。
|
||||
// - agent.propose_options:给用户 2–3 个方案。登记一个待处理事项(dedupe agent-options:<run>:<invocation>)后
|
||||
// 会话挂起(runAgent waitForFact,不占 worker 槽位、不调用模型);用户在会话页或待处理中心作答、或超时,
|
||||
// 会话恢复并重放这次调用,把选择作为工具结果返回模型。
|
||||
// - 内联审批:控制台管理员会话(scope.interactive)里,需单次审批的能力(download.submit / organize.execute /
|
||||
// policy.apply)没有持续授权时,授权器在对应步骤建审批 + 待处理事项并挂起会话;管理员批准 / 拒绝后恢复,
|
||||
// 重放的调用按审批结果放行或以拒绝理由失败。审批绑定 plan id + payload hash + revision + 有效期,只用一次。
|
||||
|
||||
const OPTIONS_KEY = "agent-options";
|
||||
const APPROVAL_KEY = "agent-approval";
|
||||
|
||||
/** agent_run.scope 里的会话选项 */
|
||||
export interface RunSessionOptions {
|
||||
/** 控制台管理员会话发起:需审批的动作在步骤处出审批卡(否则按缺少授权直接拒绝) */
|
||||
interactive?: boolean;
|
||||
/** 先看方案再执行 */
|
||||
proposeFirst?: boolean;
|
||||
}
|
||||
|
||||
export const sessionOptions = (run: Pick<AgentRunRow, "scope">): RunSessionOptions =>
|
||||
(run.scope ?? {}) as RunSessionOptions;
|
||||
|
||||
/** 可以让用户选方案的会话:用户发起的顶层会话(子会话 / 后台会话自己决定) */
|
||||
export const canAskUser = (run: Pick<AgentRunRow, "trigger" | "parent_run_id">) =>
|
||||
run.trigger === "user" && !run.parent_run_id;
|
||||
|
||||
/** 会话内审批:控制台管理员会话发起的顶层会话 */
|
||||
export const canRequestApproval = (run: Pick<AgentRunRow, "trigger" | "parent_run_id" | "scope">) =>
|
||||
canAskUser(run) && sessionOptions(run).interactive === true;
|
||||
|
||||
// 授权器(capability-registry 在 context 装配时创建)拿不到 ServerContext:模块装配时按 db 绑定
|
||||
const bound = new WeakMap<object, ServerContext>();
|
||||
export function bindInteraction(ctx: ServerContext) {
|
||||
bound.set(ctx.db, ctx);
|
||||
}
|
||||
|
||||
const clip = (s: string, n: number) => (s.length > n ? `${s.slice(0, n)}…` : s);
|
||||
const safe = (s: string | null | undefined, n = 500) => clip(redactString(s ?? ""), n);
|
||||
const isUuid = (s: string) => /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(s);
|
||||
|
||||
async function runOf(ctx: ServerContext, c: HandlerContext | InvokeContext): Promise<AgentRunRow> {
|
||||
const id = c.agentRunId && isUuid(c.agentRunId) ? c.agentRunId : null;
|
||||
if (!id) throw new CapabilityError("precondition", "只有 Agent 会话可以使用这个能力");
|
||||
const run = await ctx.db.selectFrom("agent_run").selectAll().where("id", "=", id).executeTakeFirst();
|
||||
if (!run) throw new CapabilityError("not_found", "Agent 会话不存在");
|
||||
return run;
|
||||
}
|
||||
|
||||
// ── 步骤清单 ──
|
||||
|
||||
/** 最近一次 plan_updated(没有 = 空清单) */
|
||||
export async function loadPlan(db: Kysely<DB>, runId: string): Promise<{ steps: AgentPlanStep[]; revision: number }> {
|
||||
const r = await db
|
||||
.selectFrom("agent_run_event")
|
||||
.select("payload")
|
||||
.where("run_id", "=", runId)
|
||||
.where("type", "=", "plan_updated")
|
||||
.orderBy("seq", "desc")
|
||||
.executeTakeFirst();
|
||||
const p = (r?.payload ?? null) as { steps?: AgentPlanStep[]; revision?: number } | null;
|
||||
return { steps: p?.steps ?? [], revision: p?.revision ?? 0 };
|
||||
}
|
||||
|
||||
async function currentPlan(ctx: ServerContext, live: LiveRun | undefined, runId: string) {
|
||||
if (live?.plan) return live.plan;
|
||||
const p = await loadPlan(ctx.db, runId);
|
||||
if (live) live.plan = p;
|
||||
return p;
|
||||
}
|
||||
|
||||
/** 当前正在进行的步骤(审批卡 / 方案卡挂在它下面);没有进行中的取第一个未完成的 */
|
||||
async function currentStepId(ctx: ServerContext, live: LiveRun | undefined, runId: string): Promise<string | null> {
|
||||
const { steps } = await currentPlan(ctx, live, runId);
|
||||
return (
|
||||
steps.find((s) => s.status === "running")?.id ??
|
||||
steps.find((s) => s.status === "pending")?.id ??
|
||||
steps.at(-1)?.id ??
|
||||
null
|
||||
);
|
||||
}
|
||||
|
||||
// ── 待处理事项(方案选择 / 审批各一个,用户也能在待处理中心处理) ──
|
||||
|
||||
async function insertDecision(
|
||||
ctx: ServerContext,
|
||||
d: {
|
||||
kind: "other" | "approval_required";
|
||||
title: string;
|
||||
summary: string;
|
||||
runId: string;
|
||||
dedupeKey: string;
|
||||
evidence: object;
|
||||
options: DecisionOption[];
|
||||
},
|
||||
): Promise<{ id: string; status: DecisionStatus; evidence: unknown; resolution: unknown }> {
|
||||
return ctx.db.transaction().execute(async (trx) => {
|
||||
const r = await sql<{
|
||||
id: string;
|
||||
status: DecisionStatus;
|
||||
evidence: unknown;
|
||||
resolution: unknown;
|
||||
inserted: boolean;
|
||||
}>`
|
||||
insert into decision (kind, status, severity, title, summary, subject_type, subject_id, dedupe_key, evidence,
|
||||
options, agent_run_id)
|
||||
values (${d.kind}, 'awaiting_user', 'info', ${d.title}, ${d.summary}, 'agent_run', ${d.runId}, ${d.dedupeKey},
|
||||
${JSON.stringify(d.evidence)}::jsonb, ${JSON.stringify(d.options)}::jsonb, ${d.runId})
|
||||
on conflict (dedupe_key) where status in ('open', 'investigating', 'awaiting_user')
|
||||
do update set updated_at = decision.updated_at
|
||||
returning id, status, evidence, resolution, (xmax = 0) as inserted`.execute(trx);
|
||||
const row = r.rows[0]!;
|
||||
if (row.inserted)
|
||||
await ctx.outbox.emit(
|
||||
trx,
|
||||
"decision.created",
|
||||
{ decisionId: row.id, kind: d.kind, severity: "info", title: d.title },
|
||||
{ idempotencyKey: `decision.created:${row.id}` },
|
||||
);
|
||||
return row;
|
||||
});
|
||||
}
|
||||
|
||||
async function closeDecision(
|
||||
ctx: ServerContext,
|
||||
id: string,
|
||||
status: "resolved" | "dismissed" | "expired",
|
||||
resolution: Record<string, unknown>,
|
||||
by: string,
|
||||
): Promise<boolean> {
|
||||
return ctx.db.transaction().execute(async (trx) => {
|
||||
const r = await trx
|
||||
.updateTable("decision")
|
||||
.set({
|
||||
status,
|
||||
resolution: sql`coalesce(resolution, '{}'::jsonb) || ${JSON.stringify(resolution)}::jsonb`,
|
||||
resolved_by: by,
|
||||
resolved_at: new Date(),
|
||||
})
|
||||
.where("id", "=", id)
|
||||
.where("status", "in", ["open", "investigating", "awaiting_user"])
|
||||
.returning("id")
|
||||
.executeTakeFirst();
|
||||
if (r)
|
||||
await ctx.outbox.emit(
|
||||
trx,
|
||||
"decision.updated",
|
||||
{ decisionId: id, status },
|
||||
{ idempotencyKey: `decision.updated:${id}:${status}` },
|
||||
);
|
||||
return !!r;
|
||||
});
|
||||
}
|
||||
|
||||
// ── 方案选择 ──
|
||||
|
||||
interface OptionsEvidence {
|
||||
source: "agent.propose_options";
|
||||
runId: string;
|
||||
invocationId: string;
|
||||
question: string;
|
||||
options: AgentOptionItem[];
|
||||
recommendedId: string | null;
|
||||
expiresAt: string;
|
||||
timeoutAction: AgentOptionsTimeoutAction;
|
||||
}
|
||||
interface OptionsResolution {
|
||||
optionKey?: string | null;
|
||||
outcome?: "chosen" | "rejected" | "stop";
|
||||
reason?: string | null;
|
||||
note?: string | null;
|
||||
by?: string;
|
||||
timeout?: boolean;
|
||||
}
|
||||
|
||||
function optionsResult(
|
||||
ev: OptionsEvidence,
|
||||
status: DecisionStatus,
|
||||
res: OptionsResolution,
|
||||
by: "user" | "timeout",
|
||||
): AgentProposeOptionsOutput {
|
||||
const choice = res.optionKey ?? null;
|
||||
const option = choice ? (ev.options.find((o) => o.id === choice) ?? null) : null;
|
||||
const outcome = res.outcome ?? (status === "resolved" && option ? "chosen" : "rejected");
|
||||
return {
|
||||
outcome: option && outcome === "chosen" ? "chosen" : outcome === "stop" ? "stop" : "rejected",
|
||||
choice: option ? choice : null,
|
||||
option,
|
||||
reason: safe(res.reason ?? res.note ?? null, 1000) || null,
|
||||
by,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* 用户回答方案选择(会话页或 API):选一个方案或都不要。写待处理事项后唤醒挂起的会话。
|
||||
* 返回 false = 事项不属于这个会话 / 已处理。
|
||||
*/
|
||||
export async function answerOptions(
|
||||
ctx: ServerContext,
|
||||
runId: string,
|
||||
input: { decisionId: string; optionId: string | null; reason?: string | undefined },
|
||||
by: string,
|
||||
): Promise<"ok" | "not_found" | "closed" | "bad_option"> {
|
||||
const d = await ctx.db
|
||||
.selectFrom("decision")
|
||||
.select(["id", "status", "evidence", "agent_run_id", "dedupe_key"])
|
||||
.where("id", "=", input.decisionId)
|
||||
.executeTakeFirst();
|
||||
if (!d || d.agent_run_id !== runId || !d.dedupe_key.startsWith(`${OPTIONS_KEY}:`)) return "not_found";
|
||||
const ev = d.evidence as unknown as OptionsEvidence;
|
||||
if (input.optionId !== null && !ev.options?.some((o) => o.id === input.optionId)) return "bad_option";
|
||||
const closed = await closeDecision(
|
||||
ctx,
|
||||
d.id,
|
||||
input.optionId ? "resolved" : "dismissed",
|
||||
{
|
||||
optionKey: input.optionId,
|
||||
outcome: input.optionId ? "chosen" : "rejected",
|
||||
reason: input.reason ?? null,
|
||||
by,
|
||||
},
|
||||
by,
|
||||
);
|
||||
if (!closed) return "closed";
|
||||
await wakeRun(ctx, runId, { decisionId: d.id });
|
||||
return "ok";
|
||||
}
|
||||
|
||||
// ── 审批 ──
|
||||
|
||||
interface ApprovalScope {
|
||||
agentRunId: string;
|
||||
invocationId: string;
|
||||
capabilityId: string;
|
||||
stepId: string | null;
|
||||
summary: string;
|
||||
}
|
||||
|
||||
/** 计划摘要(审批卡标题下那行):作品 / 资源 / 体积,或整理 / 策略的名字 */
|
||||
async function planSummary(db: Kysely<DB>, planType: string, planId: string): Promise<string> {
|
||||
if (planType === "acquisition_plan") {
|
||||
const r = await db
|
||||
.selectFrom("acquisition_plan as p")
|
||||
.innerJoin("acquisition_slot as s", "s.id", "p.slot_id")
|
||||
.innerJoin("media_entity as e", "e.id", "s.media_entity_id")
|
||||
.leftJoin("source_release as r", "r.id", "p.source_release_id")
|
||||
.select(["e.title as entity", "r.title as release", "p.reserved_bytes"])
|
||||
.where("p.id", "=", planId)
|
||||
.executeTakeFirst();
|
||||
if (r) {
|
||||
const gb = Number(r.reserved_bytes ?? 0) / 1024 ** 3;
|
||||
return safe(`下载《${r.entity}》:${r.release ?? "资源"}${gb > 0 ? `(${gb.toFixed(1)} GB)` : ""}`, 300);
|
||||
}
|
||||
}
|
||||
if (planType === "organize_plan") {
|
||||
const r = await db.selectFrom("artifact").select("title").where("id", "=", planId).executeTakeFirst();
|
||||
if (r) return safe(`整理:${r.title}`, 300);
|
||||
}
|
||||
if (planType === "policy_revision") {
|
||||
const r = await db.selectFrom("policy_revision").select("revision").where("id", "=", planId).executeTakeFirst();
|
||||
if (r) return `发布策略版本 r${r.revision}`;
|
||||
}
|
||||
return `${planType} ${planId.slice(0, 8)}`;
|
||||
}
|
||||
|
||||
async function alreadyEmitted(db: Kysely<DB>, runId: string, type: string, key: string, value: string) {
|
||||
const r = await sql<{ n: number }>`
|
||||
select count(*)::int as n from agent_run_event
|
||||
where run_id = ${runId} and type = ${type} and payload->>${key} = ${value}`.execute(db);
|
||||
return (r.rows[0]?.n ?? 0) > 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* 授权器在"没有持续授权、输入里也没有审批"时调用:控制台管理员会话里建审批并挂起会话,恢复后按审批结果放行 / 拒绝。
|
||||
* 返回 null = 不是可内联审批的调用(按原来的缺少授权处理)。
|
||||
*/
|
||||
export async function inlineApproval(
|
||||
db: Kysely<DB>,
|
||||
d: CapabilitySpec,
|
||||
plan: { planType: string; planId: string; binding: PlanBinding },
|
||||
ctx: InvokeContext,
|
||||
now: Date,
|
||||
): Promise<AuthorizationDecision | null> {
|
||||
const sctx = bound.get(db);
|
||||
const runId = ctx.agentRunId && isUuid(ctx.agentRunId) ? ctx.agentRunId : null;
|
||||
const inv = ctx.invocationId;
|
||||
if (!sctx || ctx.source !== "agent" || !runId || !inv) return null;
|
||||
const run = await db.selectFrom("agent_run").selectAll().where("id", "=", runId).executeTakeFirst();
|
||||
if (!run || !canRequestApproval(run)) return null;
|
||||
const live = liveRun(runId);
|
||||
const existing = await db
|
||||
.selectFrom("approval")
|
||||
.selectAll()
|
||||
.where(sql<boolean>`scope->>'agentRunId' = ${runId} and scope->>'invocationId' = ${inv}`)
|
||||
.orderBy("created_at", "desc")
|
||||
.executeTakeFirst();
|
||||
|
||||
const resolved = async (
|
||||
status: "approved" | "rejected" | "expired" | "revoked" | "invalid",
|
||||
reason: string | null,
|
||||
) => {
|
||||
if (!existing || !live) return;
|
||||
if (await alreadyEmitted(db, runId, "approval_resolved", "approvalId", existing.id)) return;
|
||||
live.handle.appendEvent({
|
||||
type: "approval_resolved",
|
||||
approvalId: existing.id,
|
||||
status,
|
||||
reason: reason ? safe(reason, 500) : null,
|
||||
decidedBy: existing.decided_by,
|
||||
});
|
||||
};
|
||||
|
||||
if (existing) {
|
||||
const b = plan.binding;
|
||||
if (existing.status === "approved") {
|
||||
if (existing.plan_type !== plan.planType || existing.plan_id !== plan.planId) {
|
||||
await resolved("invalid", "审批绑定的计划与本次调用不一致");
|
||||
return { allowed: false, reason: "审批绑定的计划与本次调用不一致" };
|
||||
}
|
||||
if (existing.payload_hash !== b.payloadHash || existing.expected_revision !== b.revision) {
|
||||
await resolved("invalid", "计划在审批期间变化,旧审批失效");
|
||||
return { allowed: false, reason: "计划在审批期间变化,旧审批失效;重新生成计划后再提交" };
|
||||
}
|
||||
if (existing.expires_at && existing.expires_at.getTime() <= now.getTime()) {
|
||||
await resolved("expired", null);
|
||||
return { allowed: false, reason: "审批已过期" };
|
||||
}
|
||||
await resolved("approved", existing.reason);
|
||||
return { allowed: true, approvalId: existing.id };
|
||||
}
|
||||
if (existing.status === "consumed") {
|
||||
// 已用这份审批执行成功过(重放):交给操作账本返回记录的结果
|
||||
const ok = await db
|
||||
.selectFrom("capability_invocation")
|
||||
.select("id")
|
||||
.where("approval_id", "=", existing.id)
|
||||
.where("status", "=", "succeeded")
|
||||
.executeTakeFirst();
|
||||
return ok ? { allowed: true, approvalId: existing.id } : { allowed: false, reason: "审批已使用" };
|
||||
}
|
||||
if (existing.status === "rejected" || existing.status === "revoked") {
|
||||
await resolved(existing.status, existing.reason);
|
||||
return {
|
||||
allowed: false,
|
||||
reason: `管理员${existing.status === "rejected" ? "拒绝" : "撤销"}了审批${existing.reason ? `:${safe(existing.reason, 500)}` : ""}`,
|
||||
};
|
||||
}
|
||||
if (existing.status === "expired" || (existing.expires_at && existing.expires_at.getTime() <= now.getTime())) {
|
||||
await expireApproval(sctx, existing.id);
|
||||
await resolved("expired", null);
|
||||
return { allowed: false, reason: "审批已过期,没有执行" };
|
||||
}
|
||||
// 仍在等待:重新挂起(崩溃重放 / 兜底轮询提前醒来)
|
||||
if (live?.suspend && !inv.startsWith("reconcile:"))
|
||||
await suspendRun(live, {
|
||||
kind: "approval",
|
||||
invocationId: inv,
|
||||
runIds: null,
|
||||
deadline: (existing.expires_at ?? new Date(now.getTime() + 86400_000)).toISOString(),
|
||||
approvalId: existing.id,
|
||||
});
|
||||
return { allowed: false, reason: "等待管理员审批" };
|
||||
}
|
||||
|
||||
// 新调用:只有运行中的会话能挂起等审批;崩溃对账不发起审批
|
||||
if (!live?.suspend || inv.startsWith("reconcile:")) return null;
|
||||
const s = await sctx.settings.get("agent");
|
||||
const expiresAt = new Date(now.getTime() + s.approvalTimeoutSec * 1000);
|
||||
const stepId = await currentStepId(sctx, live, runId);
|
||||
const summary = await planSummary(db, plan.planType, plan.planId);
|
||||
const decision = await insertDecision(sctx, {
|
||||
kind: "approval_required",
|
||||
title: `Agent 请求审批:${CAPABILITY_CATALOG[d.id as KnownCapabilityId]?.title ?? d.id}`,
|
||||
summary,
|
||||
runId,
|
||||
dedupeKey: `${APPROVAL_KEY}:${runId}:${inv}`,
|
||||
evidence: { source: "agent.approval", runId, invocationId: inv, capabilityId: d.id, ...plan, binding: undefined },
|
||||
options: [],
|
||||
});
|
||||
const scope: ApprovalScope = { agentRunId: runId, invocationId: inv, capabilityId: d.id, stepId, summary };
|
||||
const a = await db
|
||||
.insertInto("approval")
|
||||
.values({
|
||||
plan_type: plan.planType,
|
||||
plan_id: plan.planId,
|
||||
payload_hash: plan.binding.payloadHash,
|
||||
expected_revision: plan.binding.revision,
|
||||
scope: JSON.stringify(scope),
|
||||
decision_id: decision.id,
|
||||
requested_by: `agent:${runId}`,
|
||||
expires_at: expiresAt,
|
||||
reason: summary,
|
||||
})
|
||||
.returningAll()
|
||||
.executeTakeFirstOrThrow();
|
||||
live.handle.appendEvent({
|
||||
type: "approval_requested",
|
||||
approvalId: a.id,
|
||||
decisionId: decision.id,
|
||||
invocationId: inv,
|
||||
capabilityId: d.id,
|
||||
planType: plan.planType,
|
||||
planId: plan.planId,
|
||||
payloadHash: a.payload_hash,
|
||||
expectedRevision: a.expected_revision,
|
||||
riskLevel: d.riskLevel,
|
||||
summary,
|
||||
stepId,
|
||||
expiresAt: expiresAt.toISOString(),
|
||||
});
|
||||
await suspendRun(live, {
|
||||
kind: "approval",
|
||||
invocationId: inv,
|
||||
runIds: null,
|
||||
deadline: expiresAt.toISOString(),
|
||||
approvalId: a.id,
|
||||
});
|
||||
return { allowed: false, reason: "已在步骤处发起审批,等待管理员批准" };
|
||||
}
|
||||
|
||||
async function expireApproval(ctx: ServerContext, approvalId: string) {
|
||||
const a = await ctx.db
|
||||
.updateTable("approval")
|
||||
.set({ status: "expired" })
|
||||
.where("id", "=", approvalId)
|
||||
.where("status", "=", "pending")
|
||||
.returning("decision_id")
|
||||
.executeTakeFirst();
|
||||
if (a?.decision_id) await closeDecision(ctx, a.decision_id, "expired", { approval: "expired" }, "system:timeout");
|
||||
}
|
||||
|
||||
/**
|
||||
* 审批决定之后(PATCH /approvals/{id}):关联的待处理事项写结果(拒绝时带理由与建议),唤醒挂起的会话。
|
||||
*/
|
||||
export async function afterApprovalDecided(
|
||||
ctx: ServerContext,
|
||||
a: { id: string; status: string; scope: unknown; decision_id: string | null; reason: string | null },
|
||||
by: string,
|
||||
): Promise<void> {
|
||||
const scope = (a.scope ?? {}) as Partial<ApprovalScope>;
|
||||
if (!scope.agentRunId) return;
|
||||
if (a.decision_id) {
|
||||
const status = a.status === "approved" ? "resolved" : "dismissed";
|
||||
await closeDecision(
|
||||
ctx,
|
||||
a.decision_id,
|
||||
status,
|
||||
{
|
||||
approval: a.status,
|
||||
reason: a.reason,
|
||||
...(a.status === "rejected" ? { suggestion: a.reason ?? "按拒绝理由调整方案" } : {}),
|
||||
by,
|
||||
},
|
||||
by,
|
||||
);
|
||||
}
|
||||
await wakeRun(ctx, scope.agentRunId, { approvalId: a.id });
|
||||
}
|
||||
|
||||
/** 停止会话时:收起它还在等用户处理的方案选择与审批 */
|
||||
export async function dismissPendingInteractions(ctx: ServerContext, runId: string, by: string) {
|
||||
const open = await ctx.db
|
||||
.selectFrom("decision")
|
||||
.select("id")
|
||||
.where("agent_run_id", "=", runId)
|
||||
.where("status", "in", ["open", "awaiting_user"])
|
||||
.where((eb) => eb.or([eb("dedupe_key", "like", `${OPTIONS_KEY}:%`), eb("dedupe_key", "like", `${APPROVAL_KEY}:%`)]))
|
||||
.execute();
|
||||
for (const d of open) await closeDecision(ctx, d.id, "dismissed", { reason: "会话已停止" }, by);
|
||||
await ctx.db
|
||||
.updateTable("approval")
|
||||
.set({ status: "revoked", decided_by: by, decided_at: new Date(), reason: "会话已停止" })
|
||||
.where(sql<boolean>`scope->>'agentRunId' = ${runId}`)
|
||||
.where("status", "=", "pending")
|
||||
.execute();
|
||||
}
|
||||
|
||||
/** 先看方案再执行:用户选定方案前拒绝执行类能力。返回拒绝原因或 null */
|
||||
export async function proposeFirstBlock(db: Kysely<DB>, runId: string): Promise<string | null> {
|
||||
const self = await db.selectFrom("agent_run").selectAll().where("id", "=", runId).executeTakeFirst();
|
||||
if (!self) return null;
|
||||
// 子会话随父会话:顶层会话要求先看方案时,选定前子会话也不能执行
|
||||
const root = [self, ...(await ancestors(db, runId))].at(-1)!;
|
||||
if (!sessionOptions(root).proposeFirst) return null;
|
||||
const chosen = await db
|
||||
.selectFrom("decision")
|
||||
.select("id")
|
||||
.where("agent_run_id", "=", root.id)
|
||||
.where("dedupe_key", "like", `${OPTIONS_KEY}:%`)
|
||||
.where("status", "=", "resolved")
|
||||
.executeTakeFirst();
|
||||
return chosen ? null : "本会话要求先看方案再执行:先用 agent.propose_options 给出方案,用户选定后才能执行";
|
||||
}
|
||||
|
||||
// ── 挂起唤醒条件(runAgent 的 waitForFact) ──
|
||||
|
||||
export async function interactionReady(ctx: ServerContext, runId: string, sus: AwaitSuspension): Promise<true | null> {
|
||||
if (Date.now() >= Date.parse(sus.deadline)) return true;
|
||||
const run = await ctx.db
|
||||
.selectFrom("agent_run")
|
||||
.select("cancel_requested_at")
|
||||
.where("id", "=", runId)
|
||||
.executeTakeFirst();
|
||||
if (!run || run.cancel_requested_at) return true;
|
||||
if (sus.kind === "options" && sus.decisionId) {
|
||||
const d = await ctx.db.selectFrom("decision").select("status").where("id", "=", sus.decisionId).executeTakeFirst();
|
||||
return !d || !["open", "investigating", "awaiting_user"].includes(d.status) ? true : null;
|
||||
}
|
||||
if (sus.kind === "approval" && sus.approvalId) {
|
||||
const a = await ctx.db.selectFrom("approval").select("status").where("id", "=", sus.approvalId).executeTakeFirst();
|
||||
return a?.status !== "pending" ? true : null;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/** 挂起时会话显示"等待中"(等用户) */
|
||||
export async function markWaitingForUser(ctx: ServerContext, runId: string, sus: AwaitSuspension): Promise<null> {
|
||||
const summary = sus.kind === "options" ? "等待选择方案" : "等待审批";
|
||||
await ctx.db
|
||||
.updateTable("agent_run")
|
||||
.set({ status: "waiting", summary })
|
||||
.where("id", "=", runId)
|
||||
.where("status", "in", ["running", "waiting"])
|
||||
.execute();
|
||||
void ctx.events.publish("agent.run_status", { runId, status: "waiting", summary });
|
||||
return null;
|
||||
}
|
||||
|
||||
// ── 能力处理器 ──
|
||||
|
||||
export function registerInteractionCapabilities(registry: CapabilityRegistry, ctx: ServerContext) {
|
||||
bindInteraction(ctx);
|
||||
|
||||
registry.register(capabilitySpec("agent.plan"), async (input, c) => {
|
||||
const run = await runOf(ctx, c);
|
||||
const live = liveRun(run.id);
|
||||
if (!live) throw new CapabilityError("precondition", "会话不在运行中,不能更新步骤清单");
|
||||
const cur = await currentPlan(ctx, live, run.id);
|
||||
const byId = new Map(input.replace ? [] : cur.steps.map((s) => [s.id, s] as const));
|
||||
const order = input.replace ? [] : cur.steps.map((s) => s.id);
|
||||
for (const s of input.steps) {
|
||||
const old = byId.get(s.id);
|
||||
byId.set(s.id, {
|
||||
id: s.id,
|
||||
title: s.title ?? old?.title ?? s.id,
|
||||
status: s.status ?? old?.status ?? "pending",
|
||||
note: s.note === undefined ? (old?.note ?? null) : s.note,
|
||||
});
|
||||
if (!order.includes(s.id)) order.push(s.id);
|
||||
}
|
||||
if (order.length > 30) throw new CapabilityError("invalid_input", "步骤最多 30 个;合并细碎步骤后再更新");
|
||||
const next = { steps: order.map((id) => byId.get(id)!), revision: cur.revision + 1 };
|
||||
live.plan = next;
|
||||
live.handle.appendEvent({ type: "plan_updated", steps: next.steps, revision: next.revision });
|
||||
return next;
|
||||
});
|
||||
|
||||
// replay=safe:挂起后恢复时重放本调用取用户的选择;同一 invocationId → 同一个待处理事项
|
||||
registry.register({ ...capabilitySpec("agent.propose_options"), replay: "safe" }, async (input, c) => {
|
||||
const run = await runOf(ctx, c);
|
||||
if (!canAskUser(run))
|
||||
throw new CapabilityError("precondition", "子会话和后台会话不能让用户选方案:自己按证据决定,在汇报里说明取舍");
|
||||
const live = liveRun(run.id);
|
||||
const inv = c.invocationId;
|
||||
if (!live?.suspend || !inv) throw new CapabilityError("precondition", "会话不在运行中");
|
||||
const ids = input.options.map((o) => o.id);
|
||||
if (new Set(ids).size !== ids.length) throw new CapabilityError("invalid_input", "方案 id 不能重复");
|
||||
const recommendedId = input.recommendedId && ids.includes(input.recommendedId) ? input.recommendedId : null;
|
||||
const s = await ctx.settings.get("agent");
|
||||
const now = Date.now();
|
||||
const evidence: OptionsEvidence = {
|
||||
source: "agent.propose_options",
|
||||
runId: run.id,
|
||||
invocationId: inv,
|
||||
question: input.question,
|
||||
options: input.options,
|
||||
recommendedId,
|
||||
expiresAt: new Date(now + s.optionsTimeoutSec * 1000).toISOString(),
|
||||
timeoutAction: s.optionsTimeoutAction,
|
||||
};
|
||||
const dedupeKey = `${OPTIONS_KEY}:${run.id}:${inv}`;
|
||||
let d = await ctx.db
|
||||
.selectFrom("decision")
|
||||
.select(["id", "status", "evidence", "resolution"])
|
||||
.where("dedupe_key", "=", dedupeKey)
|
||||
.orderBy("created_at", "desc")
|
||||
.executeTakeFirst();
|
||||
if (!d) {
|
||||
if (inv.startsWith("reconcile:")) throw new AwaitSuspendedError();
|
||||
const stepId = await currentStepId(ctx, live, run.id);
|
||||
d = await insertDecision(ctx, {
|
||||
kind: "other",
|
||||
title: `Agent 等你选方案:${clip(input.question, 80)}`,
|
||||
summary: input.options.map((o) => `${o.id}. ${o.title}`).join(";"),
|
||||
runId: run.id,
|
||||
dedupeKey,
|
||||
evidence: { ...evidence, stepId },
|
||||
options: input.options.map((o) => ({ key: o.id, label: o.title, description: o.summary })),
|
||||
});
|
||||
live.handle.appendEvent({
|
||||
type: "options_proposed",
|
||||
invocationId: inv,
|
||||
decisionId: d.id,
|
||||
question: safe(input.question, 500),
|
||||
options: input.options.map((o) => ({
|
||||
...o,
|
||||
title: safe(o.title, 200),
|
||||
summary: safe(o.summary, 1000),
|
||||
risks: o.risks.map((r) => safe(r, 300)),
|
||||
estimatedCost: o.estimatedCost ? safe(o.estimatedCost, 200) : null,
|
||||
})),
|
||||
recommendedId,
|
||||
stepId,
|
||||
expiresAt: evidence.expiresAt,
|
||||
timeoutAction: evidence.timeoutAction,
|
||||
});
|
||||
}
|
||||
const ev = d.evidence as unknown as OptionsEvidence;
|
||||
let result: AgentProposeOptionsOutput | null = null;
|
||||
if (d.status === "resolved" || d.status === "dismissed" || d.status === "expired") {
|
||||
const res = (d.resolution ?? {}) as OptionsResolution;
|
||||
result = optionsResult(ev, d.status, res, res.timeout ? "timeout" : "user");
|
||||
} else if (Date.now() >= Date.parse(ev.expiresAt)) {
|
||||
// 超时:按设置自动选推荐项(没有推荐项选第一个)或停止
|
||||
const pick = ev.recommendedId ?? ev.options[0]?.id ?? null;
|
||||
const res: OptionsResolution =
|
||||
ev.timeoutAction === "recommended" && pick
|
||||
? { optionKey: pick, outcome: "chosen", reason: "超时未选,按设置自动选推荐项", timeout: true }
|
||||
: { optionKey: null, outcome: "stop", reason: "超时未选,按设置停止", timeout: true };
|
||||
await closeDecision(ctx, d.id, res.outcome === "chosen" ? "resolved" : "expired", { ...res }, "system:timeout");
|
||||
result = optionsResult(ev, res.outcome === "chosen" ? "resolved" : "expired", res, "timeout");
|
||||
}
|
||||
if (!result) {
|
||||
// 挂起:等用户作答 / 超时(崩溃对账不挂起)
|
||||
if (!inv.startsWith("reconcile:"))
|
||||
await suspendRun(live, {
|
||||
kind: "options",
|
||||
invocationId: inv,
|
||||
runIds: null,
|
||||
deadline: ev.expiresAt,
|
||||
decisionId: d.id,
|
||||
});
|
||||
throw new AwaitSuspendedError();
|
||||
}
|
||||
if (!(await alreadyEmitted(ctx.db, run.id, "options_resolved", "decisionId", d.id)))
|
||||
live.handle.appendEvent({
|
||||
type: "options_resolved",
|
||||
invocationId: inv,
|
||||
decisionId: d.id,
|
||||
outcome: result.outcome,
|
||||
choice: result.choice,
|
||||
reason: result.reason,
|
||||
by: result.by,
|
||||
});
|
||||
return result;
|
||||
});
|
||||
}
|
||||
@@ -14,6 +14,7 @@ import type { ModuleApp, ServerContext } from "../types.ts";
|
||||
import { planBinding } from "./authorize.ts";
|
||||
import { cancelTask, retryTask, taskToApi } from "./capabilities.ts";
|
||||
import { route } from "./http.ts";
|
||||
import { afterApprovalDecided, answerOptions } from "./interaction.ts";
|
||||
import { type InvokeLimiter, sharedInvokeLimiter, UpstreamBusyError, UpstreamTimeoutError } from "./invoke-limits.ts";
|
||||
import { AgentRunError, abortRun, appendMessage, startRun } from "./service.ts";
|
||||
import { descendants, descendantUsage } from "./subagents.ts";
|
||||
@@ -56,7 +57,6 @@ export function runToApi(r: AgentRunRow) {
|
||||
startedAt: toIso(r.started_at),
|
||||
finishedAt: toIso(r.finished_at),
|
||||
createdBy: r.created_by,
|
||||
mode: r.mode,
|
||||
dedupeKey: r.dedupe_key,
|
||||
depth: r.depth,
|
||||
spawnReason: r.spawn_reason,
|
||||
@@ -73,7 +73,6 @@ export function runToNode(r: AgentRunRow) {
|
||||
parentRunId: r.parent_run_id,
|
||||
goal: r.goal,
|
||||
status: r.status,
|
||||
mode: r.mode,
|
||||
depth: r.depth,
|
||||
spawnReason: r.spawn_reason,
|
||||
toolCallCount: r.tool_call_count,
|
||||
@@ -196,13 +195,21 @@ export function registerAgentRoutes(
|
||||
route(
|
||||
app,
|
||||
Api.agentRuns.create,
|
||||
async ({ body, actorId }) => {
|
||||
async ({ body, actorId, c }) => {
|
||||
// 运行模式已取消:用户会话一律按授权执行;旧客户端传 mode=plan_only 按"先看方案再执行"处理
|
||||
const proposeFirst = body.proposeFirst || body.mode === "plan_only";
|
||||
if (body.mode) c.header("Deprecation", "true");
|
||||
const r = await startRun(ctx, {
|
||||
goal: body.goal,
|
||||
mode: body.mode,
|
||||
mode: "execute",
|
||||
capabilityAllowlist: body.capabilityAllowlist,
|
||||
trigger: "user",
|
||||
createdBy: actorId,
|
||||
session: {
|
||||
// 控制台管理员会话发起:需审批的动作在步骤处出审批卡(只有管理员会话能批准)
|
||||
interactive: c.get("principal")?.kind === "admin_session",
|
||||
proposeFirst,
|
||||
},
|
||||
}).catch(mapErr);
|
||||
return runToApi(r.run);
|
||||
},
|
||||
@@ -213,6 +220,19 @@ export function registerAgentRoutes(
|
||||
runToApi(await appendMessage(ctx, params.id, body.text, actorId).catch(mapErr)),
|
||||
);
|
||||
|
||||
route(app, Api.agentRuns.choose, async ({ params, body, actorId }) => {
|
||||
const run = await db().selectFrom("agent_run").selectAll().where("id", "=", params.id).executeTakeFirst();
|
||||
if (!run) throw new ApiException("not_found", "Agent 运行不存在");
|
||||
if (body.optionId === null && !body.reason?.trim())
|
||||
throw new ApiException("validation_failed", "都不要时写一句原因,Agent 按它调整方案");
|
||||
const r = await answerOptions(ctx, run.id, body, actorId);
|
||||
if (r === "not_found") throw new ApiException("not_found", "这个会话没有这项方案选择");
|
||||
if (r === "bad_option") throw new ApiException("validation_failed", "没有这个方案");
|
||||
if (r === "closed") throw new ApiException("conflict", "这项方案选择已处理");
|
||||
const fresh = await db().selectFrom("agent_run").selectAll().where("id", "=", run.id).executeTakeFirstOrThrow();
|
||||
return runToApi(fresh);
|
||||
});
|
||||
|
||||
route(app, Api.agentRuns.abort, async ({ params, body, actorId }) =>
|
||||
runToApi(await abortRun(ctx, params.id, body.reason, actorId).catch(mapErr)),
|
||||
);
|
||||
@@ -353,6 +373,8 @@ export function registerAgentRoutes(
|
||||
.executeTakeFirstOrThrow();
|
||||
});
|
||||
await audit(ctx, actorId, `approval.${body.status}`, r.id, { planType: r.plan_type, planId: r.plan_id });
|
||||
// 会话内审批:写关联待处理事项的结果(拒绝带理由),唤醒挂起在这份审批上的会话
|
||||
await afterApprovalDecided(ctx, r, actorId);
|
||||
return camel(r);
|
||||
});
|
||||
|
||||
|
||||
@@ -26,6 +26,7 @@ import { type AgentRunRow, type DownloaderInstanceRow, sql } from "@umt/db";
|
||||
import { type RunUpstream, resolveRunUpstream } from "../pi/index.ts";
|
||||
import type { ServerContext } from "../types.ts";
|
||||
import { DEMO_SCRIPTS, parseFakeScripts } from "./fake-scripts.ts";
|
||||
import { canAskUser, canRequestApproval, sessionOptions } from "./interaction.ts";
|
||||
import { buildGoalReport } from "./report.ts";
|
||||
import { type AwaitSuspension, attachLiveRun, type LiveRun, subagentSettings, wakeParent } from "./subagents.ts";
|
||||
|
||||
@@ -72,9 +73,12 @@ export interface RunScope {
|
||||
capabilities: string[];
|
||||
allowedKinds: CapabilityKind[] | null;
|
||||
decisionId?: string | null;
|
||||
/** 会话选项(见 interaction.ts RunSessionOptions) */
|
||||
interactive?: boolean;
|
||||
proposeFirst?: boolean;
|
||||
}
|
||||
|
||||
/** 按模式确定能力白名单与类别上限;plan_only 下给了 execute 类能力直接报错 */
|
||||
/** 按内部能力上限确定白名单与类别上限;plan_only(后台只读调查)下给了 execute 类能力直接报错 */
|
||||
export function scopeFor(mode: "execute" | "plan_only", allowlist?: readonly string[]): RunScope {
|
||||
const all = Object.values(CAPABILITY_CATALOG);
|
||||
if (mode === "plan_only") {
|
||||
@@ -251,7 +255,7 @@ function systemPromptFor(
|
||||
`可用能力:${scope.capabilities.join(", ")}`,
|
||||
"有副作用的能力(提交下载、执行整理、发布策略、重试/取消任务)需要服务端授权;被拒时说明缺少哪项授权。",
|
||||
];
|
||||
if (run.mode === "plan_only") constraints.push("本次只给方案:只调查与规划,不提交下载、不改文件、不发布策略。");
|
||||
if (run.mode === "plan_only") constraints.push("本次只读:只调查与规划,不提交下载、不改文件、不发布策略。");
|
||||
if (sub.canSpawn)
|
||||
constraints.push(
|
||||
`可以用 agent.spawn 把彼此独立的有界子目标交给子会话(同时最多 ${sub.maxConcurrent} 个,派生层数上限 ${sub.maxDepth});` +
|
||||
@@ -265,10 +269,22 @@ function systemPromptFor(
|
||||
);
|
||||
else if (run.trigger === "event")
|
||||
constraints.push("本任务由后台事件触发,只做只读调查:收集证据,给出结论与建议,不要反复重试同一操作。");
|
||||
const opts = sessionOptions(run);
|
||||
return buildSystemPrompt({
|
||||
role: run.trigger === "event" ? "媒体库异常调查员" : "媒体库自动化 Agent",
|
||||
goal: run.goal,
|
||||
constraints,
|
||||
// 自动化优先:用户会话默认执行,先出步骤清单;有取舍时给方案;"先看方案再执行"是会话选项
|
||||
...(canAskUser(run)
|
||||
? {
|
||||
interaction: {
|
||||
plan: scope.capabilities.includes("agent.plan"),
|
||||
options: scope.capabilities.includes("agent.propose_options"),
|
||||
proposeFirst: opts.proposeFirst === true,
|
||||
approvals: canRequestApproval(run),
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
});
|
||||
}
|
||||
|
||||
@@ -389,9 +405,20 @@ export async function drive(
|
||||
.select("id")
|
||||
.where("parent_run_id", "=", runId)
|
||||
.executeTakeFirst());
|
||||
// 方案选择只在用户发起的顶层会话里提供(子会话 / 后台会话没有人可问)
|
||||
const askUser = canAskUser(run);
|
||||
const capabilities = (scope.capabilities?.length ? scope.capabilities : Object.keys(CAPABILITY_CATALOG)).filter(
|
||||
(id) => (id === "agent.spawn" ? canSpawn : id === "agent.await" ? canSpawn || hasChildren : true),
|
||||
(id) =>
|
||||
id === "agent.spawn"
|
||||
? canSpawn
|
||||
: id === "agent.await"
|
||||
? canSpawn || hasChildren
|
||||
: id === "agent.propose_options"
|
||||
? askUser
|
||||
: true,
|
||||
);
|
||||
// 会话内审批:挂起的审批调用恢复后要重放(只放行本次恢复针对的那个调用)
|
||||
const resumed = new Set(opts.resumedAwaits ?? []);
|
||||
let external: LiveRun | null = null;
|
||||
let handle: AgentRunHandle;
|
||||
try {
|
||||
@@ -405,6 +432,7 @@ export async function drive(
|
||||
actor,
|
||||
firstSeq: maxSeq?.m === null || maxSeq?.m === undefined ? 0 : Number(maxSeq.m) + 1,
|
||||
externalUsage: () => external?.externalUsage() ?? { toolCalls: 0, totalTokens: 0, costUsd: 0 },
|
||||
...(canRequestApproval(run) ? { replayGate: (_cap: string, inv: string) => resumed.has(inv) } : {}),
|
||||
});
|
||||
} catch (err) {
|
||||
return { status: "failed", reason: "error", budget: null, message: (err as Error).message, finalText: "" };
|
||||
|
||||
@@ -10,9 +10,17 @@ import type {
|
||||
} from "@umt/contracts";
|
||||
import { type AgentRunRow, type DB, type Kysely, sql } from "@umt/db";
|
||||
import type { ServerContext } from "../types.ts";
|
||||
import { dismissPendingInteractions } from "./interaction.ts";
|
||||
import { buildGoalReport } from "./report.ts";
|
||||
import { type AgentEnv, type DriveResult, INVESTIGATE_CAPABILITIES, publishStatus, scopeFor } from "./runner.ts";
|
||||
import { cascadeCancel, descendants, descendantUsage, recordChildCompletion, wakeParent } from "./subagents.ts";
|
||||
import {
|
||||
cascadeCancel,
|
||||
descendants,
|
||||
descendantUsage,
|
||||
recordChildCompletion,
|
||||
wakeParent,
|
||||
wakeRun,
|
||||
} from "./subagents.ts";
|
||||
|
||||
// agent_run 生命周期:建行 + 业务任务行 + 入队 runAgent;追加消息 / 中止;终态汇报(业务表核查)与后台调查去重。
|
||||
|
||||
@@ -37,6 +45,8 @@ export interface StartRunInput {
|
||||
dedupeKey?: string | null;
|
||||
budget?: AgentBudget;
|
||||
allowedKinds?: ("read" | "preview" | "propose")[];
|
||||
/** 会话选项:控制台管理员会话(可内联审批)、先看方案再执行 */
|
||||
session?: { interactive?: boolean; proposeFirst?: boolean };
|
||||
/** 子代理(agent.spawn):父会话、深度、理由、上游与已算好的范围(白名单 ⊆ 父、类别取交集) */
|
||||
subagent?: {
|
||||
parentRunId: string;
|
||||
@@ -61,6 +71,8 @@ export async function startRun(ctx: ServerContext, p: StartRunInput): Promise<{
|
||||
}
|
||||
}
|
||||
if (p.allowedKinds) scope = { ...scope, allowedKinds: p.allowedKinds };
|
||||
if (p.session?.interactive) scope = { ...scope, interactive: true };
|
||||
if (p.session?.proposeFirst) scope = { ...scope, proposeFirst: true };
|
||||
const sub = p.subagent;
|
||||
const ins = await ctx.db.transaction().execute(async (trx) => {
|
||||
const row = await sql<AgentRunRow>`
|
||||
@@ -181,6 +193,9 @@ export async function abortRun(ctx: ServerContext, runId: string, reason: string
|
||||
.execute();
|
||||
// 子代理共享取消状态:递归传播到全部未结束的子孙会话
|
||||
await cascadeCancel(ctx, runId, reason ?? "用户中止");
|
||||
// 还在等用户的方案选择 / 审批一并收起(挂起中的会话会被唤醒并按取消收尾)
|
||||
await dismissPendingInteractions(ctx, runId, by);
|
||||
await wakeRun(ctx, runId, { cancelled: true });
|
||||
return row;
|
||||
}
|
||||
|
||||
|
||||
@@ -9,6 +9,7 @@ import {
|
||||
} from "@umt/agent";
|
||||
import {
|
||||
type AgentBudget,
|
||||
type AgentPlanStep,
|
||||
type AgentRunEvent,
|
||||
type AgentRunMode,
|
||||
type AgentRunStatus,
|
||||
@@ -84,14 +85,22 @@ export interface AwaitWake {
|
||||
fact: string;
|
||||
}
|
||||
|
||||
/** agent.await / agent.spawn{await:true} 挂起请求:runAgent 据此 waitForFact,落定后以 resumedAwaits 重开会话 */
|
||||
/**
|
||||
* 会话挂起请求:runAgent 据此 waitForFact,条件满足后以 resumedAwaits 重开会话并重放挂起的调用。
|
||||
* - subagents(缺省):agent.await / agent.spawn{await:true} 等子会话落定;
|
||||
* - options:agent.propose_options 等用户选方案(decisionId);
|
||||
* - approval:需审批的能力等管理员批准 / 拒绝(approvalId)。
|
||||
*/
|
||||
export interface AwaitSuspension {
|
||||
kind?: "subagents" | "options" | "approval";
|
||||
/** 挂起的工具调用(恢复时重放,同一 invocationId 不再挂起) */
|
||||
invocationId: string;
|
||||
/** 等待的子会话;null = 本会话全部子会话 */
|
||||
runIds: string[] | null;
|
||||
/** 超过这个时间(ISO)不再等,恢复后返回当前状态(未结束的标 pending) */
|
||||
/** 超过这个时间(ISO)不再等,恢复后返回当前状态(未结束的标 pending / 按超时处理) */
|
||||
deadline: string;
|
||||
decisionId?: string;
|
||||
approvalId?: string;
|
||||
}
|
||||
|
||||
export interface LiveRun {
|
||||
@@ -102,6 +111,8 @@ export interface LiveRun {
|
||||
pump(): Promise<void>;
|
||||
/** 子孙会话已用量(同步读缓存,pump 刷新) */
|
||||
externalUsage(): ExternalUsage;
|
||||
/** agent.plan 的当前步骤清单(首次用到时从事件流载入) */
|
||||
plan?: { steps: AgentPlanStep[]; revision: number } | null;
|
||||
}
|
||||
|
||||
const live = new Map<string, LiveRun>();
|
||||
@@ -638,17 +649,41 @@ export async function wakeParent(ctx: ServerContext, childRunId: string): Promis
|
||||
}
|
||||
}
|
||||
|
||||
/** 唤醒挂起中的会话自身(用户选了方案 / 审批已决定):按 await_wake 发引擎信号;没挂起时什么也不做 */
|
||||
export async function wakeRun(
|
||||
ctx: ServerContext,
|
||||
runId: string,
|
||||
payload: Record<string, string | boolean>,
|
||||
): Promise<boolean> {
|
||||
try {
|
||||
const r = await ctx.db.selectFrom("agent_run").select("await_wake").where("id", "=", runId).executeTakeFirst();
|
||||
const w = r?.await_wake as AwaitWake | null | undefined;
|
||||
if (!w?.engineRunId || !w.fact) return false;
|
||||
await ctx.engine.signal(w.engineRunId, w.fact, payload);
|
||||
return true;
|
||||
} catch (err) {
|
||||
ctx.logger.warn({ runId, err: { message: (err as Error).message } }, "唤醒会话失败(会话会兜底轮询)");
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 子会话还没落定时请求挂起:记下等待条件并关闭会话(本调用在 JSONL 里保持未完成),drive 看到 requested 后返回
|
||||
* suspended。同一批里已有别的调用在挂起时不再重复登记。
|
||||
*/
|
||||
async function requestSuspension(live: LiveRun, invocationId: string, runIds: string[] | null, timeoutSec: number) {
|
||||
if (!live.suspend || live.suspend.requested) return;
|
||||
live.suspend.requested = {
|
||||
await suspendRun(live, {
|
||||
kind: "subagents",
|
||||
invocationId,
|
||||
runIds,
|
||||
deadline: new Date(Date.now() + timeoutSec * 1000).toISOString(),
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
/** 登记挂起条件并关闭会话(挂起的调用在 JSONL 里保持未完成);同一批里已有挂起时不再重复登记 */
|
||||
export async function suspendRun(live: LiveRun, sus: AwaitSuspension): Promise<void> {
|
||||
if (!live.suspend || live.suspend.requested) return;
|
||||
live.suspend.requested = sus;
|
||||
await new Promise((r) => setTimeout(r, 0));
|
||||
await live.handle.close();
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ import { defineTask, type OutboxRouteFn, type TaskRegistry } from "@umt/tasks";
|
||||
import { z } from "zod";
|
||||
import type { ServerContext } from "../types.ts";
|
||||
import { retryTask } from "./capabilities.ts";
|
||||
import { interactionReady, markWaitingForUser } from "./interaction.ts";
|
||||
import {
|
||||
annotateRunSummary,
|
||||
applyRecoveryOutcome,
|
||||
@@ -90,11 +91,17 @@ export function registerAgentTasks(registry: TaskRegistry, env: () => AgentEnv)
|
||||
break;
|
||||
}
|
||||
const waitMs = Math.max(0, Date.parse(sus.deadline) - Date.now());
|
||||
// 等用户(方案选择 / 审批)与等子会话共用挂起机制;步骤名保持 subagent-await-N 以兼容已有运行的步骤历史
|
||||
const user = sus.kind === "options" || sus.kind === "approval";
|
||||
const fact = `subagent-await-${n}`;
|
||||
const engineRunId = t.runId;
|
||||
await t.step.run(`${fact}-arm`, () => setAwaitWake(e.ctx, runId, { engineRunId, fact }));
|
||||
await t.step.run(`${fact}-arm`, async () => {
|
||||
await setAwaitWake(e.ctx, runId, { engineRunId, fact });
|
||||
if (user) await markWaitingForUser(e.ctx, runId, sus);
|
||||
return null;
|
||||
});
|
||||
await t.step.waitForFact<true>(fact, {
|
||||
check: () => awaitReady(e.ctx, runId, sus),
|
||||
check: () => (user ? interactionReady(e.ctx, runId, sus) : awaitReady(e.ctx, runId, sus)),
|
||||
timeout: `${AWAIT_POLL_SEC}s`,
|
||||
maxRounds: Math.max(1, Math.ceil(waitMs / 1000 / AWAIT_POLL_SEC) + 1),
|
||||
});
|
||||
|
||||
@@ -0,0 +1,429 @@
|
||||
import type { FakeScript, FakeToolResult } from "@umt/agent";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { ensureSetupToken } from "../../src/modules/auth-settings/auth-service.ts";
|
||||
import { Browser } from "../auth-settings/harness.ts";
|
||||
import { createE2E, type E2E, pgAvailable, runEvents, runRow, seedBase, seedMovieTarget, waitFor } from "./harness.ts";
|
||||
|
||||
// E12 Agent 交互自动化优先(AG-2):
|
||||
// - 计划事件顺序:agent.plan 先出步骤清单(plan_updated rev1)再执行,执行中更新,revision 递增;
|
||||
// - 方案选择:agent.propose_options 登记待处理事项并挂起会话(status=waiting、await_wake 已登记,不占 worker);
|
||||
// 用户选定后作为工具结果续跑;都不要 → rejected + 原因;超时按设置自动选推荐项;
|
||||
// - 先看方案再执行:选定方案前执行类能力被服务端拒绝;旧客户端 mode=plan_only 按它处理;
|
||||
// - 内联审批(控制台管理员会话):download.submit 没有持续授权 → 步骤处出审批卡并挂起;批准后重放执行、审批 consumed;
|
||||
// 拒绝 → 调用以拒绝理由失败,关联待处理事项写结果与建议。
|
||||
|
||||
const hasPg = await pgAvailable();
|
||||
// biome-ignore lint/suspicious/noExplicitAny: 测试断言直接读 JSON
|
||||
type Any = any;
|
||||
const ZERO = "00000000-0000-4000-8000-000000000000";
|
||||
const PASSWORD = "correct horse battery";
|
||||
|
||||
const lastOk = (results: FakeToolResult[], cap: string): Any =>
|
||||
[...results].reverse().find((r) => r.capabilityId === cap && !r.isError)?.json;
|
||||
const last = (results: FakeToolResult[], cap: string) => [...results].reverse().find((r) => r.capabilityId === cap);
|
||||
|
||||
const OPTIONS_GOAL = "E12 先看方案:沙丘2 收哪个版本";
|
||||
const NONE_GOAL = "E12 都不要:沙丘2 收哪个版本";
|
||||
const TIMEOUT_GOAL = "E12 超时:沙丘2 收哪个版本";
|
||||
|
||||
const OPTIONS = [
|
||||
{ id: "a", title: "REMUX", summary: "画质最好", risks: ["体积大"], estimatedCost: "78 GB" },
|
||||
{ id: "b", title: "WEB-DL", summary: "体积小", risks: [], estimatedCost: "24 GB" },
|
||||
];
|
||||
|
||||
const optionsScript = (goal: string): FakeScript => ({
|
||||
name: `e12-options-${goal}`,
|
||||
match: goal,
|
||||
steps: [
|
||||
{
|
||||
text: "先列步骤。",
|
||||
toolCalls: [
|
||||
{
|
||||
capability: "agent.plan",
|
||||
input: {
|
||||
steps: [
|
||||
{ id: "s1", title: "调查候选", status: "running" },
|
||||
{ id: "s2", title: "请你选方案" },
|
||||
{ id: "s3", title: "按方案执行" },
|
||||
],
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
// 先看方案:选定前试图直接提交 → 服务端拒绝
|
||||
{ toolCalls: [{ capability: "download.submit", input: { planId: ZERO } }] },
|
||||
{
|
||||
toolCalls: [
|
||||
{
|
||||
capability: "agent.plan",
|
||||
input: {
|
||||
steps: [
|
||||
{ id: "s1", status: "done" },
|
||||
{ id: "s2", status: "running" },
|
||||
],
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
toolCalls: [
|
||||
{
|
||||
capability: "agent.propose_options",
|
||||
input: { question: "收哪个版本?", options: OPTIONS, recommendedId: "a" },
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
toolCalls: [
|
||||
{
|
||||
capability: "agent.plan",
|
||||
input: {
|
||||
steps: [
|
||||
{ id: "s2", status: "done" },
|
||||
{ id: "s3", status: "done" },
|
||||
],
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
finalText: (r) => {
|
||||
const o = lastOk(r, "agent.propose_options");
|
||||
return `方案结果:${o?.outcome}/${o?.choice ?? "none"}/${o?.by}/${o?.reason ?? ""}`;
|
||||
},
|
||||
});
|
||||
|
||||
// 审批:两个目标各用一个版本目标(拒绝与批准互不影响);ID 在 beforeAll 里填
|
||||
const T: Record<"approve" | "reject", { slotId: string; releaseId: string }> = {
|
||||
approve: { slotId: ZERO, releaseId: ZERO },
|
||||
reject: { slotId: ZERO, releaseId: ZERO },
|
||||
};
|
||||
const APPROVE_GOAL = "E12 审批通过:提交沙丘2 1080p";
|
||||
const REJECT_GOAL = "E12 审批拒绝:提交沙丘2 1080p";
|
||||
const approvalScript = (goal: string, key: "approve" | "reject"): FakeScript => ({
|
||||
name: `e12-approval-${key}`,
|
||||
match: goal,
|
||||
steps: [
|
||||
{
|
||||
toolCalls: [
|
||||
{
|
||||
capability: "agent.plan",
|
||||
input: {
|
||||
steps: [
|
||||
{ id: "s1", title: "生成下载计划", status: "running" },
|
||||
{ id: "s2", title: "提交下载" },
|
||||
],
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
toolCalls: [
|
||||
{ capability: "download.plan", input: () => ({ slotId: T[key].slotId, sourceReleaseId: T[key].releaseId }) },
|
||||
],
|
||||
},
|
||||
{
|
||||
toolCalls: [
|
||||
{
|
||||
capability: "agent.plan",
|
||||
input: {
|
||||
steps: [
|
||||
{ id: "s1", status: "done" },
|
||||
{ id: "s2", status: "running" },
|
||||
],
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
toolCalls: [
|
||||
{
|
||||
capability: "download.submit",
|
||||
input: (r: FakeToolResult[]) => ({ planId: lastOk(r, "download.plan")?.plan?.id ?? ZERO }),
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
finalText: (r) => {
|
||||
const s = last(r, "download.submit");
|
||||
return s && !s.isError ? `已提交:${(s.json as Any)?.taskId}` : `未提交:${s?.text ?? "无结果"}`;
|
||||
},
|
||||
});
|
||||
|
||||
describe.skipIf(!hasPg)("E12 Agent 交互:步骤清单 / 方案选择 / 内联审批", () => {
|
||||
let e: E2E;
|
||||
let admin: Browser;
|
||||
|
||||
beforeAll(async () => {
|
||||
e = await createE2E("e12", {
|
||||
fakeScripts: [
|
||||
optionsScript(OPTIONS_GOAL),
|
||||
optionsScript(NONE_GOAL),
|
||||
optionsScript(TIMEOUT_GOAL),
|
||||
approvalScript(APPROVE_GOAL, "approve"),
|
||||
approvalScript(REJECT_GOAL, "reject"),
|
||||
],
|
||||
});
|
||||
const base = await seedBase(e);
|
||||
const a = await seedMovieTarget(e, base.siteId);
|
||||
T.approve = { slotId: a.slotId, releaseId: a.r1080 };
|
||||
const site2 = await e.db
|
||||
.insertInto("site")
|
||||
.values({
|
||||
key: "e12-second",
|
||||
name: "第二站",
|
||||
adapter: "nexusphp",
|
||||
base_url: "https://second.example/",
|
||||
rate_limit: JSON.stringify({ minIntervalMs: 0, concurrency: 5 }),
|
||||
})
|
||||
.returning("id")
|
||||
.executeTakeFirstOrThrow();
|
||||
const b = await seedMovieTarget(e, site2.id);
|
||||
T.reject = { slotId: b.slotId, releaseId: b.r1080 };
|
||||
// 控制台管理员会话(审批只接受它;它发起的会话才出会话内审批卡)
|
||||
const token = await ensureSetupToken(e.booted.ctx);
|
||||
admin = new Browser(e.app);
|
||||
const r = await admin.req("POST", "/auth/setup", { setupToken: token, username: "admin", password: PASSWORD });
|
||||
expect(r.status).toBe(200);
|
||||
}, 120_000);
|
||||
|
||||
afterAll(async () => {
|
||||
await e?.close();
|
||||
}, 60_000);
|
||||
|
||||
const waitStatus = (runId: string, statuses: string[], ms = 60_000) =>
|
||||
waitFor(
|
||||
async () => {
|
||||
const r = await runRow(e, runId);
|
||||
return statuses.includes(r.status) ? r : null;
|
||||
},
|
||||
ms,
|
||||
`run ${statuses.join("/")}`,
|
||||
);
|
||||
const waitEvent = (runId: string, type: string, ms = 60_000) =>
|
||||
waitFor(async () => (await runEvents(e, runId)).find((x) => x.type === type), ms, `事件 ${type}`);
|
||||
|
||||
it("方案选择:步骤清单先行、先看方案时执行被拒、挂起等待、选定后续跑", async () => {
|
||||
const created = await e.api("POST", "/agent/runs", { goal: OPTIONS_GOAL, proposeFirst: true });
|
||||
expect(created.status).toBe(201);
|
||||
expect(created.body.data.scope).toMatchObject({ proposeFirst: true });
|
||||
expect(created.body.data.mode).toBeUndefined();
|
||||
const runId = created.body.data.id as string;
|
||||
|
||||
const proposed = await waitEvent(runId, "options_proposed");
|
||||
const p = proposed.payload as Any;
|
||||
expect(p).toMatchObject({
|
||||
question: "收哪个版本?",
|
||||
recommendedId: "a",
|
||||
stepId: "s2",
|
||||
timeoutAction: "recommended",
|
||||
});
|
||||
expect(p.options.map((o: Any) => o.id)).toEqual(["a", "b"]);
|
||||
// 挂起:会话等待中,唤醒信号已登记(runAgent 在 waitForFact 里,不占 worker、不调用模型)
|
||||
const waiting = await waitFor(
|
||||
async () => {
|
||||
const r = await runRow(e, runId);
|
||||
return r.status === "waiting" && r.await_wake ? r : null;
|
||||
},
|
||||
30_000,
|
||||
"挂起等待方案",
|
||||
);
|
||||
expect(waiting.summary).toBe("等待选择方案");
|
||||
const decision = await e.db
|
||||
.selectFrom("decision")
|
||||
.selectAll()
|
||||
.where("id", "=", p.decisionId)
|
||||
.executeTakeFirstOrThrow();
|
||||
expect(decision).toMatchObject({ status: "awaiting_user", agent_run_id: runId, kind: "other" });
|
||||
|
||||
// 先看方案:选定前的 download.submit 被服务端拒绝
|
||||
const events = await runEvents(e, runId);
|
||||
const submitEnd = events.find(
|
||||
(x) => x.type === "tool_call_end" && (x.payload as Any).capabilityId === "download.submit",
|
||||
);
|
||||
expect(submitEnd?.payload).toMatchObject({ isError: true });
|
||||
expect(JSON.stringify(submitEnd?.payload)).toContain("先看方案");
|
||||
|
||||
// 选不存在的方案 → 422;选 a → 续跑
|
||||
expect(
|
||||
(await e.api("POST", `/agent/runs/${runId}/choice`, { decisionId: p.decisionId, optionId: "zzz" })).status,
|
||||
).toBe(422);
|
||||
const chosen = await e.api("POST", `/agent/runs/${runId}/choice`, { decisionId: p.decisionId, optionId: "a" });
|
||||
expect(chosen.status).toBe(200);
|
||||
// 重复作答 → 409
|
||||
expect(
|
||||
(await e.api("POST", `/agent/runs/${runId}/choice`, { decisionId: p.decisionId, optionId: "b" })).status,
|
||||
).toBe(409);
|
||||
const done = await waitStatus(runId, ["succeeded", "failed"]);
|
||||
expect(done.status).toBe("succeeded");
|
||||
expect(done.summary).toContain("方案结果:chosen/a/user/");
|
||||
expect(done.await_wake).toBeNull();
|
||||
|
||||
const all = await runEvents(e, runId);
|
||||
const resolved = all.find((x) => x.type === "options_resolved");
|
||||
expect(resolved?.payload).toMatchObject({ outcome: "chosen", choice: "a", by: "user", decisionId: p.decisionId });
|
||||
// 计划事件顺序:第一版步骤清单早于第一个业务工具调用;版本递增;最后全部完成
|
||||
const plans = all.filter((x) => x.type === "plan_updated");
|
||||
expect(plans.map((x) => (x.payload as Any).revision)).toEqual([1, 2, 3]);
|
||||
const firstBiz = all.find(
|
||||
(x) => x.type === "tool_call_start" && !String((x.payload as Any).capabilityId).startsWith("agent."),
|
||||
);
|
||||
expect(Number(plans[0]!.seq)).toBeLessThan(Number(firstBiz!.seq));
|
||||
expect(Number(proposed.seq)).toBeLessThan(Number(resolved!.seq));
|
||||
expect((plans.at(-1)!.payload as Any).steps.map((s: Any) => `${s.id}:${s.status}`)).toEqual([
|
||||
"s1:done",
|
||||
"s2:done",
|
||||
"s3:done",
|
||||
]);
|
||||
const d2 = await e.db.selectFrom("decision").selectAll().where("id", "=", p.decisionId).executeTakeFirstOrThrow();
|
||||
expect(d2.status).toBe("resolved");
|
||||
expect(d2.resolution).toMatchObject({ optionKey: "a", outcome: "chosen" });
|
||||
}, 120_000);
|
||||
|
||||
it("方案都不要:带原因退回,模型拿到 rejected 与原因", async () => {
|
||||
const runId = (await e.api("POST", "/agent/runs", { goal: NONE_GOAL })).body.data.id as string;
|
||||
const p = (await waitEvent(runId, "options_proposed")).payload as Any;
|
||||
await waitStatus(runId, ["waiting"], 30_000);
|
||||
// 都不要必须写原因
|
||||
expect(
|
||||
(await e.api("POST", `/agent/runs/${runId}/choice`, { decisionId: p.decisionId, optionId: null })).status,
|
||||
).toBe(422);
|
||||
const r = await e.api("POST", `/agent/runs/${runId}/choice`, {
|
||||
decisionId: p.decisionId,
|
||||
optionId: null,
|
||||
reason: "只要 WEB-DL 以外的",
|
||||
});
|
||||
expect(r.status).toBe(200);
|
||||
const done = await waitStatus(runId, ["succeeded", "failed"]);
|
||||
expect(done.summary).toContain("方案结果:rejected/none/user/只要 WEB-DL 以外的");
|
||||
const d = await e.db
|
||||
.selectFrom("decision")
|
||||
.select("status")
|
||||
.where("id", "=", p.decisionId)
|
||||
.executeTakeFirstOrThrow();
|
||||
expect(d.status).toBe("dismissed");
|
||||
}, 120_000);
|
||||
|
||||
it("方案超时:按设置自动选推荐项", async () => {
|
||||
await e.settings({ agent: { optionsTimeoutSec: 5, optionsTimeoutAction: "recommended" } });
|
||||
const runId = (await e.api("POST", "/agent/runs", { goal: TIMEOUT_GOAL })).body.data.id as string;
|
||||
const done = await waitStatus(runId, ["succeeded", "failed"], 90_000);
|
||||
expect(done.summary).toContain("方案结果:chosen/a/timeout/");
|
||||
const resolved = (await runEvents(e, runId)).find((x) => x.type === "options_resolved");
|
||||
expect(resolved?.payload).toMatchObject({ outcome: "chosen", choice: "a", by: "timeout" });
|
||||
await e.settings({ agent: { optionsTimeoutSec: 1800 } });
|
||||
}, 120_000);
|
||||
|
||||
it("旧客户端 mode=plan_only:按先看方案再执行处理", async () => {
|
||||
const r = await e.api("POST", "/agent/runs", { goal: "E12 旧客户端只给方案", mode: "plan_only" });
|
||||
expect(r.status).toBe(201);
|
||||
expect(r.body.data.scope).toMatchObject({ proposeFirst: true });
|
||||
const row = await runRow(e, r.body.data.id);
|
||||
expect(row.mode).toBe("execute");
|
||||
await e.api("POST", `/agent/runs/${r.body.data.id}/abort`, { reason: "测试" });
|
||||
});
|
||||
|
||||
it("内联审批:管理员会话发起,步骤处出审批卡;批准后重放执行,审批 consumed", async () => {
|
||||
const res = await admin.req("POST", "/agent/runs", { goal: APPROVE_GOAL });
|
||||
expect(res.status).toBe(201);
|
||||
const run = ((await res.json()) as Any).data;
|
||||
expect(run.scope).toMatchObject({ interactive: true });
|
||||
const runId = run.id as string;
|
||||
|
||||
const req = (await waitEvent(runId, "approval_requested")).payload as Any;
|
||||
expect(req).toMatchObject({
|
||||
capabilityId: "download.submit",
|
||||
planType: "acquisition_plan",
|
||||
riskLevel: "high",
|
||||
stepId: "s2",
|
||||
});
|
||||
expect(req.summary).toContain("沙丘2");
|
||||
await waitFor(
|
||||
async () => {
|
||||
const r = await runRow(e, runId);
|
||||
return r.status === "waiting" && r.summary === "等待审批" && r.await_wake ? r : null;
|
||||
},
|
||||
30_000,
|
||||
"挂起等待审批",
|
||||
);
|
||||
const approval = await e.db
|
||||
.selectFrom("approval")
|
||||
.selectAll()
|
||||
.where("id", "=", req.approvalId)
|
||||
.executeTakeFirstOrThrow();
|
||||
expect(approval).toMatchObject({ status: "pending", plan_id: req.planId, payload_hash: req.payloadHash });
|
||||
expect(approval.decision_id).not.toBeNull();
|
||||
|
||||
// API Key(开发密钥)不能审批:只接受控制台管理员会话
|
||||
expect(
|
||||
(await e.api("PATCH", `/approvals/${req.approvalId}`, { status: "approved", payloadHash: req.payloadHash }))
|
||||
.status,
|
||||
).toBe(403);
|
||||
const ok = await admin.req("PATCH", `/approvals/${req.approvalId}`, {
|
||||
status: "approved",
|
||||
payloadHash: req.payloadHash,
|
||||
});
|
||||
expect(ok.status).toBe(200);
|
||||
|
||||
// 提交后会话等下载 / 入库落定(waiting);模型本轮的汇报在 run_completed 里
|
||||
const completed = (await waitEvent(runId, "run_completed")).payload as Any;
|
||||
expect(completed.finalText).toMatch(/^已提交:[0-9a-f-]{36}$/);
|
||||
const events = await runEvents(e, runId);
|
||||
expect(events.find((x) => x.type === "approval_resolved")?.payload).toMatchObject({
|
||||
approvalId: req.approvalId,
|
||||
status: "approved",
|
||||
});
|
||||
const submitEnd = events.filter(
|
||||
(x) => x.type === "tool_call_end" && (x.payload as Any).capabilityId === "download.submit",
|
||||
);
|
||||
expect(submitEnd.at(-1)?.payload).toMatchObject({ isError: false, recovery: true });
|
||||
const a2 = await e.db.selectFrom("approval").selectAll().where("id", "=", req.approvalId).executeTakeFirstOrThrow();
|
||||
expect(a2.status).toBe("consumed");
|
||||
const inv = await e.db
|
||||
.selectFrom("capability_invocation")
|
||||
.selectAll()
|
||||
.where("approval_id", "=", req.approvalId)
|
||||
.executeTakeFirstOrThrow();
|
||||
expect(inv).toMatchObject({ capability_id: "download.submit", status: "succeeded" });
|
||||
const d = await e.db
|
||||
.selectFrom("decision")
|
||||
.selectAll()
|
||||
.where("id", "=", approval.decision_id!)
|
||||
.executeTakeFirstOrThrow();
|
||||
expect(d.status).toBe("resolved");
|
||||
}, 150_000);
|
||||
|
||||
it("内联审批:拒绝后调用以理由失败,待处理事项写结果与建议", async () => {
|
||||
const res = await admin.req("POST", "/agent/runs", { goal: REJECT_GOAL });
|
||||
const runId = ((await res.json()) as Any).data.id as string;
|
||||
const req = (await waitEvent(runId, "approval_requested")).payload as Any;
|
||||
await waitStatus(runId, ["waiting"], 30_000);
|
||||
const rej = await admin.req("PATCH", `/approvals/${req.approvalId}`, {
|
||||
status: "rejected",
|
||||
payloadHash: req.payloadHash,
|
||||
reason: "先别下,等 REMUX",
|
||||
});
|
||||
expect(rej.status).toBe(200);
|
||||
const completed = (await waitEvent(runId, "run_completed")).payload as Any;
|
||||
expect(completed.finalText).toContain("未提交");
|
||||
expect(completed.finalText).toContain("管理员拒绝了审批:先别下,等 REMUX");
|
||||
const events = await runEvents(e, runId);
|
||||
expect(events.find((x) => x.type === "approval_resolved")?.payload).toMatchObject({
|
||||
status: "rejected",
|
||||
reason: "先别下,等 REMUX",
|
||||
});
|
||||
const a = await e.db.selectFrom("approval").selectAll().where("id", "=", req.approvalId).executeTakeFirstOrThrow();
|
||||
expect(a.status).toBe("rejected");
|
||||
const d = await e.db.selectFrom("decision").selectAll().where("id", "=", a.decision_id!).executeTakeFirstOrThrow();
|
||||
expect(d.status).toBe("dismissed");
|
||||
expect(d.resolution).toMatchObject({
|
||||
approval: "rejected",
|
||||
reason: "先别下,等 REMUX",
|
||||
suggestion: "先别下,等 REMUX",
|
||||
});
|
||||
// 没有执行:没有这份审批的成功调用
|
||||
expect(
|
||||
await e.db.selectFrom("capability_invocation").select("id").where("approval_id", "=", req.approvalId).execute(),
|
||||
).toHaveLength(0);
|
||||
}, 150_000);
|
||||
});
|
||||
@@ -36,14 +36,17 @@ describe.skipIf(!hasPg)("E2 Agent 委托闭环(fake 模型 + 真实能力/授权
|
||||
}, 60_000);
|
||||
|
||||
const createRun = async () => {
|
||||
const r = await e.api("POST", "/agent/runs", { goal: GOAL, mode: "execute" });
|
||||
const r = await e.api("POST", "/agent/runs", { goal: GOAL });
|
||||
expect(r.status).toBe(201);
|
||||
expect(r.body.data).toMatchObject({ status: "queued", trigger: "user", mode: "execute" });
|
||||
expect(r.body.data).toMatchObject({ status: "queued", trigger: "user" });
|
||||
// 运行模式已取消(AG-2):API 不再返回 mode;API Key / 开发密钥发起的会话不出会话内审批卡
|
||||
expect(r.body.data.mode).toBeUndefined();
|
||||
expect(r.body.data.scope).not.toHaveProperty("interactive");
|
||||
return r.body.data.id as string;
|
||||
};
|
||||
|
||||
it("启动断言:25 个能力全部注册,并在能力目录里可见", async () => {
|
||||
expect(e.booted.ctx.capabilities.list()).toHaveLength(25);
|
||||
it("启动断言:27 个能力全部注册,并在能力目录里可见", async () => {
|
||||
expect(e.booted.ctx.capabilities.list()).toHaveLength(27);
|
||||
const r = await e.api("GET", "/capabilities");
|
||||
expect(r.body.data.items.map((c: { id: string }) => c.id).sort()).toEqual(
|
||||
e.booted.ctx.capabilities
|
||||
|
||||
@@ -64,6 +64,8 @@ export interface E2E {
|
||||
lib: TempTree;
|
||||
cache: TempTree;
|
||||
sessionDir: string;
|
||||
/** Hono 应用(管理员会话用例用 cookie 调用,见 auth-settings/harness.ts 的 Browser) */
|
||||
app: ReturnType<typeof createApp>;
|
||||
api(method: string, path: string, body?: unknown): Promise<{ status: number; body: Loose }>;
|
||||
/** PATCH /settings 并断言成功 */
|
||||
settings(patch: Record<string, unknown>): Promise<Loose>;
|
||||
@@ -164,6 +166,7 @@ export async function createE2E(name: string, opts: E2EOptions = {}): Promise<E2
|
||||
lib,
|
||||
cache,
|
||||
sessionDir,
|
||||
app,
|
||||
async api(method, path, body) {
|
||||
const res = await app.request(`/api/v1${path}`, {
|
||||
method,
|
||||
|
||||
@@ -22,8 +22,8 @@ import {
|
||||
type AgentRuntimePort,
|
||||
type CreateRunSpec,
|
||||
emptyUsage,
|
||||
type InjectedEventBody,
|
||||
type RunOutcome,
|
||||
type SubagentEventBody,
|
||||
} from "../runtime/port.ts";
|
||||
import { type ResolvedAgentSettings, resolveAgentSettings } from "../settings.ts";
|
||||
import { compactionFromSettings, type PiModelResolver, resolveModelFromSettings } from "./model.ts";
|
||||
@@ -58,6 +58,8 @@ export const DEFAULT_COMPACTION_INSTRUCTIONS =
|
||||
"保留:本次目标、约束与授权范围、证据引用(对象 ID/记录)、已完成的动作及其结果(已提交≠已入库)、剩余事项。不要保留任何凭证。";
|
||||
|
||||
const LANE = "main";
|
||||
/** 可挂起重放的能力:首次执行时写入的调用级 memo(重放时据此区分) */
|
||||
const STARTED_MEMO = "umt.started";
|
||||
const SUMMARY_LIMIT = 600;
|
||||
|
||||
/** Pi(pi-agent-core AgentHarness)实现的 AgentRuntimePort */
|
||||
@@ -152,15 +154,32 @@ class PiAgentRun implements AgentRunHandle {
|
||||
const { descriptor } = entry;
|
||||
const registry = this.options.registry;
|
||||
const actor = this.spec.actor;
|
||||
const gate = this.spec.replayGate;
|
||||
// 可挂起重放(见 CreateRunSpec.replayGate):对 harness 声明 safe,执行时用 memo 判断是否重放
|
||||
const gated = descriptor.replay === "never" && !!gate;
|
||||
return {
|
||||
name: entry.toolName,
|
||||
label: descriptor.title,
|
||||
description: `${descriptor.description}\n[能力 ${descriptor.id} v${descriptor.version};类别 ${descriptor.kind};风险 ${descriptor.riskLevel}]`,
|
||||
parameters: registry.inputJsonSchema(descriptor.id) as unknown as TSchema,
|
||||
replay: descriptor.replay,
|
||||
replay: gated ? "safe" : descriptor.replay,
|
||||
...(descriptor.kind === "execute" ? { executionMode: "sequential" as const } : {}),
|
||||
execute: async (toolCallId, params, _onUpdate, _toolCtx, invocation, context) => {
|
||||
const signal = context.abortSignal ?? this.abortController.signal;
|
||||
const invocationId = invocation.invocationId ?? toolCallId;
|
||||
if (gated) {
|
||||
const replayed = (await invocation.getMemo(STARTED_MEMO)) === true;
|
||||
if (!replayed) await invocation.setMemo(STARTED_MEMO, true);
|
||||
else if (!gate(descriptor.id, invocationId))
|
||||
throw new Error(
|
||||
JSON.stringify({
|
||||
error: {
|
||||
code: "interrupted",
|
||||
message: "执行被中断,外部结果未知;先用查询工具核对实际状态,不要直接重复提交",
|
||||
},
|
||||
}),
|
||||
);
|
||||
}
|
||||
try {
|
||||
const output = await registry.invoke(descriptor.id, params, {
|
||||
actorId: actor.actorId,
|
||||
@@ -169,7 +188,7 @@ class PiAgentRun implements AgentRunHandle {
|
||||
...(actor.allowedKinds ? { allowedKinds: actor.allowedKinds } : {}),
|
||||
signal,
|
||||
source: "agent",
|
||||
invocationId: invocation.invocationId ?? toolCallId,
|
||||
invocationId,
|
||||
agentRunId: this.runId,
|
||||
});
|
||||
return {
|
||||
@@ -303,7 +322,7 @@ class PiAgentRun implements AgentRunHandle {
|
||||
await this.lane.abort(BACKGROUND_CONTEXT).catch(() => undefined);
|
||||
}
|
||||
|
||||
appendEvent(body: SubagentEventBody): boolean {
|
||||
appendEvent(body: InjectedEventBody): boolean {
|
||||
if (this.closed) return false;
|
||||
this.emit(body);
|
||||
return true;
|
||||
|
||||
@@ -16,6 +16,25 @@ export type SubagentEventBody = Extract<
|
||||
{ type: "subagent_spawned" | "subagent_event" | "subagent_completed" }
|
||||
>;
|
||||
|
||||
/**
|
||||
* 由服务端(而不是模型 / harness)产生、追加进会话事件流的事件:子会话镜像与会话交互
|
||||
* (步骤清单、方案选择、审批卡)。与运行时自身事件共用 seq(单写者)。
|
||||
*/
|
||||
export type InjectedEventBody = Extract<
|
||||
AgentRunEventBody,
|
||||
{
|
||||
type:
|
||||
| "subagent_spawned"
|
||||
| "subagent_event"
|
||||
| "subagent_completed"
|
||||
| "plan_updated"
|
||||
| "options_proposed"
|
||||
| "options_resolved"
|
||||
| "approval_requested"
|
||||
| "approval_resolved";
|
||||
}
|
||||
>;
|
||||
|
||||
/** 调用能力时使用的业务身份;授权回调据此到 automation_grant / approval 查询 */
|
||||
export interface AgentActor {
|
||||
actorId: string;
|
||||
@@ -47,6 +66,12 @@ export interface CreateRunSpec {
|
||||
* 运行时在每次工具调用与每轮请求前把它加到自己的用量上再比预算。由调用方缓存,须同步返回。
|
||||
*/
|
||||
externalUsage?: () => ExternalUsage;
|
||||
/**
|
||||
* 挂起后可重放的"不可重放"能力(replay=never,如 download.submit 等审批)。给了它时,这类能力对 harness 声明为
|
||||
* replay=safe,真正执行前用调用级 memo 区分首次与重放:重放时只有 replayGate 返回 true(本次恢复正是为了它,
|
||||
* 如审批已决定)才重新调用注册中心(授权 / 账本照常),否则按"执行被中断、外部结果未知"返回错误(与 never 语义一致)。
|
||||
*/
|
||||
replayGate?: (capabilityId: string, invocationId: string) => boolean;
|
||||
}
|
||||
|
||||
export interface ExternalUsage {
|
||||
@@ -80,10 +105,10 @@ export interface AgentRunHandle {
|
||||
resume(): Promise<RunOutcome>;
|
||||
abort(reason?: string): Promise<void>;
|
||||
/**
|
||||
* 往本会话事件流追加一条子代理事件(subagent_*),与运行时自身事件共用 seq 计数(单写者)。
|
||||
* 往本会话事件流追加一条服务端事件(子会话镜像 / 步骤清单 / 方案 / 审批),与运行时自身事件共用 seq 计数(单写者)。
|
||||
* 会话已关闭时返回 false,调用方自行落库。
|
||||
*/
|
||||
appendEvent(body: SubagentEventBody): boolean;
|
||||
appendEvent(body: InjectedEventBody): boolean;
|
||||
/** 本会话自身已用量(不含子会话) */
|
||||
stats(): { toolCalls: number; usage: AgentUsageTotals };
|
||||
compact(customInstructions?: string): Promise<void>;
|
||||
|
||||
@@ -5,12 +5,46 @@ export interface SystemPromptInput {
|
||||
goal: string;
|
||||
/** 附加约束(范围、预算、"只给方案不执行"等) */
|
||||
constraints?: readonly string[];
|
||||
/** 会话交互(AG-2):按本次可用的能力与会话选项生成"工作方式"一节 */
|
||||
interaction?: {
|
||||
/** 提供了 agent.plan */
|
||||
plan?: boolean;
|
||||
/** 提供了 agent.propose_options */
|
||||
options?: boolean;
|
||||
/** 会话选项"先看方案再执行" */
|
||||
proposeFirst?: boolean;
|
||||
/** 需审批的能力在步骤处出审批卡(控制台管理员会话) */
|
||||
approvals?: boolean;
|
||||
};
|
||||
}
|
||||
|
||||
/** "工作方式":自动化优先——先出步骤清单再执行,执行中更新;有取舍时给方案;需授权的动作等审批 */
|
||||
export function interactionRules(i: NonNullable<SystemPromptInput["interaction"]>): string[] {
|
||||
const out: string[] = [];
|
||||
if (i.plan)
|
||||
out.push(
|
||||
"开始时先调用 agent.plan 列出 3–7 个步骤(id 用 s1、s2…),再执行;每开始、完成、跳过或失败一步都调用 agent.plan 更新状态,只传变化的步骤;计划变了就增改步骤。",
|
||||
);
|
||||
if (i.options)
|
||||
out.push(
|
||||
"遇到有明显取舍的决定(选哪个资源、画质与体积、是否覆盖已有版本)时,用 agent.propose_options 给 2–3 个方案并标出推荐项,按用户选定的方案继续;用户都不要时按他说明的原因调整。不要为小事询问。",
|
||||
);
|
||||
if (i.proposeFirst)
|
||||
out.push(
|
||||
"本会话要求先看方案再执行:先只做调查,再用 agent.propose_options 给出方案;用户选定前不要调用任何有副作用的能力(提交下载、执行整理、发布策略、重试或取消任务),服务端也会拒绝。",
|
||||
);
|
||||
else out.push("默认直接执行:在授权范围内按步骤推进,不必事先征求同意。");
|
||||
if (i.approvals)
|
||||
out.push(
|
||||
"需要审批的动作(提交下载、执行整理、发布策略)在没有持续授权时会在对应步骤出审批卡,会话等待管理员批准或拒绝;被拒绝时按拒绝理由调整方案,不要换参数重试同一动作。",
|
||||
);
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* 系统提示模板。硬性规则不可被用户消息、工具输出、网页/RSS 内容覆盖。
|
||||
*/
|
||||
export function buildSystemPrompt({ role, goal, constraints = [] }: SystemPromptInput): string {
|
||||
export function buildSystemPrompt({ role, goal, constraints = [], interaction }: SystemPromptInput): string {
|
||||
const rules = [
|
||||
"只用提供的工具操作业务;没有对应能力就说做不到,不要假装执行。",
|
||||
"结论必须有证据:引用工具返回的对象 ID、记录或数值;没有证据就写“未确认”。",
|
||||
@@ -31,6 +65,8 @@ export function buildSystemPrompt({ role, goal, constraints = [] }: SystemPrompt
|
||||
"## 必须遵守",
|
||||
...rules.map((r, i) => `${i + 1}. ${r}`),
|
||||
];
|
||||
const ways = interaction ? interactionRules(interaction) : [];
|
||||
if (ways.length > 0) lines.push("", "## 工作方式", ...ways.map((r) => `- ${r}`));
|
||||
if (constraints.length > 0) {
|
||||
lines.push("", "## 本次任务约束", ...constraints.map((c) => `- ${c}`));
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user