508 lines
20 KiB
TypeScript
508 lines
20 KiB
TypeScript
import { bindSessionCredential, releaseSessionCredential, requestWithSessionCredential } from "./credential-refresh.js";
|
|
import { taskReportSchema, taskProgressSchema, type TaskProgress } from "./investigation-report.js";
|
|
import { taskPresentationCommandSchema, taskPresentationResultSchema, type TaskPresentationCommand } from "./task-presentation.js";
|
|
export { taskReportSchema } from "./investigation-report.js";
|
|
import { z } from "zod";
|
|
import { interactionQuestionsSchema, type InteractionQuestion } from "../chat/interaction-request-contract.js";
|
|
|
|
import { config } from "../config.js";
|
|
import {
|
|
proposedTaskActionSchema,
|
|
type ProposedTaskAction,
|
|
} from "./operational-task-contract.js";
|
|
import type { ProductCredential } from "./server-auth-client.js";
|
|
|
|
export { proposedTaskActionSchema } from "./operational-task-contract.js";
|
|
|
|
const jsonRecordSchema = z.record(z.unknown());
|
|
const taskDataSourceSchema = z.enum([
|
|
"workbench_runtime",
|
|
"scada_devices",
|
|
"scada_readings",
|
|
"peripheral_points",
|
|
"peripheral_work_orders",
|
|
"peripheral_observations",
|
|
"peripheral_test_data",
|
|
]);
|
|
|
|
const operationalTaskRunSchema = z.object({
|
|
run_id: z.string().uuid(),
|
|
task_id: z.string().uuid(),
|
|
generation: z.number().int().positive(),
|
|
execution_request_id: z.string().uuid(),
|
|
status: z.enum(["queued", "claimed", "running", "succeeded", "failed", "timed_out", "cancelled"]),
|
|
revision: z.number().int().positive(),
|
|
agent_session_id: z.string().min(1).nullable(),
|
|
agent_instance_id: z.string().min(1).nullable().default(null),
|
|
lease_expires_at: z.string().datetime({ offset: true }).nullable(),
|
|
heartbeat_at: z.string().datetime({ offset: true }).nullable(),
|
|
started_at: z.string().datetime({ offset: true }).nullable(),
|
|
finished_at: z.string().datetime({ offset: true }).nullable(),
|
|
failure_code: z.string().min(1).nullable(),
|
|
failure_detail: z.string().min(1).nullable(),
|
|
dispatch_attempts: z.number().int().nonnegative(),
|
|
next_dispatch_at: z.string().datetime({ offset: true }),
|
|
created_at: z.string().datetime({ offset: true }),
|
|
updated_at: z.string().datetime({ offset: true }),
|
|
}).strict();
|
|
|
|
const operationalTaskSummarySchema = z.object({
|
|
task_id: z.string().uuid(),
|
|
task_type: z.enum(["scheduled_flow", "external_work_order", "event_investigation"]),
|
|
workflow: z.string().min(1),
|
|
lifecycle_status: z.enum(["open", "waiting_human", "completed", "failed", "cancelled"]),
|
|
active_run: operationalTaskRunSchema.nullable(),
|
|
status: z.enum(["queued", "running", "waiting_human", "completed", "failed", "cancelled"]),
|
|
revision: z.number().int().positive(),
|
|
execution_request_id: z.string().uuid(),
|
|
title: z.string().min(1),
|
|
summary: z.string().min(1),
|
|
}).passthrough();
|
|
|
|
const agentTaskContextSchema = z.object({
|
|
task: operationalTaskSummarySchema.extend({
|
|
latest_plan: z.unknown().nullable(),
|
|
latest_report: z.unknown().nullable(),
|
|
evidence: z.array(z.unknown()),
|
|
actions: z.array(z.unknown()),
|
|
permissions: z.record(z.boolean()),
|
|
}).passthrough(),
|
|
approved_data_scopes: z.array(taskDataSourceSchema),
|
|
selected_report: z.object({
|
|
revision: z.number().int().positive(), execution_generation: z.number().int().positive().nullish(),
|
|
report: z.unknown(),
|
|
}).passthrough().nullish(),
|
|
});
|
|
|
|
const taskArtifactsResultSchema = z.object({
|
|
task: operationalTaskSummarySchema,
|
|
plan_revision: z.number().int().positive().nullable(),
|
|
report_revision: z.number().int().positive().nullable(),
|
|
progress_revision: z.number().int().positive().nullable().optional(),
|
|
});
|
|
|
|
export const taskEvidenceSchema = z.object({
|
|
evidence_id: z.string().trim().min(1).max(128),
|
|
kind: z.literal("synthetic"),
|
|
summary: z.string().trim().min(1).max(4000),
|
|
source_label: z.string().trim().min(1).max(500),
|
|
source_ref: z.string().trim().max(500).optional(),
|
|
observed_at: z.string().datetime({ offset: true }),
|
|
});
|
|
|
|
export const humanTaskEvidenceCommandSchema = z.object({
|
|
request_id: z.string().uuid(),
|
|
evidence: z.object({
|
|
evidence_id: z.string().trim().min(1).max(128),
|
|
kind: z.literal("human_input"),
|
|
summary: z.string().trim().min(1).max(4000),
|
|
source_label: z.string().trim().min(1).max(500),
|
|
source_ref: z.string().trim().min(1).max(500).optional(),
|
|
observed_at: z.string().datetime({ offset: true }),
|
|
}),
|
|
});
|
|
|
|
export type TaskEvidence = z.infer<typeof taskEvidenceSchema>;
|
|
export type TaskReport = z.infer<typeof taskReportSchema>;
|
|
export const expressAccessRequestSchema = z.object({
|
|
requestId: z.string().uuid(),
|
|
nodeId: z.string().min(1),
|
|
scopes: z.array(taskDataSourceSchema).min(1).max(7),
|
|
reason: z.string().min(1),
|
|
});
|
|
export type HumanTaskEvidenceCommand = z.infer<typeof humanTaskEvidenceCommandSchema>;
|
|
|
|
export type OperationalTaskContext = Record<string, unknown>;
|
|
|
|
type TaskArtifactInput = {
|
|
requestId: string;
|
|
sourceSessionId: string;
|
|
evidence?: TaskEvidence[];
|
|
executionPlan?: Record<string, unknown>;
|
|
report?: TaskReport;
|
|
progress?: TaskProgress;
|
|
};
|
|
|
|
export const taskArtifactBridgeSchema = z.object({
|
|
request_id: z.string().uuid(),
|
|
evidence: z.array(taskEvidenceSchema).max(40).optional(),
|
|
execution_plan: jsonRecordSchema.optional(),
|
|
report: taskReportSchema.optional(),
|
|
progress: taskProgressSchema.optional(),
|
|
}).refine(
|
|
(value) => value.evidence !== undefined || value.execution_plan !== undefined || value.report !== undefined || value.progress !== undefined,
|
|
{ message: "at least one task artifact is required" },
|
|
);
|
|
|
|
const taskRunStates = new Map<string, { runId: string; revision: number }>();
|
|
const taskRunMutationQueues = new Map<string, Promise<void>>();
|
|
export const hasOperationalTaskRun = (runtimeSessionId: string) => taskRunStates.has(runtimeSessionId);
|
|
|
|
export const bindOperationalTaskCredential = (
|
|
runtimeSessionId: string,
|
|
credential: ProductCredential | null,
|
|
) => {
|
|
bindSessionCredential(runtimeSessionId, credential);
|
|
};
|
|
|
|
export const releaseOperationalTaskCredential = (runtimeSessionId: string) => {
|
|
releaseSessionCredential(runtimeSessionId);
|
|
taskRunStates.delete(runtimeSessionId);
|
|
taskRunMutationQueues.delete(runtimeSessionId);
|
|
};
|
|
|
|
export const clearOperationalTaskLease = (runtimeSessionId: string) => {
|
|
taskRunStates.delete(runtimeSessionId);
|
|
};
|
|
|
|
export class OperationalTaskClient {
|
|
constructor(private readonly baseUrl = config.TJWATER_SERVER_BASE_URL) {}
|
|
|
|
async getAgentContext(runtimeSessionId: string, taskId: string, reportRevision?: number, signal?: AbortSignal) {
|
|
const query = reportRevision == null ? "" : `?report_revision=${reportRevision}`;
|
|
const raw = await this.request(runtimeSessionId, `/api/v1/operational-tasks/${encodeURIComponent(taskId)}/agent-context${query}`, {
|
|
method: "GET",
|
|
signal,
|
|
});
|
|
const result = agentTaskContextSchema.parse(raw);
|
|
rememberLeaseRevision(runtimeSessionId, result);
|
|
return result;
|
|
}
|
|
|
|
async claimTask(
|
|
runtimeSessionId: string,
|
|
taskId: string,
|
|
runId: string,
|
|
input: { requestId: string; expectedRevision: number; sourceSessionId: string },
|
|
) {
|
|
return this.mutateRun(runtimeSessionId, async () => {
|
|
const raw = await this.request(runtimeSessionId, `/api/v1/operational-tasks/${encodeURIComponent(taskId)}/runs/${encodeURIComponent(runId)}/claim`, {
|
|
method: "POST",
|
|
body: JSON.stringify({
|
|
request_id: input.requestId,
|
|
expected_revision: input.expectedRevision,
|
|
source_session_id: input.sourceSessionId,
|
|
agent_instance_id: config.AGENT_INSTANCE_ID,
|
|
}),
|
|
});
|
|
const result = operationalTaskRunSchema.parse(raw);
|
|
rememberLeaseRevision(runtimeSessionId, result);
|
|
return result;
|
|
});
|
|
}
|
|
|
|
async heartbeatTask(
|
|
runtimeSessionId: string,
|
|
taskId: string,
|
|
runId: string,
|
|
sourceSessionId: string,
|
|
) {
|
|
return this.mutateRun(runtimeSessionId, async () => {
|
|
const state = taskRunStates.get(runtimeSessionId);
|
|
if (!state || state.runId !== runId) throw new OperationalTaskClientError(409, "task run lease revision unavailable");
|
|
const raw = await this.request(runtimeSessionId, `/api/v1/operational-tasks/${encodeURIComponent(taskId)}/runs/${encodeURIComponent(runId)}/heartbeat`, {
|
|
method: "POST",
|
|
body: JSON.stringify({
|
|
request_id: crypto.randomUUID(),
|
|
expected_revision: state.revision,
|
|
source_session_id: sourceSessionId,
|
|
agent_instance_id: config.AGENT_INSTANCE_ID,
|
|
}),
|
|
});
|
|
const result = operationalTaskRunSchema.parse(raw);
|
|
rememberLeaseRevision(runtimeSessionId, result);
|
|
return result;
|
|
});
|
|
}
|
|
|
|
async publishArtifacts(runtimeSessionId: string, taskId: string, input: TaskArtifactInput) {
|
|
return this.mutateRun(runtimeSessionId, async () => {
|
|
const state = taskRunStates.get(runtimeSessionId);
|
|
if (!state) throw new OperationalTaskClientError(409, "task run context unavailable");
|
|
const raw = await this.request(runtimeSessionId, `/api/v1/operational-tasks/${encodeURIComponent(taskId)}/agent-artifacts`, {
|
|
method: "POST",
|
|
body: JSON.stringify({
|
|
request_id: input.requestId,
|
|
source_session_id: input.sourceSessionId,
|
|
run_id: state.runId,
|
|
...(input.evidence ? { evidence: input.evidence } : {}),
|
|
...(input.executionPlan ? { plan: toServerPlan(input.executionPlan) } : {}),
|
|
...(input.report ? { report: input.report } : {}),
|
|
...(input.progress ? { progress: taskProgressSchema.parse(input.progress) } : {}),
|
|
}),
|
|
});
|
|
const result = taskArtifactsResultSchema.parse(raw);
|
|
rememberLeaseRevision(runtimeSessionId, result);
|
|
return result;
|
|
});
|
|
}
|
|
|
|
async completeTaskRun(runtimeSessionId: string, taskId: string, runId: string, sourceSessionId: string) {
|
|
return this.finishTaskRun(runtimeSessionId, taskId, runId, "complete", sourceSessionId);
|
|
}
|
|
|
|
async publishPresentation(runtimeSessionId: string, taskId: string, input: {
|
|
requestId: string; sourceSessionId: string; command: TaskPresentationCommand;
|
|
}) {
|
|
return this.mutateRun(runtimeSessionId, async () => {
|
|
const state = taskRunStates.get(runtimeSessionId);
|
|
if (!state) throw new OperationalTaskClientError(409, "task run context unavailable");
|
|
const raw = await this.request(runtimeSessionId, `/api/v1/operational-tasks/${encodeURIComponent(taskId)}/presentation-events`, {
|
|
method: "POST",
|
|
body: JSON.stringify({
|
|
request_id: input.requestId,
|
|
run_id: state.runId,
|
|
source_session_id: input.sourceSessionId,
|
|
command: taskPresentationCommandSchema.parse(input.command),
|
|
}),
|
|
});
|
|
return taskPresentationResultSchema.parse(raw);
|
|
});
|
|
}
|
|
|
|
async failTaskRun(
|
|
runtimeSessionId: string,
|
|
taskId: string,
|
|
runId: string,
|
|
sourceSessionId: string,
|
|
failureCode: string,
|
|
failureDetail: string,
|
|
) {
|
|
return this.finishTaskRun(runtimeSessionId, taskId, runId, "fail", sourceSessionId, {
|
|
failure_code: failureCode,
|
|
failure_detail: failureDetail,
|
|
});
|
|
}
|
|
|
|
private async finishTaskRun(
|
|
runtimeSessionId: string,
|
|
taskId: string,
|
|
runId: string,
|
|
action: "complete" | "fail",
|
|
sourceSessionId: string,
|
|
extra: Record<string, string> = {},
|
|
) {
|
|
return this.mutateRun(runtimeSessionId, async () => {
|
|
const state = taskRunStates.get(runtimeSessionId);
|
|
if (!state || state.runId !== runId) throw new OperationalTaskClientError(409, "task run lease revision unavailable");
|
|
const raw = await this.request(
|
|
runtimeSessionId,
|
|
`/api/v1/operational-tasks/${encodeURIComponent(taskId)}/runs/${encodeURIComponent(runId)}/${action}`,
|
|
{
|
|
method: "POST",
|
|
body: JSON.stringify({
|
|
request_id: crypto.randomUUID(),
|
|
expected_revision: state.revision,
|
|
source_session_id: sourceSessionId,
|
|
agent_instance_id: config.AGENT_INSTANCE_ID,
|
|
...extra,
|
|
}),
|
|
},
|
|
);
|
|
const result = operationalTaskSummarySchema.parse(raw);
|
|
rememberLeaseRevision(runtimeSessionId, result);
|
|
return result;
|
|
});
|
|
}
|
|
|
|
private async mutateRun<T>(runtimeSessionId: string, mutation: () => Promise<T>): Promise<T> {
|
|
const previous = taskRunMutationQueues.get(runtimeSessionId) ?? Promise.resolve();
|
|
const result = previous.catch(() => undefined).then(mutation);
|
|
const tail = result.then(() => undefined, () => undefined);
|
|
taskRunMutationQueues.set(runtimeSessionId, tail);
|
|
try {
|
|
return await result;
|
|
} finally {
|
|
if (taskRunMutationQueues.get(runtimeSessionId) === tail) {
|
|
taskRunMutationQueues.delete(runtimeSessionId);
|
|
}
|
|
}
|
|
}
|
|
|
|
async requestDataAccess(runtimeSessionId: string, taskId: string, input: {
|
|
requestId: string; sourceSessionId: string; nodeId: string; scopes: z.infer<typeof taskDataSourceSchema>[]; reason: string;
|
|
}) {
|
|
return this.request(runtimeSessionId, `/api/v1/operational-tasks/${encodeURIComponent(taskId)}/access-requests`, {
|
|
method: "POST",
|
|
body: JSON.stringify({
|
|
request_id: input.requestId, source_session_id: input.sourceSessionId,
|
|
node_id: input.nodeId, scopes: input.scopes, reason: input.reason,
|
|
}),
|
|
});
|
|
}
|
|
|
|
async askTaskUser(runtimeSessionId: string, taskId: string, input: {
|
|
request_id: string; questions: InteractionQuestion[]; sourceSessionId: string;
|
|
}) {
|
|
return this.mutateRun(runtimeSessionId, async () => {
|
|
const state = taskRunStates.get(runtimeSessionId);
|
|
if (!state) throw new OperationalTaskClientError(409, "task run context unavailable");
|
|
const raw = await this.request(runtimeSessionId, `/api/v1/operational-tasks/${encodeURIComponent(taskId)}/questions`, {
|
|
method: "POST", body: JSON.stringify({ request_id: input.request_id,
|
|
source_session_id: input.sourceSessionId, run_id: state.runId,
|
|
questions: interactionQuestionsSchema.parse(input.questions) }),
|
|
});
|
|
return z.object({ question_request_id: z.string().uuid(), status: z.enum(["pending", "answered", "cancelled"]),
|
|
version: z.number().int().positive() }).passthrough().parse(raw);
|
|
});
|
|
}
|
|
|
|
async proposeAction(runtimeSessionId: string, taskId: string, input: ProposedTaskAction) {
|
|
const result = await this.request(runtimeSessionId, `/api/v1/operational-tasks/${encodeURIComponent(taskId)}/actions`, {
|
|
method: "POST",
|
|
body: JSON.stringify(proposedTaskActionSchema.parse(input)),
|
|
});
|
|
return z.object({
|
|
action_id: z.string().uuid(),
|
|
approval_status: z.literal("pending"),
|
|
execution_status: z.enum(["not_submitted", "submitted", "succeeded", "failed", "unsupported", "unknown"]),
|
|
}).passthrough().parse(result);
|
|
}
|
|
|
|
async addHumanEvidence(runtimeSessionId: string, taskId: string, input: HumanTaskEvidenceCommand) {
|
|
const result = await this.request(runtimeSessionId, `/api/v1/operational-tasks/${encodeURIComponent(taskId)}/human-evidence`, {
|
|
method: "POST",
|
|
body: JSON.stringify(humanTaskEvidenceCommandSchema.parse(input)),
|
|
});
|
|
return operationalTaskSummarySchema.parse(result);
|
|
}
|
|
|
|
private async request(runtimeSessionId: string, path: string, init: RequestInit) {
|
|
const response = await requestWithSessionCredential(runtimeSessionId, async (credential) => {
|
|
try {
|
|
return await fetch(new URL(path, this.baseUrl), {
|
|
...init,
|
|
signal: init.signal ?? AbortSignal.timeout(config.TJWATER_SERVER_TIMEOUT_MS),
|
|
headers: {
|
|
Accept: "application/json",
|
|
Authorization: credential.authorization,
|
|
"Content-Type": "application/json",
|
|
"X-Project-Id": credential.projectId,
|
|
},
|
|
});
|
|
} catch {
|
|
throw new OperationalTaskClientError(503, "TJWater Server task API unavailable");
|
|
}
|
|
});
|
|
const text = await response.text();
|
|
const value = parseJson(text);
|
|
if (!response.ok) {
|
|
const problem = readProblem(value);
|
|
throw new OperationalTaskClientError(response.status, problem.message, problem.code);
|
|
}
|
|
return jsonRecordSchema.parse(value);
|
|
}
|
|
}
|
|
|
|
export class OperationalTaskClientError extends Error {
|
|
constructor(
|
|
readonly status: number,
|
|
message: string,
|
|
readonly code?: string,
|
|
) {
|
|
super(message);
|
|
}
|
|
}
|
|
|
|
/** Feed a concrete Server coverage rejection back into the same active investigation. */
|
|
export function scadaPresentationContinuation(error: unknown): string | null {
|
|
if (!(error instanceof OperationalTaskClientError) || error.code !== "SCADA_PRESENTATION_INCOMPLETE") return null;
|
|
return `Server尚未接受本轮调查结束：${error.message}\n请按返回设备补充publish_task_presentation监测图，必要时更新publish_task_report。保留目标设备与真实缺测说明，完成后再结束本轮。`;
|
|
}
|
|
|
|
const parseJson = (text: string): unknown => {
|
|
if (!text.trim()) return {};
|
|
try {
|
|
return JSON.parse(text);
|
|
} catch {
|
|
throw new OperationalTaskClientError(502, "TJWater Server task API returned invalid JSON");
|
|
}
|
|
};
|
|
|
|
const readProblem = (value: unknown): { code?: string; message: string } => {
|
|
if (value && typeof value === "object") {
|
|
const record = value as Record<string, unknown>;
|
|
const detail = record.detail;
|
|
const nested = detail && typeof detail === "object"
|
|
? detail as Record<string, unknown>
|
|
: null;
|
|
const code = typeof nested?.code === "string"
|
|
? nested.code
|
|
: typeof record.code === "string"
|
|
? record.code
|
|
: undefined;
|
|
const message = typeof nested?.message === "string"
|
|
? nested.message
|
|
: typeof detail === "string"
|
|
? detail
|
|
: typeof record.message === "string"
|
|
? record.message
|
|
: "TJWater Server rejected task request";
|
|
return { code, message };
|
|
}
|
|
return { message: "TJWater Server rejected task request" };
|
|
};
|
|
|
|
const rememberLeaseRevision = (runtimeSessionId: string, value: Record<string, unknown>) => {
|
|
const directRun = operationalTaskRunSchema.safeParse(value);
|
|
if (directRun.success) {
|
|
taskRunStates.set(runtimeSessionId, {
|
|
runId: directRun.data.run_id,
|
|
revision: directRun.data.revision,
|
|
});
|
|
return;
|
|
}
|
|
const task = value.task && typeof value.task === "object"
|
|
? value.task as Record<string, unknown>
|
|
: value;
|
|
const activeRun = task.active_run;
|
|
const parsed = operationalTaskRunSchema.safeParse(activeRun);
|
|
if (parsed.success) {
|
|
taskRunStates.set(runtimeSessionId, {
|
|
runId: parsed.data.run_id,
|
|
revision: parsed.data.revision,
|
|
});
|
|
} else if (Object.hasOwn(task, "active_run") && activeRun === null) {
|
|
taskRunStates.delete(runtimeSessionId);
|
|
}
|
|
};
|
|
|
|
export const operationalTaskClient = new OperationalTaskClient();
|
|
|
|
const toServerPlan = (input: Record<string, unknown>) => {
|
|
const nodes = Array.isArray(input.nodes) ? input.nodes : [];
|
|
const edges = Array.isArray(input.edges) ? input.edges : [];
|
|
return {
|
|
schema: input.schema === "execution-plan@2" ? "execution-plan@2" : "execution-plan@1",
|
|
title: input.title,
|
|
reason: input.reason,
|
|
nodes: nodes.map((value) => {
|
|
const node = value as Record<string, unknown>;
|
|
return {
|
|
id: node.id,
|
|
title: node.title,
|
|
owner: node.owner,
|
|
status: node.status,
|
|
detail: node.detail,
|
|
evidence_refs: Array.isArray(node.evidence) ? node.evidence : [],
|
|
action_refs: Array.isArray(node.action_refs) ? node.action_refs : [],
|
|
...(input.schema === "execution-plan@2" ? {
|
|
kind: node.kind,
|
|
selected_edge_id: node.selected_edge_id ?? null,
|
|
gate_ref: node.gate_ref ?? null,
|
|
revisits_node_id: node.revisits_node_id ?? null,
|
|
} : {}),
|
|
};
|
|
}),
|
|
edges: edges.map((value) => {
|
|
const edge = value as Record<string, unknown>;
|
|
return {
|
|
from: edge.from,
|
|
to: edge.to,
|
|
...(typeof edge.label === "string" && edge.label.trim() ? { condition: edge.label } : {}),
|
|
...(input.schema === "execution-plan@2" ? { id: edge.id } : {}),
|
|
};
|
|
}),
|
|
};
|
|
};
|