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

feat(investigation): publish incremental reference graphs

Browse Source 
Publish graph deltas using the Server-owned cumulative graph contract. Validate historical comparison chart publications and consolidate Agent documentation.
This commit is contained in:
jiang
2026-11-11 19:10:41 +08:00
parent 5ee93329a4
commit db87323b69
28 changed files with 2329 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/product/investigationReferenceContract.test.ts
+87
View File
@@ -0,0 +1,87 @@
import { describe, expect, it } from "bun:test";
import express from "express";
import { tool } from "../../.opencode/tool.js";
import publishProgress from "../../.opencode/tools/publish_task_progress.js";
import type { OperationalTaskClient } from "../../src/product/operational-task-client.js";
import { investigationNodeAnnotations } from "../../src/product/investigation-node-state.js";
import { registerTaskPublicationBridge } from "../../src/routes/task-publication-bridge.js";
import { removeRuntimeSessionContext, setRuntimeSessionContext } from "../../src/runtime/sessionContext.js";
const native = tool.schema.object(publishProgress.args);
describe("investigation reference contract", () => {
it("advertises five optional annotations to the model without execution identities or Mermaid", () => {
const schema = tool.schema.toJSONSchema(native, { io: "input" });
const progress = schema.properties!.progress as {
properties: Record<string, {
properties: { nodes: { items: { properties: { status: { anyOf: Array<{ enum?: unknown[] }> } }; required?: string[] } } };
}>;
};
const node = progress.properties.graph_update.properties.nodes.items;
const statuses = node.properties.status.anyOf.flatMap((variant) => variant.enum ?? []);
expect(statuses).toEqual([...investigationNodeAnnotations]);
expect(node.required).not.toContain("status");
expect(Object.keys(schema.properties!)).toEqual(["request_id", "progress"]);
expect(Object.keys(progress.properties)).toEqual(["summary", "graph_update"]);
});
it("transports a native graph update and text-only update through the trusted task bridge", async () => {
const runtimeSessionId = "reference-contract-native-session";
const taskId = "50000000-0000-4000-8000-000000000031";
const token = "reference-contract-test-token";
const published: unknown[][] = [];
setRuntimeSessionContext({
sessionId: runtimeSessionId, clientSessionId: "reference-contract-client", actorKey: "test-actor",
projectKey: "test-project", traceId: "test", taskId, purpose: "task_execution",
});
const app = express();
app.use(express.json());
registerTaskPublicationBridge(app, { internalToken: token, client: {
publishArtifacts: async (...args: unknown[]) => {
published.push(args);
return { progress_revision: published.length };
},
} as unknown as Pick<OperationalTaskClient, "publishArtifacts" | "publishPresentation"> });
const server = app.listen(0, "127.0.0.1");
await new Promise<void>((resolve) => server.once("listening", resolve));
const baseUrl = `http://127.0.0.1:${(server.address() as { port: number }).port}`;
const oldToken = process.env.AGENT_INTERNAL_TOKEN;
const oldBaseUrl = process.env.TJWATER_AGENT_INTERNAL_BASE_URL;
process.env.AGENT_INTERNAL_TOKEN = token;
process.env.TJWATER_AGENT_INTERNAL_BASE_URL = baseUrl;
try {
const requestId = "60000000-0000-4000-8000-000000000031";
const progress = { summary: "整理已有路线与结果", graph_update: {
nodes: investigationNodeAnnotations.map((status, index) => ({
node_id: `step_${index}`, label: `参考步骤 ${index}`, status,
...(status === "failed" ? { detail: "此前查询超时，保留失败路径" } : {}),
})), edges: [{ source: "step_3", target: "step_0", label: "替代路线" }],
} };
const args = native.parse({ request_id: requestId, progress });
expect(JSON.parse(await publishProgress.execute(args, { sessionID: runtimeSessionId }))).toEqual({ progress_revision: 1 });
expect(published[0]?.slice(0, 2)).toEqual([runtimeSessionId, taskId]);
expect(published[0]?.[2]).toMatchObject({ requestId, sourceSessionId: "reference-contract-client",
progress: { ...progress, diagram: null },
});
await publishProgress.execute(native.parse({ request_id: "60000000-0000-4000-8000-000000000032", progress: { summary: "补充本轮说明" } }),
{ sessionID: runtimeSessionId });
expect(published[1]?.[2]).toMatchObject({ progress: { summary: "补充本轮说明", diagram: null } });
const invalid = await fetch(`${baseUrl}/internal/operational-task/artifacts`, {
method: "POST", headers: { "Content-Type": "application/json", "x-agent-internal-token": token },
body: JSON.stringify({ session_id: runtimeSessionId, request_id: requestId, progress: {
summary: "旧执行状态不进入新图", graph_update: { nodes: [{ node_id: "old", label: "核查", status: "running" }] },
} }),
});
expect(invalid.status).toBe(400);
expect(published).toHaveLength(2);
} finally {
if (oldToken === undefined) delete process.env.AGENT_INTERNAL_TOKEN;
else process.env.AGENT_INTERNAL_TOKEN = oldToken;
if (oldBaseUrl === undefined) delete process.env.TJWATER_AGENT_INTERNAL_BASE_URL;
else process.env.TJWATER_AGENT_INTERNAL_BASE_URL = oldBaseUrl;
removeRuntimeSessionContext(runtimeSessionId);
await new Promise<void>((resolve) => server.close(() => resolve()));
}
});
});
tests/product/investigationReport.test.ts
+8 -13
View File
@@ -1,6 +1,6 @@
import { describe, expect, it } from "bun:test";
import { taskReportSchema, taskProgressSchema } from "../../src/product/investigation-report.js";
import { taskPublicationDiagramSchema, taskPublicationProgressSchema, taskReportUiBlocksSchema } from "../../.opencode/tools/task-publication-schema.js";
import { taskReportUiBlocksSchema } from "../../.opencode/tools/task-publication-schema.js";
const report = { title: "调查报告", current_conclusion: "无法确定原因", conclusion_boundary: "仅有上游数据",
evidence_refs: ["query-1"], recommendations: [], pending_items: [], stage: "final", outcome: "insufficient_data",
@@ -12,11 +12,9 @@ describe("new investigation report writes", () => {
const diagram = { title: "状态图", mermaid: "flowchart TD\nQUERY:::running --> REVIEW:::pending",
node_refs: [{ node_id: "QUERY", label: "核查", status: "running" },
{ node_id: "REVIEW", label: "复核", status: "pending" }] };
expect(taskPublicationDiagramSchema.safeParse(diagram).success).toBe(true);
expect(taskReportSchema.safeParse({ ...report, diagram }).success).toBe(true);
for (const node_id of ["missing", "QUER"]) {
const invalid = { ...diagram, node_refs: [{ ...diagram.node_refs[0], node_id }] };
expect(taskPublicationDiagramSchema.safeParse(invalid).success).toBe(false);
expect(taskReportSchema.safeParse({ ...report, diagram: invalid }).success).toBe(false);
}
});
@@ -41,7 +39,6 @@ describe("new investigation report writes", () => {
}
const mismatched = { ...diagram, node_refs: [{ ...diagram.node_refs[0], node_id: "query" }] };
expect(() => taskReportSchema.parse({ ...report, diagram: mismatched })).toThrow("未在 Mermaid 源码中声明");
expect(taskPublicationDiagramSchema.safeParse(mismatched).success).toBe(false);
});
it("rejects downstream work or unsupported claims on insufficient-data exit", () => {
for (const changes of [{ recommendations: ["派采样工单"] }, { pending_items: ["等待人工"] },
@@ -50,13 +47,13 @@ describe("new investigation report writes", () => {
expect(() => taskReportSchema.parse({ ...report, ...changes })).toThrow();
}
});
it("shares explicit node activity and supply feature targets with progress", () => {
it("shares saved reference annotations and supply feature targets with progress", () => {
const diagram = { title: "当前调查", mermaid: "flowchart LR\nQUERY --> HUMAN",
node_refs: [{ node_id: "QUERY", label: "核对监测", status: "running", evidence_refs: ["query-1"],
map_targets: [{ source_id: "scada", feature_id: "MP01" }] },
{ node_id: "HUMAN", label: "现场核验", status: "waiting_human" }] };
const progress = taskProgressSchema.parse({ summary: "正在核对投诉时段的实际监测记录", diagram });
expect(progress.diagram?.node_refs[0]).toMatchObject({ status: "running", map_targets: [{ source_id: "scada", feature_id: "MP01" }] });
expect(progress.diagram?.node_refs[0]).toMatchObject({ status: "reference", map_targets: [{ source_id: "scada", feature_id: "MP01" }] });
expect(taskReportSchema.parse({ ...report, diagram }).diagram).toEqual(progress.diagram);
expect(taskProgressSchema.parse({ summary: "调整调查方向", diagram: null }).diagram).toBeNull();
for (const change of [{ status: "blocked" }, { map_targets: [{ source_id: "drainage", feature_id: "X" }] },
@@ -66,7 +63,7 @@ describe("new investigation report writes", () => {
} })).toThrow();
}
});
it("preserves waiting, unknown and evidence-confirmed states at every diagram boundary", () => {
it("normalizes legacy states into saved reference annotations at every read boundary", () => {
const diagram = { title: "等待系统结果", mermaid: "flowchart LR\nDONE --> WAIT --> UNKNOWN --> UNSET", node_refs: [
{ node_id: "DONE", label: "查询完成", status: "completed", evidence_refs: ["query-1"] },
{ node_id: "WAIT", label: "等待系统结果", status: "waiting" },
@@ -74,12 +71,10 @@ describe("new investigation report writes", () => {
{ node_id: "UNSET", label: "未确认" },
] };
const nodes = taskProgressSchema.parse({ summary: "等待系统结果", diagram }).diagram!.node_refs;
expect(nodes?.map((node) => node.status)).toEqual(["completed", "waiting", null, undefined]);
expect(nodes?.map((node) => node.status)).toEqual(["recorded", "needs_input", null, undefined]);
expect(taskReportSchema.parse({ ...report, diagram }).diagram?.node_refs).toEqual(nodes);
expect(taskPublicationDiagramSchema.parse(diagram).node_refs).toEqual(nodes);
expect(taskPublicationProgressSchema.parse({ summary: "等待系统结果", diagram }).diagram?.node_refs).toEqual(nodes);
});
it("preserves parallel branches with independent activity and a pending join", () => {
it("preserves parallel branches and recorded findings without live activity assertions", () => {
const mermaid = 'flowchart TD\nSTART[受理投诉] --> SPLIT{并行核查}\nSPLIT --> TICKETS[工单核查]\nSPLIT --> SCADA[监测核查]\nTICKETS --> JOIN[汇总证据]\nSCADA --> JOIN';
const diagram = { title: "并行调查", mermaid, node_refs: [
{ node_id: "TICKETS", label: "工单核查", status: "running" },
@@ -88,11 +83,11 @@ describe("new investigation report writes", () => {
] };
const progress = taskProgressSchema.parse({ summary: "两个独立核查正在执行", diagram });
expect(progress.diagram?.mermaid).toBe(mermaid);
expect(progress.diagram?.node_refs.map((node) => node.status)).toEqual(["running", "running", "pending"]);
expect(progress.diagram?.node_refs.map((node) => node.status)).toEqual(["reference", "reference", "reference"]);
const updated = taskProgressSchema.parse({ summary: "工单核查已完成，监测核查仍在执行", diagram: {
...diagram, node_refs: [{ ...diagram.node_refs[0], status: "completed", evidence_refs: ["query-1"] }, ...diagram.node_refs.slice(1)],
} });
expect(updated.diagram?.node_refs.map((node) => node.status)).toEqual(["completed", "running", "pending"]);
expect(updated.diagram?.node_refs.map((node) => node.status)).toEqual(["recorded", "reference", "reference"]);
});
});
tests/product/taskPresentation.test.ts
+38 -12
View File
@@ -2,20 +2,11 @@ import { describe, expect, it } from "bun:test";
import publishPresentation from "../../.opencode/tools/publish_task_presentation.js";
import publishProgress from "../../.opencode/tools/publish_task_progress.js";
import { taskPublicationCommandSchema, taskPublicationProgressSchema } from "../../.opencode/tools/task-publication-schema.js";
import { taskPublicationCommandSchema, taskPublicationGraphProgressSchema } from "../../.opencode/tools/task-publication-schema.js";
import { taskProgressSchema } from "../../src/product/investigation-report.js";
import { taskPresentationCommandSchema } from "../../src/product/task-presentation.js";
describe("task presentation input boundaries", () => {
it("aligns native and product progress schemas for legacy and current diagrams", () => {
const progress = { summary: "实际读取监测资料", diagram: { title: "当前路线", mermaid: "flowchart TD\nQUERY --> REVIEW", node_refs: [
{ node_id: "QUERY", label: "查询", status: "running", map_targets: [{ source_id: "pipes", feature_id: "P-1" }] },
{ node_id: "REVIEW", label: "复核", status: null },
] } };
expect(taskProgressSchema.parse(taskPublicationProgressSchema.parse(progress))).toEqual(taskProgressSchema.parse(progress));
expect(taskProgressSchema.parse(taskPublicationProgressSchema.parse({ summary: "当前无图" }))).toEqual({ summary: "当前无图", diagram: null });
});
it("accepts finite display commands and rejects unsupported data, layers and chart evidence", () => {
const chart = { title: "压力", chart_type: "line", x_data: ["10:00"], series: [{ name: "MP01", data: [0.4] }] };
const commands = [
@@ -38,6 +29,41 @@ describe("task presentation input boundaries", () => {
]) expect(() => taskPresentationCommandSchema.parse(command)).toThrow();
});
it("exposes incremental graph updates instead of generated Mermaid to the Agent", () => {
const update = { summary: "首次定位超时，调整窗口重试", graph_update: {
nodes: [{ node_id: "locate_1", label: "首次定位", status: "failed", detail: "计算超时" },
{ node_id: "locate_2", label: "调整窗口重试", status: "reference" }],
edges: [{ source: "locate_1", target: "locate_2", label: "重试" }],
} };
expect(taskProgressSchema.parse(taskPublicationGraphProgressSchema.parse(update)))
.toEqual(taskProgressSchema.parse(update));
expect(taskPublicationGraphProgressSchema.safeParse({ summary: "重新生成", diagram: {
title: "替换图", mermaid: "flowchart TD\nA[新图]",
} }).success).toBe(false);
expect(taskProgressSchema.safeParse({ ...update, diagram: {
title: "替换图", mermaid: "flowchart TD\nA[新图]",
} }).success).toBe(false);
expect(taskProgressSchema.parse(taskPublicationGraphProgressSchema.parse({ summary: "继续等待" })))
.toEqual({ summary: "继续等待", diagram: null });
});
it("accepts optional reference annotations and rejects live execution states in new graph updates", () => {
for (const status of [undefined, "reference", "recorded", "needs_input", "failed", "not_taken"]) {
const progress = { summary: "整理流程参考", graph_update: {
nodes: [{ node_id: "check", label: "核查路线", ...(status ? { status } : {}) }], edges: [],
} };
expect(taskPublicationGraphProgressSchema.safeParse(progress).success).toBe(true);
expect(taskProgressSchema.safeParse(progress).success).toBe(true);
}
for (const status of ["running", "completed", "pending", "waiting", "waiting_human", "skipped", "cancelled"]) {
const progress = { summary: "整理流程参考", graph_update: {
nodes: [{ node_id: "check", label: "核查路线", status }], edges: [],
} };
expect(taskPublicationGraphProgressSchema.safeParse(progress).success).toBe(false);
expect(taskProgressSchema.safeParse(progress).success).toBe(false);
}
});
it("revalidates semantic chart and Mermaid inputs before sending a tool request", async () => {
const request_id = "60000000-0000-4000-8000-000000000001";
await expect(publishPresentation.execute({ request_id, command: {
@@ -46,11 +72,11 @@ describe("task presentation input boundaries", () => {
} }, { sessionID: "test" })).rejects.toThrow("series must align");
await expect(publishProgress.execute({ request_id, progress: {
summary: "查询", diagram: { title: "路线", mermaid: "flowchart TD; A-->B; click A callback", node_refs: [] },
} }, { sessionID: "test" })).rejects.toThrow("Mermaid");
} } as never, { sessionID: "test" })).rejects.toThrow();
await expect(publishProgress.execute({ request_id, progress: {
summary: "查询", diagram: { title: "路线", mermaid: "flowchart TD\nA-->B",
node_refs: [{ node_id: "query", label: "查询", status: "running",
evidence_refs: [], action_refs: [], map_targets: [] }] },
} }, { sessionID: "test" })).rejects.toThrow("未在 Mermaid 源码中声明");
} } as never, { sessionID: "test" })).rejects.toThrow();
});
});