feat(agent): budget reminders, usage while waiting, budget-exceeded status and dedupe

- tool results carry a budget reminder near limits; text before the last tool
  call is dropped as report when cut by budget/abort
- usage persisted when a run suspends, so waiting sessions show spent tokens
- task page shows budget_exceeded via business.outcome instead of completed
- budget-exceeded decisions merge by root cause (decision / schedule / goal
  template); new summaries added to investigate allowlist

Co-Authored-By: Claude Opus 5.5 <[email protected]>

apps/server/src/modules/agent/runner.ts

@@ -59,6 +59,9 @@ export const INVESTIGATE_CAPABILITIES = [
   "release.inspect",
   "inventory.query",
   "task.inspect",
+  "download.inspect",
+  "decision.list",
+  "subscription.list",
   "identity.resolve",
 ] as const;
 /**
@@ -266,7 +269,8 @@ function systemPromptFor(
   if (sub.canSpawn)
     constraints.push(
       `可以用 agent.spawn 把彼此独立的有界子目标交给子会话（同时最多 ${sub.maxConcurrent} 个，派生层数上限 ${sub.maxDepth}）；` +
-        "子会话的能力、模式与预算都不会超过本会话，且会占用本会话预算；派生后用 agent.await 等结果再汇总，不要为小事派生。",
+        "子会话的能力、模式与预算都不会超过本会话，且会占用本会话预算；budget 的每一项都必须小于本会话剩余预算（超过会被拒绝，不给则按剩余的一半划拨）；" +
+        "派生后用 agent.await 等结果再汇总，不要为小事派生。",
     );
   if (scope.capabilities.includes("web.search"))
     constraints.push(
@@ -506,6 +510,8 @@ export async function drive(
   } finally {
     clearInterval(watch);
     signal.removeEventListener("abort", abortOnce);
+    // 挂起等待(审批 / 选方案 / 子会话)时没有 run_completed 事件:把已用累计写回,等待期间界面照样显示 token / 费用
+    await persistUsage(ctx, runId, handle.stats().usage).catch(() => undefined);
     await liveRun.stop();
     await handle.close();
   }
@@ -543,6 +549,17 @@ export async function drive(
   }
 }
 
+/** 用量累计写回 agent_run.usage:按字段取较大值(重开会话后运行时计数可能从 0 开始,不能倒退) */
+export async function persistUsage(ctx: ServerContext, runId: string, usage: Record<string, unknown>) {
+  const nums = Object.fromEntries(Object.entries(usage).filter(([, v]) => typeof v === "number" && v > 0));
+  if (Object.keys(nums).length === 0) return;
+  await sql`
+    update agent_run set usage = coalesce(usage, '{}'::jsonb) || (
+      select coalesce(jsonb_object_agg(k, greatest(v::float8, coalesce((usage->>k)::float8, 0))), '{}'::jsonb)
+        from jsonb_each_text(${JSON.stringify(nums)}::jsonb) as n(k, v))
+     where id = ${runId}`.execute(ctx.db);
+}
+
 /** 等待目标关联的下载/入库落定:返回报告(state≠waiting)或 null(继续等) */
 export async function goalSettled(env: AgentEnv, runId: string, modelText: string): Promise<AgentGoalReport | null> {
   const r = await buildGoalReport(env.ctx.db, runId, modelText);

apps/server/src/modules/agent/service.ts

@@ -209,7 +209,15 @@ async function writeDecision(
       values (${d.kind}, 'warning', ${d.title}, ${d.summary}, 'agent_run', ${d.subjectId}, ${d.dedupeKey},
               ${JSON.stringify(d.evidence)}::jsonb, ${d.subjectId})
       on conflict (dedupe_key) where status in ('open', 'investigating', 'awaiting_user')
-      do update set attempts = decision.attempts + 1, evidence = excluded.evidence, updated_at = now()
+      do update set attempts = decision.attempts + 1, summary = excluded.summary,
+                    agent_run_id = excluded.agent_run_id, updated_at = now(),
+                    -- 同根因合并:保留历次 run id(最多 50 个),其余证据取最新
+                    evidence = decision.evidence || excluded.evidence || jsonb_build_object('runIds',
+                      (select coalesce(jsonb_agg(u.x), '[]'::jsonb) from (
+                         select v.x from (
+                           select jsonb_array_elements(coalesce(decision.evidence->'runIds', '[]'::jsonb))
+                           union select to_jsonb(${d.subjectId}::text)) v(x)
+                         limit 50) u))
       returning id, (xmax = 0) as inserted`.execute(trx);
     const row = r.rows[0]!;
     if (row.inserted)
@@ -222,6 +230,32 @@ async function writeDecision(
   });
 }
 
+/** 目标模板归一化:去掉时间戳、UUID、数字与空白,同一模板渲染出的目标得到同一个键 */
+export function goalTemplateKey(goal: string): string {
+  const norm = goal
+    .replace(/\d{4}-\d{2}-\d{2}[T ][\d:.]+Z?/g, "<t>")
+    .replace(/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/gi, "<id>")
+    .replace(/\d+/g, "<n>")
+    .replace(/\s+/g, "")
+    .toLowerCase();
+  return sha(norm).slice(0, 16);
+}
+
+/**
+ * 超预算待处理事项的去重键(按根因):
+ *   后台调查 → 被调查事项的去重键(同一下载 / 同一身份的新事项也算同根因);定时巡检 → 定时任务;
+ *   其余 → 目标模板。
+ */
+export async function budgetDedupeKey(ctx: ServerContext, run: AgentRunRow): Promise<string> {
+  if (run.dedupe_key?.startsWith("investigate:")) {
+    const id = run.dedupe_key.slice("investigate:".length);
+    const d = await ctx.db.selectFrom("decision").select("dedupe_key").where("id", "=", id).executeTakeFirst();
+    return `agent_budget:root:${d?.dedupe_key ?? `decision:${id}`}`;
+  }
+  if (run.created_by?.startsWith("schedule:")) return `agent_budget:${run.created_by}`;
+  return `agent_budget:goal:${goalTemplateKey(run.goal)}`;
+}
+
 /** 终态:按业务表汇总的报告决定状态;预算超限 → budget_exceeded + 待处理;调查运行回写 decision */
 export async function finalize(
   env: AgentEnv,
@@ -316,13 +350,14 @@ export async function finalize(
       .execute();
   }
   if (status === "budget_exceeded") {
+    // 同根因(同一待处理事项 / 同一定时巡检 / 同一目标模板)超预算合并到已有事项,不重复新建
     await writeDecision(ctx, {
       kind: "other",
       title: "Agent 运行超出预算",
       summary: res.message ?? "超出预算",
       subjectId: runId,
-      dedupeKey: `agent_budget:${runId}`,
-      evidence: { runId, budget: res.budget, message: res.message, goal: run.goal },
+      dedupeKey: await budgetDedupeKey(ctx, run),
+      evidence: { runId, runIds: [runId], budget: res.budget, message: res.message, goal: run.goal },
     });
   }
   // 后台调查:结论回写待处理事项,交还人工(仍可被新证据重新唤醒)

apps/server/src/modules/auth-settings/engine-business.ts

@@ -218,10 +218,12 @@ export async function resolveBusiness(
     }),
   );
   const agentIds = [...(byKind.get("agent") ?? [])];
-  const agents = new Map<string, { trigger: string; created_by: string | null }>();
+  const agents = new Map<string, { trigger: string; created_by: string | null; status: string }>();
   if (agentIds.length) {
-    const r = await sql<{ id: string; trigger: string; created_by: string | null }>`
-      select id::text as id, trigger, created_by from agent_run where id::text = any(${agentIds}::text[])`.execute(db);
+    const r = await sql<{ id: string; trigger: string; created_by: string | null; status: string }>`
+      select id::text as id, trigger, created_by, status from agent_run where id::text = any(${agentIds}::text[])`.execute(
+      db,
+    );
     for (const a of r.rows) agents.set(a.id, a);
   }
 
@@ -240,6 +242,7 @@ export async function resolveBusiness(
         agentCreatedBy: agent?.created_by ?? null,
         fromOutbox: !!r.idempotency_key && outboxKeys.has(r.idempotency_key),
       }),
+      ...(agent ? { outcome: agent.status } : {}),
     });
   }
   return out;

apps/server/test/e2e/e12-interaction.test.ts

@@ -229,6 +229,8 @@ describe.skipIf(!hasPg)("E12 Agent 交互:步骤清单 / 方案选择 / 内联
       "挂起等待方案",
     );
     expect(waiting.summary).toBe("等待选择方案");
+    // AG-3:挂起时没有 run_completed,已用累计照样写回(等待期间界面显示已用 token / 费用,而不是"未报告")
+    expect(Number((waiting.usage as Any).totalTokens ?? 0)).toBeGreaterThan(0);
     const decision = await e.db
       .selectFrom("decision")
       .selectAll()

apps/server/test/system/system.test.ts

@@ -427,7 +427,14 @@ describe.skipIf(!hasPg)("W9-I 进程与健康(真实 PG)", () => {
     const list = await json(await admin("GET", "/engine/runs?from=2026-09-10T00:00:00Z&to=2026-09-11T00:00:00Z"));
     const by = new Map(list.data.items.map((r: { id: string; business: unknown }) => [r.id, r.business]));
     expect(by.get("b1")).toEqual({ kind: "feed", refId: feedId, title: "HDSky RSS", triggeredBy: "event" });
-    expect(by.get("b2")).toEqual({ kind: "agent", refId: agentId, title: "巡检:待处理", triggeredBy: "schedule" });
+    expect(by.get("b2")).toEqual({
+      kind: "agent",
+      refId: agentId,
+      title: "巡检:待处理",
+      triggeredBy: "schedule",
+      // AG-3:Agent 会话带业务结果(任务页据此显示超出预算 / 失败,而不是引擎的「完成」)
+      outcome: "queued",
+    });
     expect(by.get("b3")).toEqual({
       kind: "organization",
       refId: "00000000-0000-0000-0000-0000000000aa",

apps/web/app/components/tasks/labels.test.ts

@@ -0,0 +1,25 @@
+import { describe, expect, it } from "vitest";
+import { engineRunStatus } from "./labels.ts";
+
+describe("任务行状态", () => {
+  const agent = (outcome: string | null) => ({
+    kind: "agent" as const,
+    refId: "r1",
+    title: null,
+    triggeredBy: null,
+    outcome,
+  });
+  it("引擎跑完但 Agent 会话超出预算 / 失败:显示业务结果,不显示「完成」", () => {
+    expect(engineRunStatus({ status: "completed", business: agent("budget_exceeded") })).toEqual({
+      label: "超出预算",
+      tone: "warning",
+    });
+    expect(engineRunStatus({ status: "completed", business: agent("failed") }).label).toBe("失败");
+    expect(engineRunStatus({ status: "completed", business: agent("succeeded") }).label).toBe("完成");
+  });
+  it("引擎自身失败 / 进行中照旧;没有业务对象照旧", () => {
+    expect(engineRunStatus({ status: "failed", business: agent("budget_exceeded") }).label).toBe("失败");
+    expect(engineRunStatus({ status: "running", business: agent("running") }).label).toBe("进行中");
+    expect(engineRunStatus({ status: "completed", business: null }).label).toBe("完成");
+  });
+});

apps/web/app/components/tasks/labels.ts

@@ -69,6 +69,22 @@ const RUN_STATUS: Record<string, { label: string; tone: StatusTone }> = {
   canceled: { label: "已取消", tone: "warning" },
 };
 export const runStatus = (s: string) => RUN_STATUS[s] ?? { label: s, tone: "neutral" as StatusTone };
+/**
+ * 任务行状态:引擎 run 正常结束但业务结果不是成功时以业务结果为准(Agent 会话超出预算 / 失败 / 已停止),
+ * 不把"引擎跑完"显示成「完成」
+ */
+const OUTCOME_STATUS: Record<string, { label: string; tone: StatusTone }> = {
+  budget_exceeded: { label: "超出预算", tone: "warning" },
+  failed: { label: "失败", tone: "destructive" },
+  cancelled: { label: "已停止", tone: "warning" },
+  waiting: { label: "等待", tone: "neutral" },
+};
+export function engineRunStatus(r: Pick<EngineRun, "status" | "business">) {
+  const outcome = r.business?.outcome;
+  if (outcome && (r.status === "completed" || r.status === "succeeded") && OUTCOME_STATUS[outcome])
+    return OUTCOME_STATUS[outcome];
+  return runStatus(r.status);
+}
 export const RUN_STATUS_OPTIONS = ["pending", "running", "completed", "failed", "canceled"].map((value) => ({
   value,
   label: runStatus(value).label,

apps/web/app/components/tasks/task-detail.tsx

@@ -15,7 +15,15 @@ import { Button } from "~/components/ui/button.tsx";
 import { errorMessage } from "~/lib/api.ts";
 import { formatDuration, formatRelative, spanSec } from "~/lib/labels.ts";
 import { cn, formatDateTime } from "~/lib/utils.ts";
-import { BUSINESS_KIND_LABEL, businessLink, runStatus, SOURCE_LABEL, sourceOf, workflowLabel } from "./labels.ts";
+import {
+  BUSINESS_KIND_LABEL,
+  businessLink,
+  engineRunStatus,
+  runStatus,
+  SOURCE_LABEL,
+  sourceOf,
+  workflowLabel,
+} from "./labels.ts";
 
 // 任务详情:业务摘要 → 时间线 → 引擎步骤(折叠)→ 操作。任务页抽屉与 /tasks/:id 共用。
 
@@ -125,7 +133,7 @@ function businessTimeline(r: EngineRunDetail): TimelineItem[] {
       });
   }
   if (r.finishedAt) {
-    const s = runStatus(r.status);
+    const s = engineRunStatus(r);
     items.push({
       id: "finished",
       at: r.finishedAt,
@@ -337,7 +345,7 @@ export function TaskDetailBody({ r, onOpenRun }: { r: EngineRunDetail; onOpenRun
 
 /** 头部一行:状态 + 业务类型 */
 export function TaskStatusLine({ r }: { r: EngineRunDetail }) {
-  const s = runStatus(r.status);
+  const s = engineRunStatus(r);
   return (
     <span className="flex flex-wrap items-center gap-2">
       <StatusBadge tone={s.tone}>{s.label}</StatusBadge>

apps/web/app/components/tasks/task-stream.tsx

@@ -18,8 +18,8 @@ import {
   BUSINESS_KINDS,
   businessLink,
   ENGINE_ACTIVE,
+  engineRunStatus,
   RUN_STATUS_OPTIONS,
-  runStatus,
   runTitle,
   SOURCE_LABEL,
   SOURCE_TRIGGERED_BY,
@@ -247,7 +247,7 @@ export function TaskStream({
             key: "status",
             header: "状态",
             cell: (r) => {
-              const s = runStatus(r.status);
+              const s = engineRunStatus(r);
               return <StatusBadge tone={s.tone}>{s.label}</StatusBadge>;
             },
           },

apps/web/app/routes/console/home.tsx

@@ -35,7 +35,7 @@ import { StatusBadge, type StatusTone } from "~/components/console/status-badge.
 import { useVolumeLabel } from "~/components/downloads/cache-volumes.tsx";
 import { countdown, RUN_STATUS } from "~/components/schedules/labels.ts";
 import { useLiveSchedules, useNow } from "~/components/schedules/schedule-board.tsx";
-import { runStatus, runTitle, workflowLabel } from "~/components/tasks/labels.ts";
+import { engineRunStatus, runTitle, workflowLabel } from "~/components/tasks/labels.ts";
 import { useLiveTasks } from "~/components/tasks/task-stream.tsx";
 import { Button } from "~/components/ui/button.tsx";
 import { Switch } from "~/components/ui/switch.tsx";
@@ -422,7 +422,7 @@ function RecentRuns() {
       ) : (
         <ul className="divide-y divide-border/60">
           {rows.map((r) => {
-            const s = runStatus(r.status);
+            const s = engineRunStatus(r);
             return (
               <Row key={r.id}>
                 <Link to={`/console/tasks?run=${encodeURIComponent(r.id)}`} className="min-w-0 flex-1 hover:underline">

packages/agent/src/pi/fake.ts

@@ -202,8 +202,10 @@ function toolResults(messages: Message[]): FakeToolResult[] {
     if (m.role !== "toolResult") continue;
     const text = messageText(m);
     let json: unknown;
+    // 第一块文本是能力输出;运行时可能在后面追加【预算提醒】等附注块
+    const first = (m.content as Array<{ type: string; text?: string }>).find((c) => c.type === "text")?.text ?? text;
     try {
-      json = JSON.parse(text);
+      json = JSON.parse(first);
     } catch {
       json = undefined;
     }

packages/agent/src/pi/runtime.ts

@@ -137,6 +137,8 @@ class PiAgentRun implements AgentRunHandle {
   private toolCalls = 0;
   private usage: AgentUsageTotals = emptyUsage();
   private finalText = "";
+  /** finalText 之后没有再调用工具(是收尾文字而不是"接下来我要…"的中间过程) */
+  private finalIsTail = false;
   private budget: BudgetState | undefined;
   private userAbort: string | undefined;
   private startedAt: number | undefined;
@@ -201,12 +203,19 @@ class PiAgentRun implements AgentRunHandle {
             agentRunId: this.runId,
             modelImages: this.supportsImages,
           });
-          return { content: toolContent(output, this.supportsImages), details: undefined };
+          const content = toolContent(output, this.supportsImages);
+          const notice = this.budgetNotice();
+          if (notice) content.push({ type: "text", text: notice });
+          return { content, details: undefined };
         } catch (error) {
+          const notice = this.budgetNotice();
+          const tail = notice ? `\n${notice}` : "";
           if (isCapabilityError(error)) {
-            throw new Error(JSON.stringify({ error: { code: error.code, message: redactString(error.message) } }));
+            throw new Error(
+              JSON.stringify({ error: { code: error.code, message: redactString(error.message) } }) + tail,
+            );
           }
-          throw new Error(redactString(error instanceof Error ? error.message : String(error)));
+          throw new Error(redactString(error instanceof Error ? error.message : String(error)) + tail);
         }
       },
     };
@@ -258,9 +267,13 @@ class PiAgentRun implements AgentRunHandle {
     });
     events.on("message_end", (e) => {
       const text = assistantText(e.message);
-      if (text?.trim()) this.finalText = text;
+      if (text?.trim()) {
+        this.finalText = text;
+        this.finalIsTail = true;
+      }
     });
     events.on("tool_start", (e) => {
+      this.finalIsTail = false;
       if (this.rejected.has(e.toolCallId)) return;
       const capabilityId = this.allowed.get(e.toolName);
       if (!capabilityId) {
@@ -408,7 +421,8 @@ class PiAgentRun implements AgentRunHandle {
       status: "failed",
       reason,
       message,
-      finalText: this.finalText,
+      // 被预算 / 中止打断时,工具调用之前的文字只是中间过程("我再查一下…"),不作为汇报
+      finalText: (reason === "budget" || reason === "aborted") && !this.finalIsTail ? "" : this.finalText,
       usage: this.usage,
       ...(budget ? { budget } : {}),
     };
@@ -435,6 +449,28 @@ class PiAgentRun implements AgentRunHandle {
     }, 0);
   }
 
+  /**
+   * 预算将尽提醒(AG-3):工具调用只剩 ≤2 次,或 token / 费用 / 时长用到 85%,在工具结果末尾追加一句,
+   * 让模型停止探查、先写中文小结,避免被硬截断后只剩中间过程。
+   */
+  private budgetNotice(): string | null {
+    const out: string[] = [];
+    const ext = this.spec.externalUsage?.();
+    const left = this.settings.maxToolCalls - this.toolCalls - (ext?.toolCalls ?? 0);
+    if (left <= 2) out.push(`工具调用只剩 ${Math.max(0, left)} 次`);
+    const cost = this.settings.costBudget;
+    const tokens = this.usage.totalTokens + (ext?.totalTokens ?? 0);
+    const usd = this.usage.costUsd + (ext?.costUsd ?? 0);
+    if (cost?.maxTotalTokens !== undefined && tokens >= cost.maxTotalTokens * 0.85)
+      out.push(`token 已用 ${tokens}/${cost.maxTotalTokens}`);
+    if (cost?.maxCostUsd !== undefined && usd >= cost.maxCostUsd * 0.85)
+      out.push(`费用已用 $${usd.toFixed(4)}/$${cost.maxCostUsd}`);
+    if (this.startedAt !== undefined && Date.now() - this.startedAt >= this.settings.maxDurationMs * 0.85)
+      out.push("运行时长接近上限");
+    if (out.length === 0) return null;
+    return `【预算提醒】${out.join("；")}。停止探查，下一条回复直接用中文写小结（已完成什么、证据、还缺什么）。`;
+  }
+
   /** token / 费用预算:自身用量 + 子会话用量(externalUsage) */
   private checkCost(): void {
     const cost = this.settings.costBudget;

packages/agent/test/runtime.test.ts

@@ -270,6 +270,34 @@ describe("PiAgentRuntime + FakeModelProvider", () => {
     expect(fx.audits.filter((a) => a.capabilityId === "inventory.query")).toHaveLength(2);
   });
 
+  it("预算将尽:工具结果追加【预算提醒】;被预算截断时中间过程文字不作为汇报", async () => {
+    const fx = makeFixture();
+    const loop: FakeScript = {
+      match: "Dune",
+      steps: Array.from({ length: 10 }, () => ({
+        text: "Let me check the inventory again.",
+        toolCalls: [{ capability: "inventory.query", input: { title: "Dune" } }],
+      })),
+      finalText: "不该到这里",
+    };
+    const { run, fake } = await start(fx, [loop], { settings: { maxToolCalls: 4 } });
+    const out = await run.prompt();
+    expect(out).toMatchObject({ status: "failed", reason: "budget", budget: "tool_calls", finalText: "" });
+    expect(fake.lastRequest?.transcript).toContain("【预算提醒】工具调用只剩 2 次");
+  });
+
+  it("预算:收尾文字在最后一次工具调用之后 → 被截断时仍保留为汇报", async () => {
+    const fx = makeFixture();
+    const script: FakeScript = {
+      match: "Dune",
+      steps: [{ toolCalls: [{ capability: "inventory.query", input: { title: "Dune" } }] }],
+      finalText: "小结:库里没有 Dune。",
+    };
+    const { run } = await start(fx, [script], { settings: { maxToolCalls: 5 } });
+    const out = await run.prompt();
+    expect(out).toMatchObject({ status: "completed", finalText: "小结:库里没有 Dune。" });
+  });
+
   it("预算:运行时长超限 → run_failed(budget/duration)", async () => {
     const fx = makeFixture({ slowMs: 5_000 });
     fx.grants.set("user:1", new Set(["download.submit"]));

packages/contracts/src/ops.ts

@@ -270,6 +270,11 @@ export const EngineRunBusiness = z.object({
   /** 展示名(订阅/作品标题、下载任务名、Feed/站点名、会话目标等);对象已删除时为 null */
   title: z.string().nullable(),
   triggeredBy: EngineTriggeredBy.nullable(),
+  /**
+   * 业务结果(与引擎状态不同时才有意义):Agent 会话为 agent_run.status。引擎 run 正常结束而会话超出预算 / 失败时,
+   * 任务页按它显示「超出预算」/「失败」,不显示「完成」
+   */
+  outcome: z.string().nullable().optional(),
 });
 export type EngineRunBusiness = z.infer<typeof EngineRunBusiness>;