OrgTJWater/next-tjwater-agent
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(tasks): enforce dynamic execution plans

Browse Source
This commit is contained in:
jiang
2026-09-24 16:36:12 +08:00
parent 485ceff7e3
commit e0d37f9da9
13 changed files with 1942 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/routes/agentTrigger.ts
+2 -2
View File
@@ -33,8 +33,8 @@ export const buildAgentTriggerPrompt = (trigger: AgentTriggerContext) => [
event_id: trigger.trigger_id,
})}</external_work_order>`,
"这是合成演练数据，不得表述为生产事实。",
"开始调查前调用 publish_execution_plan 发布本次调查的结构化计划，明确并行依赖和条件分支；执行前、取得结果后、遇到阻塞或修改计划时再次发布完整快照。保留节点 ID、已完成节点及证据，重试用新节点，不要按时间推算进度。",
"根据这张工单的事实动态规划本次调查，不套用预设路线。开始调查前调用 publish_execution_plan 发布 execution-plan@2，明确路线选择、并行分叉与汇合、人工审核点；执行前、取得结果后、遇到阻塞或修改计划时再次发布完整快照。保留节点 ID、已完成节点及证据，重试用新节点，不要按时间推算进度。",
"每个完成节点必须列明实际依据。预设证据标注为合成场景输入，不得声称执行过未调用的数据查询工具。现场采样、清洗、复测、回访等人工工作保持 waiting，写清等待内容，不得因回复结束标记完成。",
"外围样板随事件提供的合成场景输入（不是 Agent 实时查询结果）：上游三个模拟水质监测点的数据质量均为 valid，演示指标均在场景阈值内，且相邻片区没有同类投诉；投诉集中在演示小区，二次供水水箱最近一次清洗记录待核验。",
"请按‘上游数据质量正常、上游水质未见异常 → 疑似二次供水环节问题（待现场核验）’形成阶段性判断，并给出处置、复测和回访建议。不得将疑似判断写成已确认事故。",
"依据实际取得的证据判断可能原因，区分合成场景输入、已查询数据和待现场核验事项；给出处置、复测和回访建议。不得将疑似判断写成已确认事故。",
].join("\n");
src/routes/chatSessionRoutes.ts
+1 -1
View File
@@ -105,7 +105,7 @@ export const registerChatSessionRoutes = (
body: JSON.stringify({
session_id: record.sessionId,
model,
message: `执行工况任务${workflow ? `：${workflow}` : ""}。先读取后端任务上下文，再按任务类型更新计划、证据和报告。`,
message: `执行工况任务${workflow ? `：${workflow}` : ""}。先读取后端任务上下文，依据当前工单或人工请求动态规划 execution-plan@2；按实际选择的路线更新计划、证据和报告，数据权限不足时先申请并等待审核。`,
}),
}).then(async (response) => {
const reader = response.body?.getReader();
src/routes/executionPlan.ts
+52 -3
View File
@@ -2,6 +2,7 @@ import { z } from "zod";
const id = z.string().regex(/^[a-zA-Z0-9_-]{1,64}$/);
export const executionPlanSchema = z.object({
schema: z.enum(["execution-plan@1", "execution-plan@2"]).default("execution-plan@1"),
title: z.string().trim().min(1).max(120),
reason: z.string().trim().min(1).max(1000),
nodes: z.array(z.object({
@@ -12,18 +13,27 @@ export const executionPlanSchema = z.object({
detail: z.string().max(2000),
evidence: z.array(z.string().min(1).max(1000)).max(12),
action_refs: z.array(z.string().uuid()).max(12).default([]),
kind: z.enum(["activity", "decision", "parallel_split", "parallel_join", "gate"]).optional(),
selected_edge_id: id.nullable().optional(),
gate_ref: z.string().uuid().nullable().optional(),
revisits_node_id: id.nullable().optional(),
})).min(1).max(40),
edges: z.array(z.object({ from: id, to: id, label: z.string().max(120) })).max(80),
edges: z.array(z.object({ id: id.optional(), from: id, to: id, label: z.string().max(120) })).max(80),
}).superRefine((plan, context) => {
const ids = new Set(plan.nodes.map((node) => node.id));
if (ids.size !== plan.nodes.length) context.addIssue({ code: "custom", message: "Duplicate node ID" });
const edges = new Set<string>();
const edgeIds = new Set<string>();
for (const edge of plan.edges) {
const key = `${edge.from}:${edge.to}`;
if (!ids.has(edge.from) || !ids.has(edge.to) || edge.from === edge.to || edges.has(key)) {
context.addIssue({ code: "custom", message: "Invalid or duplicate dependency" });
}
edges.add(key);
if (plan.schema === "execution-plan@2") {
if (!edge.id || edgeIds.has(edge.id)) context.addIssue({ code: "custom", message: "V2 edges require unique IDs" });
if (edge.id) edgeIds.add(edge.id);
}
}
const visiting = new Set<string>();
const visited = new Set<string>();
@@ -37,6 +47,35 @@ export const executionPlanSchema = z.object({
return false;
};
if (plan.nodes.some((node) => visit(node.id))) context.addIssue({ code: "custom", message: "Dependencies must be acyclic; use a new node for a retry" });
if (plan.schema === "execution-plan@2") {
if (plan.nodes.filter((node) => !plan.edges.some((edge) => edge.to === node.id)).length !== 1) {
context.addIssue({ code: "custom", message: "V2 plan requires one start node" });
}
const nodeById = new Map(plan.nodes.map((node) => [node.id, node]));
for (const node of plan.nodes) {
const outgoing = plan.edges.filter((edge) => edge.from === node.id);
const incoming = plan.edges.filter((edge) => edge.to === node.id);
if (!node.kind) context.addIssue({ code: "custom", message: "V2 nodes require kind" });
if (node.kind === "decision") {
if (outgoing.length < 2 || outgoing.some((edge) => !edge.label.trim())) context.addIssue({ code: "custom", message: "Decision routes must be labeled" });
if (node.selected_edge_id && !outgoing.some((edge) => edge.id === node.selected_edge_id)) context.addIssue({ code: "custom", message: "Selected route must leave its decision" });
if (node.status === "completed" && !node.selected_edge_id) context.addIssue({ code: "custom", message: "Completed decision requires a selected route" });
if (node.selected_edge_id) {
for (const edge of outgoing) {
const targetStatus = nodeById.get(edge.to)?.status;
if (edge.id === node.selected_edge_id && ["skipped", "cancelled"].includes(targetStatus ?? "")) context.addIssue({ code: "custom", message: "Selected route cannot be skipped" });
if (edge.id !== node.selected_edge_id && ["running", "completed", "waiting"].includes(targetStatus ?? "")) context.addIssue({ code: "custom", message: "Unchosen route cannot be active" });
}
}
} else if (node.selected_edge_id) context.addIssue({ code: "custom", message: "Only decisions select routes" });
if (node.kind !== "decision" && node.kind !== "parallel_split" && outgoing.length > 1) context.addIssue({ code: "custom", message: "Only decisions or parallel splits may branch" });
if (node.kind === "parallel_split" && outgoing.length < 2) context.addIssue({ code: "custom", message: "Parallel split requires two routes" });
if (node.kind === "parallel_join" && incoming.length < 2) context.addIssue({ code: "custom", message: "Parallel join requires two inputs" });
if (node.kind === "gate" && !node.gate_ref) context.addIssue({ code: "custom", message: "Gate requires a persisted reference" });
if (node.kind !== "gate" && node.gate_ref) context.addIssue({ code: "custom", message: "Only gates reference approvals" });
if (node.revisits_node_id && !ids.has(node.revisits_node_id)) context.addIssue({ code: "custom", message: "Revisited node must exist" });
}
}
for (const node of plan.nodes) {
if (node.status === "running" || node.status === "completed") {
const hasIncompletePredecessor = plan.edges
@@ -50,11 +89,21 @@ export const executionPlanSchema = z.object({
message: "Running or completed nodes require completed or skipped predecessors",
});
}
if (node.status === "completed" && node.evidence.length === 0 && node.action_refs.length === 0) context.addIssue({ code: "custom", message: "Completed nodes require evidence or an approved action receipt" });
if (node.status === "completed" && node.evidence.length === 0 && node.action_refs.length === 0
&& !(plan.schema === "execution-plan@2" && ["parallel_split", "parallel_join", "gate"].includes(node.kind ?? ""))) {
context.addIssue({ code: "custom", message: "Completed nodes require evidence or an approved action receipt" });
}
if ((node.status === "waiting" || node.status === "failed") && !node.detail.trim()) context.addIssue({ code: "custom", message: "Waiting or failed nodes require a reason" });
}
});
// Stored v1 plans remain readable, but every newly published task plan must
// opt into the graph and approval-gate guarantees of v2.
export const newExecutionPlanSchema = executionPlanSchema.refine(
(plan) => plan.schema === "execution-plan@2",
{ message: "New task execution plans must use execution-plan@2", path: ["schema"] },
);
export const executionPlanNeedsTaskPublish = (input: unknown) =>
!(input && typeof input === "object" && (input as Record<string, unknown>).server_published === true);
@@ -84,5 +133,5 @@ export function createExecutionPlanSnapshot(input: unknown, messages: unknown[],
}
}
}
return { ...plan, schema_version: "execution-plan@1", session_id: sessionId, revision: history.length + 1, updated_at: new Date().toISOString() };
return { ...plan, schema_version: plan.schema, session_id: sessionId, revision: history.length + 1, updated_at: new Date().toISOString() };
}