feat(agent): support Responses steering and project-aware tools
Use the dedicated TJWATER_AGENT_URL and Responses stream protocol. Keep running turns active while users steer or queue follow-ups and edit pending delivery. Refresh the model catalog before generation and preserve approval mode selection. Integrate scoped investigation messaging/presentation, geographic place selection, current Agent composer styles and supporting workbench contracts. Validation: 77 focused Vitest checks, 21 browser cases, lint, typecheck and production build. Preserve direct Server/Agent routing and project authentication.
This commit is contained in:
@@ -2,6 +2,19 @@ import { describe, expect, it } from "vitest";
|
||||
import { applySessionStreamEvent, toAgentChatMessages, toAgentUiMessages } from "./agent-session-message-state";
|
||||
|
||||
describe("task context in persisted answers", () => {
|
||||
it("inserts consumed user input once before the active assistant in live and resumed streams", () => {
|
||||
const messages = toAgentUiMessages([
|
||||
{ id: "original", role: "user", parts: [{ type: "text", text: "原问题" }] },
|
||||
{ id: "answer", role: "assistant", parts: [{ type: "text", text: "处理中" }] }
|
||||
]);
|
||||
const data = { id: "new-input", text: "追加线索", delivery: "steer", status: "pending" };
|
||||
expect(applySessionStreamEvent(messages, { type: "user_input", data })).toBe(messages);
|
||||
const delivered = { type: "user_input" as const, data: { ...data, status: "delivered" } };
|
||||
const next = applySessionStreamEvent(messages, delivered);
|
||||
expect(next.map(({ id }) => id)).toEqual(["original", "new-input", "answer"]);
|
||||
expect(applySessionStreamEvent(next, delivered)).toBe(next);
|
||||
});
|
||||
|
||||
it("keeps the report reference of each answer when history is restored", () => {
|
||||
const reference = {
|
||||
task_id: "22222222-2222-4222-8222-222222222222",
|
||||
|
||||
@@ -19,10 +19,14 @@ import {
|
||||
} from "@/features/agent";
|
||||
|
||||
type AgentUiDataParts = {
|
||||
agent_message: Record<string, unknown>;
|
||||
task_context: Record<string, unknown>;
|
||||
execution_plan: Record<string, unknown>;
|
||||
progress: Record<string, unknown>;
|
||||
final_answer: Record<string, unknown>;
|
||||
credential_refresh_required: Record<string, unknown>;
|
||||
credential_refreshed: Record<string, unknown>;
|
||||
credential_refresh_failed: Record<string, unknown>;
|
||||
todo_update: Record<string, unknown>;
|
||||
permission_request: Record<string, unknown>;
|
||||
permission_response: Record<string, unknown>;
|
||||
@@ -35,6 +39,7 @@ type AgentUiDataParts = {
|
||||
answer_ui: Record<string, unknown>;
|
||||
ui_envelope: Record<string, unknown>;
|
||||
session_title: Record<string, unknown>;
|
||||
user_input: Record<string, unknown>;
|
||||
};
|
||||
|
||||
export type AgentUiMessage = UIMessage<unknown, AgentUiDataParts>;
|
||||
@@ -200,6 +205,15 @@ export function applySessionStreamEvent(messages: AgentUiMessage[], event: Agent
|
||||
return toAgentUiMessages(event.messages);
|
||||
}
|
||||
|
||||
if (event.type === "user_input") {
|
||||
const id = getDataString(event.data, "id");
|
||||
const text = getDataString(event.data, "text");
|
||||
if (event.data.status !== "delivered" || !id || !text || messages.some((message) => message.id === id)) return messages;
|
||||
const input: AgentUiMessage = { id, role: "user", parts: [{ type: "text", text }] };
|
||||
const tail = messages.at(-1);
|
||||
return tail?.role === "assistant" ? [...messages.slice(0, -1), input, tail] : [...messages, input];
|
||||
}
|
||||
|
||||
if (event.type === "token") {
|
||||
const token = getDataString(event.data, "content");
|
||||
return token ? updateLastAssistantUiMessage(messages, (message) => appendTextToUiMessage(message, token)) : messages;
|
||||
|
||||
@@ -0,0 +1,142 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import type { InvestigationPresentationCommand, InvestigationPresentationCursor } from "../api/investigation-presentation";
|
||||
import type { WorkbenchEvent } from "../api/workbench-sync-client";
|
||||
import { InvestigationPresentationFollower } from "./investigation-presentation-follower";
|
||||
import { WorkbenchEventChannel } from "./workbench-event-channel";
|
||||
|
||||
const taskId = "50000000-0000-4000-8000-000000000001";
|
||||
const runId = "55000000-0000-4000-8000-000000000001";
|
||||
const nextRunId = "55000000-0000-4000-8000-000000000002";
|
||||
const highlight: InvestigationPresentationCommand = { action: "highlight_features", layer: "scada", feature_ids: ["MP01"], effect: "pulse" };
|
||||
const openChart: InvestigationPresentationCommand = {
|
||||
action: "open_chart", chart_id: "pressure", evidence_refs: ["reading-1"],
|
||||
chart: { title: "压力", chart_type: "line", x_data: ["10:00"], series: [{ name: "MP01", data: [0.4] }] }
|
||||
};
|
||||
const closeChart: InvestigationPresentationCommand = { action: "close_chart", chart_id: "pressure" };
|
||||
|
||||
function cursor(eventCursor: number, generation = 1): InvestigationPresentationCursor {
|
||||
return { task_id: taskId, execution_generation: generation, event_cursor: eventCursor,
|
||||
active_run: { run_id: generation === 1 ? runId : nextRunId, generation, status: "running" } };
|
||||
}
|
||||
|
||||
function event(sequence: number, command = highlight, generation = 1): WorkbenchEvent {
|
||||
return { schema: "workbench-event@1", event_seq: sequence, event_id: "60000000-0000-4000-8000-000000000001",
|
||||
project_id: "10000000-0000-4000-8000-000000000001", topic: "operational.presentation.requested",
|
||||
aggregate_type: "operational_presentation", aggregate_id: taskId, aggregate_revision: 1,
|
||||
occurred_at: "2026-10-10T01:00:00Z", metadata: {
|
||||
task_id: taskId, run_id: generation === 1 ? runId : nextRunId, execution_generation: generation, command
|
||||
} };
|
||||
}
|
||||
|
||||
function deferred<T>() {
|
||||
let resolve!: (value: T) => void;
|
||||
const promise = new Promise<T>((done) => { resolve = done; });
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
function setup(overrides: Partial<ConstructorParameters<typeof InvestigationPresentationFollower>[0]> = {}) {
|
||||
const channel = new WorkbenchEventChannel();
|
||||
const execute = vi.fn(async (_command: InvestigationPresentationCommand, _signal: AbortSignal) => {});
|
||||
const fetchCursor = vi.fn(async (_signal: AbortSignal) => cursor(10));
|
||||
const fetchIdentity = vi.fn(async (_signal: AbortSignal) => cursor(10));
|
||||
const clear = vi.fn();
|
||||
const options = { taskId, subscribe: channel.subscribe, execute, fetchCursor, fetchIdentity,
|
||||
clear, stopPulse: vi.fn(), onStatus: vi.fn() };
|
||||
const follower = new InvestigationPresentationFollower({ ...options, ...overrides });
|
||||
return { channel, follower, ...options };
|
||||
}
|
||||
|
||||
describe("live investigation presentation following", () => {
|
||||
it("listens before fetching the baseline and preserves events arriving during cursor lookup", async () => {
|
||||
const baseline = deferred<InvestigationPresentationCursor>();
|
||||
const order: string[] = [];
|
||||
const test = setup();
|
||||
test.channel.publish({ type: "open" });
|
||||
const fetchCursor = vi.fn(() => {
|
||||
order.push("cursor");
|
||||
test.channel.publish({ type: "events", events: [event(8), event(11)] });
|
||||
return baseline.promise;
|
||||
});
|
||||
const follower = new InvestigationPresentationFollower({ ...test,
|
||||
subscribe: (listener) => { order.push("listen"); return test.channel.subscribe(listener); }, fetchCursor });
|
||||
follower.start();
|
||||
try {
|
||||
expect(order).toEqual(["listen", "cursor"]);
|
||||
expect(test.execute).not.toHaveBeenCalled();
|
||||
baseline.resolve(cursor(10));
|
||||
await vi.waitFor(() => expect(test.execute).toHaveBeenCalledTimes(1));
|
||||
expect(test.execute.mock.calls[0]?.[0]).toEqual(highlight);
|
||||
} finally { follower.stop(); }
|
||||
});
|
||||
|
||||
it("executes same-batch chart commands in sequence and waits for open before close", async () => {
|
||||
const painted = deferred<void>();
|
||||
const execute = vi.fn((command: InvestigationPresentationCommand) => command.action === "open_chart" ? painted.promise : Promise.resolve());
|
||||
const test = setup({ execute });
|
||||
test.follower.start();
|
||||
test.channel.publish({ type: "open" });
|
||||
try {
|
||||
await vi.waitFor(() => expect(test.onStatus).toHaveBeenCalledWith("live"));
|
||||
test.channel.publish({ type: "events", events: [event(12, closeChart), event(11, openChart)] });
|
||||
expect(execute.mock.calls.map(([command]) => command.action)).toEqual(["open_chart"]);
|
||||
painted.resolve();
|
||||
await vi.waitFor(() => expect(execute.mock.calls.map(([command]) => command.action)).toEqual(["open_chart", "close_chart"]));
|
||||
} finally { painted.resolve(); test.follower.stop(); }
|
||||
});
|
||||
|
||||
it("verifies a new generation without losing its first command or fetching a new baseline", async () => {
|
||||
const fetchIdentity = vi.fn(async () => cursor(10, 2));
|
||||
const test = setup({ fetchIdentity });
|
||||
test.follower.start();
|
||||
test.channel.publish({ type: "open" });
|
||||
try {
|
||||
await vi.waitFor(() => expect(test.onStatus).toHaveBeenCalledWith("live"));
|
||||
test.channel.publish({ type: "events", events: [event(11, highlight, 2)] });
|
||||
await vi.waitFor(() => expect(test.execute).toHaveBeenCalledTimes(1));
|
||||
expect(fetchIdentity).toHaveBeenCalledTimes(1);
|
||||
expect(test.fetchCursor).toHaveBeenCalledTimes(1);
|
||||
expect(test.execute.mock.calls[0]?.[0]).toEqual(highlight);
|
||||
} finally { test.follower.stop(); }
|
||||
});
|
||||
|
||||
it("retains a command while identity lookup is stale and applies it once the task snapshot catches up", async () => {
|
||||
const test = setup();
|
||||
test.follower.start();
|
||||
test.channel.publish({ type: "open" });
|
||||
try {
|
||||
await vi.waitFor(() => expect(test.onStatus).toHaveBeenCalledWith("live"));
|
||||
test.channel.publish({ type: "events", events: [event(11, highlight, 2)] });
|
||||
await vi.waitFor(() => expect(test.fetchIdentity).toHaveBeenCalledTimes(1));
|
||||
expect(test.execute).not.toHaveBeenCalled();
|
||||
test.follower.updateIdentity(cursor(10, 2));
|
||||
await vi.waitFor(() => expect(test.execute).toHaveBeenCalledTimes(1));
|
||||
expect(test.fetchCursor).toHaveBeenCalledTimes(1);
|
||||
} finally { test.follower.stop(); }
|
||||
});
|
||||
|
||||
it("aborts on disconnect and uses a fresh reconnect baseline without replaying old commands", async () => {
|
||||
const oldDisplay = deferred<void>();
|
||||
const execute = vi.fn(async (_command: InvestigationPresentationCommand, _signal: AbortSignal) => {})
|
||||
.mockImplementationOnce(() => oldDisplay.promise);
|
||||
const fetchCursor = vi.fn(async () => cursor(10)).mockResolvedValueOnce(cursor(10)).mockResolvedValueOnce(cursor(20));
|
||||
const test = setup({ execute, fetchCursor });
|
||||
test.follower.start();
|
||||
test.channel.publish({ type: "open" });
|
||||
try {
|
||||
await vi.waitFor(() => expect(test.onStatus).toHaveBeenCalledWith("live"));
|
||||
test.channel.publish({ type: "events", events: [event(11)] });
|
||||
const oldSignal = execute.mock.calls[0]?.[1];
|
||||
test.channel.publish({ type: "disconnect" });
|
||||
expect(oldSignal?.aborted).toBe(true);
|
||||
const clearCount = test.clear.mock.calls.length;
|
||||
test.channel.publish({ type: "events", events: [event(12, openChart)] });
|
||||
test.channel.publish({ type: "open" });
|
||||
await vi.waitFor(() => expect(fetchCursor).toHaveBeenCalledTimes(2));
|
||||
test.channel.publish({ type: "events", events: [event(12, openChart), event(20, openChart), event(21, closeChart)] });
|
||||
oldDisplay.resolve();
|
||||
await vi.waitFor(() => expect(execute.mock.calls.map(([command]) => command.action)).toEqual(["highlight_features", "close_chart"]));
|
||||
expect(test.clear.mock.calls.length).toBeGreaterThan(clearCount);
|
||||
expect(execute.mock.calls[1]?.[1].aborted).toBe(false);
|
||||
} finally { oldDisplay.resolve(); test.follower.stop(); }
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,162 @@
|
||||
import {
|
||||
investigationPresentationEventSchema,
|
||||
type InvestigationPresentationCommand,
|
||||
type InvestigationPresentationCursor,
|
||||
type InvestigationPresentationEvent
|
||||
} from "../api/investigation-presentation";
|
||||
import type { WorkbenchEvent } from "../api/workbench-sync-client";
|
||||
import type { WorkbenchStreamNotification } from "./workbench-event-channel";
|
||||
|
||||
export type InvestigationIdentity = Pick<InvestigationPresentationCursor, "task_id" | "execution_generation" | "active_run">;
|
||||
type Options = {
|
||||
taskId: string;
|
||||
subscribe: (listener: (notification: WorkbenchStreamNotification) => void) => () => void;
|
||||
fetchCursor: (signal: AbortSignal) => Promise<InvestigationPresentationCursor>;
|
||||
fetchIdentity: (signal: AbortSignal) => Promise<InvestigationIdentity>;
|
||||
execute: (command: InvestigationPresentationCommand, signal: AbortSignal) => Promise<void>;
|
||||
clear: () => void;
|
||||
stopPulse: () => void;
|
||||
onStatus: (status: "connecting" | "live" | "disconnected" | "error", error?: string) => void;
|
||||
};
|
||||
type Pending = { sequence: number; event: InvestigationPresentationEvent };
|
||||
|
||||
/** No replay: a fresh authorized baseline is taken after every connection, with the listener already attached. */
|
||||
export class InvestigationPresentationFollower {
|
||||
private unsubscribe?: () => void;
|
||||
private binding: AbortController | null = null;
|
||||
private work = new AbortController();
|
||||
private identity: InvestigationIdentity | null = null;
|
||||
private baseline: number | null = null;
|
||||
private lastSeen = 0;
|
||||
private pending: Pending[] = [];
|
||||
private draining = false;
|
||||
private epoch = 0;
|
||||
private lookupKey: string | null = null;
|
||||
|
||||
constructor(private options: Options) {}
|
||||
|
||||
start() { this.unsubscribe = this.options.subscribe((notification) => this.receive(notification)); }
|
||||
|
||||
stop() {
|
||||
this.unsubscribe?.();
|
||||
this.unsubscribe = undefined;
|
||||
this.disconnect();
|
||||
}
|
||||
|
||||
updateIdentity(identity: InvestigationIdentity) {
|
||||
if (identity.task_id !== this.options.taskId || this.identity && identity.execution_generation < this.identity.execution_generation) return;
|
||||
const changed = this.identity && (identity.execution_generation !== this.identity.execution_generation
|
||||
|| identity.active_run && this.identity.active_run && identity.active_run.run_id !== this.identity.active_run.run_id);
|
||||
if (changed) {
|
||||
this.work.abort();
|
||||
this.work = new AbortController();
|
||||
this.options.clear();
|
||||
}
|
||||
// A finished run may no longer be active; retain its identity to finish already received close commands.
|
||||
this.identity = identity.active_run || !this.identity || identity.execution_generation !== this.identity.execution_generation
|
||||
? identity : { ...identity, active_run: this.identity.active_run ? { ...this.identity.active_run, status: "succeeded" } : null };
|
||||
if (!identity.active_run || identity.active_run.status !== "running") this.options.stopPulse();
|
||||
void this.drain();
|
||||
}
|
||||
|
||||
private receive(notification: WorkbenchStreamNotification) {
|
||||
if (notification.type === "open") { void this.connect(); return; }
|
||||
if (notification.type !== "events") { this.disconnect(); return; }
|
||||
if (!this.binding) return;
|
||||
for (const event of notification.events) this.add(event);
|
||||
void this.drain();
|
||||
}
|
||||
|
||||
private add(event: WorkbenchEvent) {
|
||||
if (!this.binding) return;
|
||||
if (event.topic !== "operational.presentation.requested" || event.metadata.task_id !== this.options.taskId) return;
|
||||
if (event.event_seq <= (this.baseline ?? -1) || event.event_seq <= this.lastSeen || this.pending.some((item) => item.sequence === event.event_seq)) return;
|
||||
const parsed = investigationPresentationEventSchema.safeParse(event.metadata);
|
||||
if (!parsed.success) { this.options.onStatus("error", "调查展示参数未通过校验"); return; }
|
||||
if (this.pending.length >= 1000) { this.disconnect(); this.options.onStatus("error", "调查展示积压，请关闭并重新开启跟随"); return; }
|
||||
this.pending.push({ sequence: event.event_seq, event: parsed.data });
|
||||
this.pending.sort((a, b) => a.sequence - b.sequence);
|
||||
}
|
||||
|
||||
private async connect() {
|
||||
this.disconnect();
|
||||
const epoch = this.epoch;
|
||||
const controller = new AbortController();
|
||||
this.binding = controller;
|
||||
this.options.onStatus("connecting");
|
||||
try {
|
||||
const cursor = await this.options.fetchCursor(controller.signal);
|
||||
if (controller.signal.aborted || epoch !== this.epoch) return;
|
||||
if (cursor.task_id !== this.options.taskId) throw new Error("调查游标与所选任务不一致");
|
||||
this.baseline = cursor.event_cursor;
|
||||
this.lastSeen = cursor.event_cursor;
|
||||
this.pending = this.pending.filter((item) => item.sequence > cursor.event_cursor);
|
||||
this.updateIdentity(cursor);
|
||||
this.options.onStatus("live");
|
||||
void this.drain();
|
||||
} catch (error) {
|
||||
if (!controller.signal.aborted) { this.disconnect(); this.options.onStatus("error", error instanceof Error ? error.message : "调查展示连接失败"); }
|
||||
}
|
||||
}
|
||||
|
||||
private disconnect() {
|
||||
this.epoch += 1;
|
||||
this.binding?.abort();
|
||||
this.binding = null;
|
||||
this.work.abort();
|
||||
this.work = new AbortController();
|
||||
this.identity = null;
|
||||
this.baseline = null;
|
||||
this.lastSeen = 0;
|
||||
this.lookupKey = null;
|
||||
this.pending = [];
|
||||
this.options.clear();
|
||||
this.options.onStatus("disconnected");
|
||||
}
|
||||
|
||||
private async resolveIdentity(event: InvestigationPresentationEvent) {
|
||||
const key = `${event.execution_generation}:${event.run_id}`;
|
||||
if (this.lookupKey === key || !this.binding) return;
|
||||
this.lookupKey = key;
|
||||
const binding = this.binding;
|
||||
try {
|
||||
const identity = await this.options.fetchIdentity(binding.signal);
|
||||
if (!binding.signal.aborted) this.updateIdentity(identity);
|
||||
} catch (error) {
|
||||
if (!binding.signal.aborted) this.options.onStatus("error", error instanceof Error ? error.message : "调查运行状态核验失败");
|
||||
}
|
||||
}
|
||||
|
||||
private async drain() {
|
||||
if (this.draining || this.baseline === null || !this.identity) return;
|
||||
this.draining = true;
|
||||
const epoch = this.epoch;
|
||||
try {
|
||||
while (this.pending.length && epoch === this.epoch) {
|
||||
const next = this.pending[0];
|
||||
if (next.sequence <= this.lastSeen || next.event.execution_generation < this.identity!.execution_generation) { this.pending.shift(); continue; }
|
||||
if (next.event.execution_generation !== this.identity!.execution_generation || next.event.run_id !== this.identity!.active_run?.run_id) {
|
||||
await this.resolveIdentity(next.event);
|
||||
if (epoch !== this.epoch) break;
|
||||
if (next.event.execution_generation !== this.identity!.execution_generation || next.event.run_id !== this.identity!.active_run?.run_id) break;
|
||||
}
|
||||
this.pending.shift();
|
||||
this.lastSeen = next.sequence;
|
||||
const signal = this.work.signal;
|
||||
try {
|
||||
const command = next.event.command.action === "highlight_features" && this.identity!.active_run?.status !== "running"
|
||||
? { ...next.event.command, effect: "steady" as const } : next.event.command;
|
||||
await this.options.execute(command, signal);
|
||||
if (!signal.aborted && epoch === this.epoch) this.options.onStatus("live");
|
||||
if (this.identity!.active_run?.status !== "running") this.options.stopPulse();
|
||||
} catch (error) {
|
||||
if (!signal.aborted && epoch === this.epoch) this.options.onStatus("error", error instanceof Error ? error.message : "调查展示失败");
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
this.draining = false;
|
||||
// A reconnect can establish a new baseline while an old asynchronous command is aborting.
|
||||
if (epoch !== this.epoch && this.baseline !== null && this.pending.length) void this.drain();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { collectRecentAgentActivities } from "./use-agent-prompt-suggestions";
|
||||
import type { AgentChatSessionSummary } from "@/features/agent";
|
||||
|
||||
describe("recent activities for Agent suggestions", () => {
|
||||
it("uses updated ordinary conversations, excludes stale/private task sessions, and preserves real versus demo origins", () => {
|
||||
const now = Date.parse("2026-10-09T08:00:00Z");
|
||||
const session = (id: string, updatedAt: number, purpose: AgentChatSessionSummary["purpose"]): AgentChatSessionSummary => ({
|
||||
id, title: id, purpose, createdAt: now - 10 * 24 * 60 * 60_000, updatedAt,
|
||||
});
|
||||
const activities = collectRecentAgentActivities({
|
||||
alerts: [
|
||||
{ id: "demo", title: "演示压力告警", description: "演示", time: "15:00", occurredAt: "2026-10-09T07:00:00Z" },
|
||||
{ id: "operational-task-notification:real", title: "现场核查待办", description: "真实后台任务", time: "15:05", occurredAt: "2026-10-09T07:05:00Z" },
|
||||
],
|
||||
alertDataSource: "mock", conditions: [], runtimeDataSource: "mock", artifact: null,
|
||||
sessions: [session("updated-old-conversation", now - 60_000, "ordinary_chat"), session("stale", now - 8 * 24 * 60 * 60_000, "ordinary_chat"), session("task-private", now, "task_collaboration")],
|
||||
}, now);
|
||||
expect(activities.find((item) => item.id === "alert:demo")?.origin).toBe("demo");
|
||||
expect(activities.find((item) => item.id === "alert:operational-task-notification:real")?.origin).toBe("backend");
|
||||
expect(activities.filter((item) => item.kind === "session").map((item) => item.title)).toEqual(["updated-old-conversation"]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,81 @@
|
||||
import { useEffect, useMemo, useRef, useState } from "react";
|
||||
import type { AgentApiClient, AgentChatSessionSummary, AgentPromptSuggestion, AgentRecentActivity } from "@/features/agent";
|
||||
import type { WorkbenchAlert, ScheduledConditionItem, WorkbenchAlertDataSource, WorkbenchRuntimeDataSource } from "../types";
|
||||
import type { AnalysisArtifact } from "../workspace/workspace-model";
|
||||
|
||||
type ActivityInputs = {
|
||||
alerts: WorkbenchAlert[];
|
||||
alertDataSource: WorkbenchAlertDataSource;
|
||||
conditions: ScheduledConditionItem[];
|
||||
runtimeDataSource: WorkbenchRuntimeDataSource;
|
||||
artifact: AnalysisArtifact | null;
|
||||
sessions: AgentChatSessionSummary[];
|
||||
};
|
||||
|
||||
const activityLabels = { alert: "告警", task: "任务", analysis: "分析成果", session: "近期会话" } as const;
|
||||
const timeFormatter = new Intl.DateTimeFormat("zh-CN", { month: "2-digit", day: "2-digit", hour: "2-digit", minute: "2-digit", hour12: false });
|
||||
type DisplaySuggestion = AgentPromptSuggestion & { sourceLabel: string };
|
||||
|
||||
export function collectRecentAgentActivities(input: ActivityInputs, now = Date.now()): AgentRecentActivity[] {
|
||||
const origin = (source: string) => source === "mock" || source === "demo" ? "demo" as const : "backend" as const;
|
||||
const validDate = (time: string | number) => Number.isFinite(new Date(time).getTime());
|
||||
const alerts: AgentRecentActivity[] = input.alerts.filter((item) => item.occurredAt && validDate(item.occurredAt))
|
||||
.sort((a, b) => Date.parse(b.occurredAt!) - Date.parse(a.occurredAt!)).slice(0, 4).map((item) => ({
|
||||
id: `alert:${item.id}`, kind: "alert", title: item.title.slice(0, 160), occurred_at: new Date(item.occurredAt!).toISOString(),
|
||||
summary: item.description.slice(0, 600), status: item.severity,
|
||||
origin: item.id.startsWith("operational-task-notification:") ? "backend" : origin(input.alertDataSource),
|
||||
}));
|
||||
const tasks: AgentRecentActivity[] = input.conditions.filter((item) => validDate(item.updatedAt))
|
||||
.sort((a, b) => b.updatedAt - a.updatedAt).slice(0, 3).map((item) => ({
|
||||
id: `task:${item.id}`, kind: "task", title: item.title.slice(0, 160), occurred_at: new Date(item.updatedAt).toISOString(),
|
||||
summary: item.summary.slice(0, 600), status: item.status,
|
||||
origin: origin(item.kind === "condition" ? item.dataSource ?? input.runtimeDataSource : input.runtimeDataSource),
|
||||
}));
|
||||
const artifact = input.artifact;
|
||||
const analyses: AgentRecentActivity[] = artifact && validDate(artifact.generatedAt) ? [{
|
||||
id: `analysis:${artifact.id}`, kind: "analysis", title: artifact.title.slice(0, 160), occurred_at: new Date(artifact.generatedAt).toISOString(),
|
||||
summary: artifact.summary.slice(0, 600), origin: artifact.origin ?? "backend",
|
||||
}] : [];
|
||||
const sessions: AgentRecentActivity[] = input.sessions.filter((item) => item.purpose === "ordinary_chat" && item.title.trim() !== "新对话" &&
|
||||
validDate(item.updatedAt) && item.updatedAt >= now - 7 * 24 * 60 * 60_000 && item.updatedAt <= now)
|
||||
.sort((a, b) => b.updatedAt - a.updatedAt).slice(0, 4).map((item) => ({
|
||||
id: `session:${item.id}`, kind: "session", title: item.title.slice(0, 160), occurred_at: new Date(item.updatedAt).toISOString(), origin: "backend",
|
||||
}));
|
||||
return [...new Map([...alerts, ...tasks, ...analyses, ...sessions].map((item) => [item.id, item])).values()];
|
||||
}
|
||||
|
||||
export function useAgentPromptSuggestions({ enabled, model, sessionId, request, alerts, alertDataSource, conditions, runtimeDataSource, artifact, sessions }: ActivityInputs & {
|
||||
enabled: boolean;
|
||||
model: string;
|
||||
sessionId: string | null;
|
||||
request: AgentApiClient["generatePromptSuggestions"];
|
||||
}) {
|
||||
const activities = useMemo(() => collectRecentAgentActivities({ alerts, alertDataSource, conditions, runtimeDataSource, artifact, sessions }),
|
||||
[alerts, alertDataSource, conditions, runtimeDataSource, artifact, sessions]);
|
||||
const snapshot = useRef(activities);
|
||||
useEffect(() => { snapshot.current = activities; }, [activities]);
|
||||
const key = JSON.stringify([enabled, model, sessionId]);
|
||||
const [state, setState] = useState<{ key: string; suggestions: DisplaySuggestion[]; loading: boolean; unavailable: boolean }>({
|
||||
key: "", suggestions: [], loading: false, unavailable: false,
|
||||
});
|
||||
useEffect(() => {
|
||||
if (!enabled) return;
|
||||
const controller = new AbortController();
|
||||
const context = snapshot.current;
|
||||
setState({ key, suggestions: [], loading: true, unavailable: false });
|
||||
void request(context, model, controller.signal).then((suggestions) => {
|
||||
if (controller.signal.aborted) return;
|
||||
const sourceById = new Map(context.map((item) => [item.id, item]));
|
||||
setState({ key, loading: false, unavailable: false, suggestions: suggestions.map((item) => {
|
||||
const source = item.activityId ? sourceById.get(item.activityId) : undefined;
|
||||
return { ...item, sourceLabel: source
|
||||
? `${source.origin === "demo" ? "演示" : ""}${activityLabels[source.kind]} · ${timeFormatter.format(new Date(source.occurred_at))}`
|
||||
: "供水任务建议" };
|
||||
}) });
|
||||
}).catch(() => {
|
||||
if (!controller.signal.aborted) setState({ key, suggestions: [], loading: false, unavailable: true });
|
||||
});
|
||||
return () => controller.abort();
|
||||
}, [enabled, key, model, request]);
|
||||
return state.key === key ? state : { suggestions: [], loading: enabled, unavailable: false };
|
||||
}
|
||||
@@ -0,0 +1,149 @@
|
||||
import { useCallback, useEffect, useMemo, useRef, useState, type RefObject } from "react";
|
||||
import type { FeatureCollection } from "geojson";
|
||||
import type { Map as MapLibreMap, PaddingOptions } from "maplibre-gl";
|
||||
import type { AccessTokenProvider } from "@/shared/auth/keycloak-auth";
|
||||
import { fetchOperationalTask, fetchOperationalTaskPresentationCursor, type OperationalTask } from "../api/operational-task-client";
|
||||
import type { InvestigationMapTarget } from "../api/investigation-report";
|
||||
import type { InvestigationOverlay, InvestigationPresentationCommand } from "../api/investigation-presentation";
|
||||
import { InvestigationMapPresentation } from "../map/investigation-map-presentation";
|
||||
import type { WorkbenchMapController } from "../map/workbench-map-controller";
|
||||
import type { WorkbenchSyncState } from "./use-workbench-sync";
|
||||
import { InvestigationPresentationFollower } from "./investigation-presentation-follower";
|
||||
|
||||
export type InvestigationChart = Extract<InvestigationPresentationCommand, { action: "open_chart" }>;
|
||||
type Options = {
|
||||
enabled: boolean; task: OperationalTask | null; projectId: string | null;
|
||||
sync: WorkbenchSyncState; getAccessToken?: AccessTokenProvider;
|
||||
mapRef: RefObject<MapLibreMap | null>; mapReady: boolean;
|
||||
controller: WorkbenchMapController; getPadding: () => PaddingOptions;
|
||||
onClearLayers: () => void;
|
||||
};
|
||||
|
||||
function frame(signal: AbortSignal) {
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
signal.throwIfAborted();
|
||||
const abort = () => { cancelAnimationFrame(id); reject(signal.reason); };
|
||||
const id = requestAnimationFrame(() => { signal.removeEventListener("abort", abort); resolve(); });
|
||||
signal.addEventListener("abort", abort, { once: true });
|
||||
});
|
||||
}
|
||||
|
||||
/** Separate commands need a rendered frame; an open and close in one SSE batch must not collapse into one React update. */
|
||||
async function waitForChart(chartId: string, signal: AbortSignal) {
|
||||
const deadline = Date.now() + 5000;
|
||||
while (Date.now() < deadline) {
|
||||
await frame(signal);
|
||||
const element = [...document.querySelectorAll<HTMLElement>("[data-investigation-chart]")].find((item) => item.dataset.investigationChart === chartId);
|
||||
if (element && element.getBoundingClientRect().height > 0) {
|
||||
const visibleUntil = Date.now() + 500;
|
||||
while (Date.now() < visibleUntil) await frame(signal);
|
||||
return;
|
||||
}
|
||||
}
|
||||
throw new Error("图表当前不可见");
|
||||
}
|
||||
|
||||
export function useInvestigationPresentation(options: Options) {
|
||||
const valuesRef = useRef(options);
|
||||
valuesRef.current = options;
|
||||
const [charts, setCharts] = useState<InvestigationChart[]>([]);
|
||||
const [overlay, setOverlay] = useState<InvestigationOverlay | null>(null);
|
||||
const [cameraPaused, setCameraPaused] = useState(false);
|
||||
const cameraPausedRef = useRef(false);
|
||||
const [status, setStatus] = useState<"connecting" | "live" | "disconnected" | "error">("disconnected");
|
||||
const [error, setError] = useState<string | null>(null);
|
||||
const [highlightLayers, setHighlightLayers] = useState<string[]>([]);
|
||||
const followerRef = useRef<InvestigationPresentationFollower | null>(null);
|
||||
const manualRequestRef = useRef<AbortController | null>(null);
|
||||
const scene = useMemo(() => new InvestigationMapPresentation(() => options.mapRef.current, () => valuesRef.current.getPadding()), [options.mapRef]);
|
||||
const taskId = options.task?.task_id ?? null;
|
||||
const subscribe = options.sync.subscribePresentationEvents;
|
||||
|
||||
const clear = useCallback(() => {
|
||||
scene.clear(); setCharts([]); setOverlay(null); setHighlightLayers([]);
|
||||
valuesRef.current.onClearLayers();
|
||||
}, [scene]);
|
||||
|
||||
useEffect(() => {
|
||||
manualRequestRef.current?.abort();
|
||||
cameraPausedRef.current = false;
|
||||
setCameraPaused(false);
|
||||
if (!options.enabled || !taskId) { clear(); return; }
|
||||
const follower = new InvestigationPresentationFollower({
|
||||
taskId, subscribe,
|
||||
fetchCursor: (signal) => fetchOperationalTaskPresentationCursor(taskId, options.getAccessToken, signal),
|
||||
fetchIdentity: (signal) => fetchOperationalTask(taskId, options.getAccessToken, signal),
|
||||
clear,
|
||||
stopPulse: () => scene.stopPulse(),
|
||||
onStatus: (next, message) => { setStatus(next); setError(message ?? null); },
|
||||
execute: async (command, signal) => {
|
||||
signal.throwIfAborted();
|
||||
if (command.action === "highlight_features") {
|
||||
const collection = await valuesRef.current.controller.queryPresentationFeatures(command.layer, command.feature_ids, signal);
|
||||
signal.throwIfAborted();
|
||||
scene.highlight(collection, command.effect === "pulse", !cameraPausedRef.current);
|
||||
setHighlightLayers([command.layer]);
|
||||
} else if (command.action === "clear_highlights") {
|
||||
scene.clearHighlights(); setHighlightLayers([]);
|
||||
} else if (command.action === "open_chart") {
|
||||
setCharts((current) => [...current.filter((item) => item.chart_id !== command.chart_id).slice(-5), command]);
|
||||
await waitForChart(command.chart_id, signal);
|
||||
} else if (command.action === "close_chart") {
|
||||
setCharts((current) => current.filter((item) => item.chart_id !== command.chart_id));
|
||||
await frame(signal);
|
||||
} else if (command.action === "set_overlay") {
|
||||
const next = { ...command, visible: true };
|
||||
scene.setOverlay(next, !cameraPausedRef.current); setOverlay(next);
|
||||
} else {
|
||||
scene.clearOverlay(command.overlay_id);
|
||||
setOverlay((current) => current?.overlay_id === command.overlay_id ? null : current);
|
||||
}
|
||||
}
|
||||
});
|
||||
followerRef.current = follower;
|
||||
follower.start();
|
||||
return () => { follower.stop(); if (followerRef.current === follower) followerRef.current = null; };
|
||||
}, [clear, options.enabled, options.getAccessToken, options.projectId, scene, subscribe, taskId]);
|
||||
|
||||
useEffect(() => {
|
||||
if (options.task) followerRef.current?.updateIdentity(options.task);
|
||||
}, [options.task]);
|
||||
|
||||
const pauseCamera = useCallback(() => {
|
||||
cameraPausedRef.current = true; setCameraPaused(true);
|
||||
}, []);
|
||||
useEffect(() => {
|
||||
const map = options.mapRef.current;
|
||||
if (!map || !options.mapReady || !options.enabled) return;
|
||||
const canvas = map.getCanvasContainer();
|
||||
// Canvas input excludes programmatic fitBounds/easeTo and also covers touch/keyboard navigation.
|
||||
canvas.addEventListener("pointerdown", pauseCamera);
|
||||
canvas.addEventListener("wheel", pauseCamera, { passive: true });
|
||||
canvas.addEventListener("keydown", pauseCamera);
|
||||
return () => { canvas.removeEventListener("pointerdown", pauseCamera); canvas.removeEventListener("wheel", pauseCamera); canvas.removeEventListener("keydown", pauseCamera); };
|
||||
}, [options.enabled, options.mapReady, options.mapRef, pauseCamera]);
|
||||
useEffect(() => () => { manualRequestRef.current?.abort(); scene.destroy(); }, [scene]);
|
||||
|
||||
const locateTargets = useCallback(async (targets: InvestigationMapTarget[]) => {
|
||||
manualRequestRef.current?.abort();
|
||||
const controller = new AbortController(); manualRequestRef.current = controller;
|
||||
pauseCamera();
|
||||
const groups = new Map<InvestigationMapTarget["source_id"], string[]>();
|
||||
for (const target of targets) groups.set(target.source_id, [...groups.get(target.source_id) ?? [], target.feature_id]);
|
||||
const collections = await Promise.all([...groups].map(([source, ids]) => valuesRef.current.controller.queryPresentationFeatures(source, ids, controller.signal)));
|
||||
controller.signal.throwIfAborted();
|
||||
const data: FeatureCollection = { type: "FeatureCollection", features: collections.flatMap((collection) => collection.features) };
|
||||
if (!data.features.length) throw new Error("未找到关联地图要素");
|
||||
scene.highlight(data, false, true); setHighlightLayers([...groups.keys()]);
|
||||
}, [pauseCamera, scene]);
|
||||
|
||||
return {
|
||||
charts, overlay, cameraPaused, status, error, highlightLayers,
|
||||
pauseCamera,
|
||||
resumeCamera: () => { cameraPausedRef.current = false; setCameraPaused(false); },
|
||||
locateTargets,
|
||||
closeChart: (id: string) => setCharts((current) => current.filter((item) => item.chart_id !== id)),
|
||||
toggleOverlay: () => { if (overlay) { scene.showOverlay(!overlay.visible); setOverlay({ ...overlay, visible: !overlay.visible }); } },
|
||||
fitOverlay: () => { pauseCamera(); scene.fitOverlay(); }
|
||||
};
|
||||
}
|
||||
@@ -40,6 +40,7 @@ export function collectOperationalTaskEventChanges(events: WorkbenchEvent[]) {
|
||||
const deletedTaskIds = new Set<string>();
|
||||
let notificationsChanged = false;
|
||||
for (const event of events) {
|
||||
if (event.topic === "operational.presentation.requested") continue;
|
||||
if (event.topic === "operational.notification.changed") {
|
||||
notificationsChanged = true;
|
||||
continue;
|
||||
|
||||
@@ -0,0 +1,95 @@
|
||||
import { cleanup, fireEvent, render, screen } from "@testing-library/react";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { animate } from "motion/mini";
|
||||
import { useReportUpdateMotion } from "./use-report-update-motion";
|
||||
|
||||
vi.mock("motion/mini", () => ({ animate: vi.fn() }));
|
||||
|
||||
function Reader({ revision = "1", scope = "task-1", live = true }) {
|
||||
const ref = useReportUpdateMotion(scope, revision, live);
|
||||
return <article ref={ref} aria-label="report"><header>调查报告</header><section><p>{revision}</p><details open><summary>依据</summary>{revision}</details></section></article>;
|
||||
}
|
||||
|
||||
afterEach(() => { cleanup(); vi.unstubAllGlobals(); vi.restoreAllMocks(); });
|
||||
|
||||
function motion(reduced = false) {
|
||||
const cancel = vi.fn();
|
||||
vi.mocked(animate).mockReset();
|
||||
vi.mocked(animate).mockReturnValue({ cancel } as unknown as ReturnType<typeof animate>);
|
||||
vi.stubGlobal("matchMedia", vi.fn(() => ({ matches: reduced, addEventListener: vi.fn(), removeEventListener: vi.fn() })));
|
||||
// jsdom does not provide Web Animations.
|
||||
const attach = (element: HTMLElement) => Array.from(element.children)
|
||||
.forEach((part) => Object.defineProperty(part, "animate", { value: vi.fn() }));
|
||||
return { animate, cancel, attach };
|
||||
}
|
||||
|
||||
describe("report update motion", () => {
|
||||
it("animates only a new revision and keeps the reading DOM", () => {
|
||||
const { animate, cancel, attach } = motion();
|
||||
const { rerender, unmount } = render(<Reader />);
|
||||
const section = screen.getByRole("article", { name: "report" });
|
||||
attach(section);
|
||||
const details = section.querySelector("details")!;
|
||||
expect(animate).not.toHaveBeenCalled();
|
||||
rerender(<Reader />);
|
||||
expect(animate).not.toHaveBeenCalled();
|
||||
rerender(<Reader revision="2" />);
|
||||
expect(animate).toHaveBeenCalledTimes(2);
|
||||
expect(animate).toHaveBeenNthCalledWith(1,
|
||||
section.querySelector("header"),
|
||||
{ opacity: [0, 1], translate: ["0 18px", "0 0"], scale: [0.98, 1] },
|
||||
{ duration: 0.4, delay: 0, ease: [0.22, 1, 0.36, 1] }
|
||||
);
|
||||
expect(animate).toHaveBeenNthCalledWith(2, section.querySelector("section"), expect.any(Object), expect.objectContaining({ delay: 0.045 }));
|
||||
expect(section.querySelector("details")).toBe(details);
|
||||
expect(details.open).toBe(true);
|
||||
rerender(<Reader revision="3" />);
|
||||
expect(cancel).toHaveBeenCalledTimes(2);
|
||||
unmount();
|
||||
expect(cancel).toHaveBeenCalledTimes(4);
|
||||
});
|
||||
|
||||
it("skips history, returning to live content, and task switches", () => {
|
||||
const { animate, attach } = motion();
|
||||
const { rerender } = render(<Reader />);
|
||||
attach(screen.getByRole("article", { name: "report" }));
|
||||
rerender(<Reader revision="2" live={false} />);
|
||||
rerender(<Reader revision="3" />);
|
||||
rerender(<Reader scope="task-2" revision="4" />);
|
||||
expect(animate).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("respects reduced motion", () => {
|
||||
const { animate, attach } = motion(true);
|
||||
const { rerender } = render(<Reader />);
|
||||
attach(screen.getByRole("article", { name: "report" }));
|
||||
rerender(<Reader revision="2" />);
|
||||
expect(animate).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("keeps report updates readable when matchMedia is unavailable", () => {
|
||||
const { animate, attach } = motion();
|
||||
vi.stubGlobal("matchMedia", undefined);
|
||||
const { rerender } = render(<Reader />);
|
||||
const section = screen.getByRole("article", { name: "report" });
|
||||
attach(section);
|
||||
rerender(<Reader revision="2" />);
|
||||
expect(section).toHaveTextContent("2");
|
||||
expect(animate).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("lets interaction interrupt the group and leaves focused content still", () => {
|
||||
const { animate, cancel, attach } = motion();
|
||||
const { rerender } = render(<Reader />);
|
||||
const reader = screen.getByRole("article", { name: "report" });
|
||||
attach(reader);
|
||||
rerender(<Reader revision="2" />);
|
||||
fireEvent.pointerDown(reader);
|
||||
expect(cancel).toHaveBeenCalledTimes(2);
|
||||
reader.querySelector("summary")!.focus();
|
||||
vi.mocked(animate).mockClear();
|
||||
rerender(<Reader revision="3" />);
|
||||
expect(animate).toHaveBeenCalledTimes(1);
|
||||
expect(animate).toHaveBeenCalledWith(reader.querySelector("header"), expect.any(Object), expect.any(Object));
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,43 @@
|
||||
import { useLayoutEffect, useRef } from "react";
|
||||
import { animate } from "motion/mini";
|
||||
|
||||
/** Reveal new published content without remounting the reader or animating heartbeats. */
|
||||
export function useReportUpdateMotion(scope: string, revision: string, live: boolean) {
|
||||
const ref = useRef<HTMLElement>(null);
|
||||
const previous = useRef({ scope, revision, live });
|
||||
|
||||
useLayoutEffect(() => {
|
||||
const before = previous.current;
|
||||
previous.current = { scope, revision, live };
|
||||
if (!live || !before.live || before.scope !== scope || before.revision === revision) return;
|
||||
const reader = ref.current;
|
||||
if (!reader || typeof window.matchMedia !== "function") return;
|
||||
const reducedMotion = window.matchMedia("(prefers-reduced-motion: reduce)");
|
||||
if (reducedMotion.matches) return;
|
||||
// Replay the Bento reveal across the report, not a browser/page reload.
|
||||
// Direct children only: portal details and Mermaid's internal DOM stay intact.
|
||||
const parts = Array.from(reader.querySelectorAll<HTMLElement>(":scope > header, :scope > section, :scope > p"))
|
||||
.filter((part) => typeof part.animate === "function" && !part.contains(document.activeElement));
|
||||
const animations = parts.map((part, index) => {
|
||||
const original = { opacity: part.style.opacity, translate: part.style.translate, scale: part.style.scale };
|
||||
// Individual transforms leave the expand-card shared-layout transform alone.
|
||||
const animation = animate(part, { opacity: [0, 1], translate: ["0 18px", "0 0"], scale: [0.98, 1] }, {
|
||||
duration: 0.4, delay: Math.min(index, 6) * 0.045, ease: [0.22, 1, 0.36, 1]
|
||||
});
|
||||
return () => { animation.cancel(); Object.assign(part.style, original); };
|
||||
});
|
||||
const cancel = () => animations.forEach((restore) => restore());
|
||||
// Interaction wins over the reveal, including expanding a shared-layout card.
|
||||
reader.addEventListener("pointerdown", cancel, true);
|
||||
reader.addEventListener("focusin", cancel);
|
||||
reducedMotion.addEventListener("change", cancel);
|
||||
return () => {
|
||||
reader.removeEventListener("pointerdown", cancel, true);
|
||||
reader.removeEventListener("focusin", cancel);
|
||||
reducedMotion.removeEventListener("change", cancel);
|
||||
cancel();
|
||||
};
|
||||
}, [scope, revision, live]);
|
||||
|
||||
return ref;
|
||||
}
|
||||
@@ -1,13 +1,13 @@
|
||||
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
|
||||
import { useChat } from "@ai-sdk/react";
|
||||
import { DefaultChatTransport } from "ai";
|
||||
import { ResponsesChatTransport, createResponseRequest } from "@/features/agent";
|
||||
import useSWR from "swr";
|
||||
import useSWRImmutable from "swr/immutable";
|
||||
import type { PersonaState } from "@/shared/ai-elements/persona";
|
||||
import type { AccessTokenProvider } from "@/shared/auth/keycloak-auth";
|
||||
import { env } from "@/shared/config/env";
|
||||
import { showMapNotice } from "@/features/map/core";
|
||||
import { findOwnedWorkbenchProject } from "../api/project-client";
|
||||
import type { InvestigationDiscussion } from "../api/investigation-report";
|
||||
import {
|
||||
FrontendActionExecutor,
|
||||
agentTaskDiscussionContextSchema,
|
||||
@@ -15,8 +15,10 @@ import {
|
||||
agentBootstrapKey,
|
||||
agentSessionsKey,
|
||||
createAgentApiClient,
|
||||
useAgentMessageInbox,
|
||||
fetchAgentBootstrap,
|
||||
fetchAgentSessions,
|
||||
resolveAgentModelSelection,
|
||||
hasPendingAgentInput,
|
||||
isUiEnvelopeAllowed,
|
||||
parseUiEnvelopePayload,
|
||||
@@ -25,6 +27,7 @@ import {
|
||||
type AgentConnectionStatus,
|
||||
type AgentEventTriggerContext,
|
||||
type AgentModelOption,
|
||||
type AgentInputDelivery,
|
||||
type AgentPermissionReply,
|
||||
type AgentPermissionRequest,
|
||||
type AgentQuestionRequest,
|
||||
@@ -68,6 +71,8 @@ export function useWorkbenchAgent({
|
||||
const mobileCollapseTimerRef = useRef<number | null>(null);
|
||||
const sessionStreamAbortRef = useRef<AbortController | null>(null);
|
||||
const projectIdPromiseRef = useRef<Promise<string> | null>(null);
|
||||
const [conversationScope, setConversationScope] = useState<"personal" | "project">("project");
|
||||
const conversationScopeRef = useRef<"personal" | "project">("project");
|
||||
const getProjectId = useCallback(() => {
|
||||
projectIdPromiseRef.current ??= findOwnedWorkbenchProject(getAccessToken).then(
|
||||
({ project }) => project.project_id,
|
||||
@@ -78,7 +83,15 @@ export function useWorkbenchAgent({
|
||||
);
|
||||
return projectIdPromiseRef.current;
|
||||
}, [getAccessToken]);
|
||||
const clientRef = useRef(createAgentApiClient(undefined, { getAccessToken, getProjectId }));
|
||||
const getConversationProjectId = useCallback(() =>
|
||||
conversationScopeRef.current === "personal" ? Promise.resolve(undefined) : getProjectId(), [getProjectId]);
|
||||
const clientRef = useRef(createAgentApiClient(undefined, { getAccessToken, getProjectId: getConversationProjectId }));
|
||||
const scopeClient = useMemo(() => createAgentApiClient(undefined, {
|
||||
getAccessToken,
|
||||
getProjectId: conversationScope === "project" ? getProjectId : undefined
|
||||
}), [conversationScope, getAccessToken, getProjectId]);
|
||||
const generatePromptSuggestions = useCallback((...args: Parameters<typeof clientRef.current.generatePromptSuggestions>) =>
|
||||
clientRef.current.generatePromptSuggestions(...args), []);
|
||||
const sessionIdRef = useRef<string | null>(null);
|
||||
const approvalModeRef = useRef<AgentApprovalMode>("request");
|
||||
const registryRef = useRef<UIRegistry | null>(null);
|
||||
@@ -92,7 +105,7 @@ export function useWorkbenchAgent({
|
||||
const [sessionId, setSessionId] = useState<string | null>(null);
|
||||
const [sessionPurpose, setSessionPurpose] = useState<AgentSessionPurpose>("ordinary_chat");
|
||||
const [sessionTaskId, setSessionTaskId] = useState<string | null>(null);
|
||||
const taskDiscussionRef = useRef<{ sessionId: string; context: AgentTaskDiscussionContext } | null>(null);
|
||||
const [taskDiscussion, setTaskDiscussion] = useState<{ sessionId: string; context: AgentTaskDiscussionContext; reportGeneration?: number | null } | null>(null);
|
||||
const [registry, setRegistry] = useState<UIRegistry | null>(null);
|
||||
const [sessionTitle, setSessionTitle] = useState("新对话");
|
||||
const [statusLabel, setStatusLabel] = useState("Agent 后端连接中");
|
||||
@@ -101,11 +114,17 @@ export function useWorkbenchAgent({
|
||||
const [resumedStreaming, setResumedStreaming] = useState(false);
|
||||
const [modelOptions, setModelOptions] = useState<AgentModelOption[]>([]);
|
||||
const [selectedModel, setSelectedModel] = useState("");
|
||||
const [inputDelivery, setInputDelivery] = useState<AgentInputDelivery>("steer");
|
||||
const selectedModelRef = useRef("");
|
||||
useEffect(() => { selectedModelRef.current = selectedModel; }, [selectedModel]);
|
||||
const [promptDraft, setPromptDraft] = useState("");
|
||||
const [promptFocusRequest, setPromptFocusRequest] = useState(0);
|
||||
const [approvalMode, setApprovalModeState] = useState<AgentApprovalMode>("request");
|
||||
const [streamRenderState, setStreamRenderState] = useState<AgentStreamRenderState>({});
|
||||
const [permissionOverrides, setPermissionOverrides] = useState<Record<string, PermissionOverride>>({});
|
||||
const [questionOverrides, setQuestionOverrides] = useState<Record<string, QuestionOverride>>({});
|
||||
const agentMessageInbox = useAgentMessageInbox(clientRef.current, sessionPurpose === "ordinary_chat" && conversationScope === "project" ? sessionId : null);
|
||||
const refreshAgentMessageInbox = agentMessageInbox.refresh;
|
||||
|
||||
useEffect(() => {
|
||||
sessionIdRef.current = sessionId;
|
||||
@@ -154,6 +173,19 @@ export function useWorkbenchAgent({
|
||||
[frontendActionExecutor]
|
||||
);
|
||||
|
||||
const credentialRequestsRef = useRef(new Set<string>());
|
||||
const handleCredentialRefresh = useCallback((data: unknown) => {
|
||||
const requestId = getDataString(data, "request_id");
|
||||
const activeSessionId = getDataString(data, "session_id") ?? sessionIdRef.current;
|
||||
if (!requestId || !activeSessionId || activeSessionId !== sessionIdRef.current) return;
|
||||
if (credentialRequestsRef.current.has(requestId)) return;
|
||||
credentialRequestsRef.current.add(requestId);
|
||||
void clientRef.current.refreshCredential(activeSessionId, requestId).catch(() => {
|
||||
credentialRequestsRef.current.delete(requestId);
|
||||
setStatusLabel("会话凭据续期失败，请重新登录后重试");
|
||||
});
|
||||
}, []);
|
||||
|
||||
const claimEnvelope = useCallback((payload: UIEnvelopePayload, activeSessionId: string) => {
|
||||
if (payload.session_id !== activeSessionId) return false;
|
||||
const key = `${activeSessionId}:${payload.envelope_id}`;
|
||||
@@ -183,6 +215,11 @@ export function useWorkbenchAgent({
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (part.type === "data-agent_message") { refreshAgentMessageInbox(); return; }
|
||||
if (part.type === "data-user_input" && part.data.status === "delivered") {
|
||||
chatRef.current.setMessages((current) => applySessionStreamEvent(current, { type: "user_input", data: part.data }));
|
||||
return;
|
||||
}
|
||||
|
||||
if (part.type === "data-progress") {
|
||||
const title = getDataString(part.data, "title");
|
||||
@@ -191,6 +228,7 @@ export function useWorkbenchAgent({
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (part.type === "data-credential_refresh_required") { handleCredentialRefresh(part.data); return; }
|
||||
if (part.type === "data-frontend_action") { void handleFrontendAction(part.data); return; }
|
||||
|
||||
if (part.type === "data-session_title") {
|
||||
@@ -225,37 +263,37 @@ export function useWorkbenchAgent({
|
||||
});
|
||||
});
|
||||
},
|
||||
[applySessionId, claimEnvelope, handleFrontendAction, onUiEnvelope]
|
||||
[refreshAgentMessageInbox, applySessionId, claimEnvelope, handleCredentialRefresh, handleFrontendAction, onUiEnvelope]
|
||||
);
|
||||
|
||||
const transport = useMemo(
|
||||
() =>
|
||||
new DefaultChatTransport<AgentUiMessage>({
|
||||
api: `${env.TJWATER_AGENT_API_BASE_URL.replace(/\/$/, "")}/api/v1/agent/chat/stream`,
|
||||
new ResponsesChatTransport<AgentUiMessage>({
|
||||
api: `${env.TJWATER_AGENT_URL.replace(/\/$/, "")}/api/v1/responses`,
|
||||
headers: async (): Promise<Record<string, string>> => {
|
||||
const accessToken = await getAccessToken?.();
|
||||
const projectId = await getProjectId();
|
||||
const projectId = await getConversationProjectId();
|
||||
return {
|
||||
...(projectId ? { "X-Project-Id": projectId } : {}),
|
||||
...(accessToken ? { Authorization: `Bearer ${accessToken}` } : {})
|
||||
};
|
||||
},
|
||||
prepareSendMessagesRequest({ id, messages, body, trigger, messageId }) {
|
||||
return {
|
||||
body: {
|
||||
...body,
|
||||
id,
|
||||
messages,
|
||||
trigger,
|
||||
messageId,
|
||||
session_id: readBodyString(body, "session_id") ?? sessionIdRef.current ?? undefined,
|
||||
approval_mode: readBodyString(body, "approval_mode") ?? approvalModeRef.current,
|
||||
capabilities: frontendActionEnabledRef.current ? ["frontend-action@1"] : []
|
||||
}
|
||||
};
|
||||
async prepareSendMessagesRequest({ messages, body }) {
|
||||
const catalog = await scopeClient.getModels();
|
||||
const model = resolveAgentModelSelection(catalog, readBodyString(body, "model") ?? selectedModelRef.current);
|
||||
setModelOptions(catalog.models);
|
||||
setSelectedModel(model);
|
||||
selectedModelRef.current = model;
|
||||
if (!model) throw new Error("Agent 后端当前没有可用模型，请检查模型配置。");
|
||||
return { body: createResponseRequest(messages, {
|
||||
conversation: readBodyString(body, "session_id") ?? sessionIdRef.current ?? undefined,
|
||||
model, approvalMode: readBodyString(body, "approval_mode") ?? approvalModeRef.current,
|
||||
capabilities: frontendActionEnabledRef.current ? ["frontend-action@1"] : [],
|
||||
triggerContext: body?.trigger_context, taskContext: body?.task_context
|
||||
}) };
|
||||
}
|
||||
}),
|
||||
[getAccessToken, getProjectId]
|
||||
[getAccessToken, getConversationProjectId, scopeClient]
|
||||
);
|
||||
|
||||
const chat = useChat<AgentUiMessage>({
|
||||
@@ -287,7 +325,7 @@ export function useWorkbenchAgent({
|
||||
const {
|
||||
data: backendConnection,
|
||||
error: backendConnectionError
|
||||
} = useSWRImmutable(agentBootstrapKey, () => fetchAgentBootstrap(clientRef.current), {
|
||||
} = useSWR([...agentBootstrapKey, getAccessToken, conversationScope], () => fetchAgentBootstrap(scopeClient), {
|
||||
shouldRetryOnError: false
|
||||
});
|
||||
const {
|
||||
@@ -295,7 +333,7 @@ export function useWorkbenchAgent({
|
||||
isLoading: sessionHistoryLoading,
|
||||
isValidating: sessionHistoryValidating,
|
||||
mutate: mutateSessionHistory
|
||||
} = useSWR(agentSessionsKey, () => fetchAgentSessions(clientRef.current), {
|
||||
} = useSWR([...agentSessionsKey, getAccessToken, conversationScope], () => fetchAgentSessions(scopeClient), {
|
||||
revalidateOnFocus: false,
|
||||
shouldRetryOnError: false
|
||||
});
|
||||
@@ -339,7 +377,7 @@ export function useWorkbenchAgent({
|
||||
frontendActionEnabledRef.current = Boolean(backendConnection.frontendActionRegistry);
|
||||
setRegistry(backendConnection.registry);
|
||||
setModelOptions(backendConnection.models);
|
||||
setSelectedModel((current) => current || backendConnection.defaultModel || backendConnection.models[0]?.id || "");
|
||||
setSelectedModel((current) => resolveAgentModelSelection(backendConnection, current));
|
||||
if (chatRef.current.status === "ready") {
|
||||
setConnectionStatus("online");
|
||||
setStatusLabel("准备就绪");
|
||||
@@ -402,6 +440,20 @@ export function useWorkbenchAgent({
|
||||
}, [clearMobileCollapseTimer]);
|
||||
|
||||
const streaming = chat.status === "submitted" || chat.status === "streaming" || resumedStreaming;
|
||||
const { data: pendingInputs = [], mutate: refreshPendingInputs } = useSWR(
|
||||
sessionId ? ["agent", "response-inputs", getAccessToken, conversationScope, sessionId] : null,
|
||||
() => clientRef.current.listResponseInputs(sessionId!),
|
||||
{ refreshInterval: streaming ? 1000 : 0, shouldRetryOnError: false }
|
||||
);
|
||||
const updateInputDelivery = useCallback(async (id: string, delivery: AgentInputDelivery) => {
|
||||
if (!sessionIdRef.current) return;
|
||||
try {
|
||||
const updated = await clientRef.current.updateResponseInput(sessionIdRef.current, id, delivery);
|
||||
await refreshPendingInputs((current = []) => current.map((input) => input.id === id ? updated : input), { revalidate: false });
|
||||
} catch {
|
||||
showMapNotice({ tone: "warning", message: "消息可能已开始处理，无法修改发送方式。" });
|
||||
} finally { void refreshPendingInputs().catch(() => {}); }
|
||||
}, [refreshPendingInputs]);
|
||||
const messages = useMemo(
|
||||
() => toAgentChatMessages(chat.messages, permissionOverrides, questionOverrides),
|
||||
[chat.messages, permissionOverrides, questionOverrides]
|
||||
@@ -446,6 +498,8 @@ export function useWorkbenchAgent({
|
||||
return;
|
||||
}
|
||||
|
||||
if (event.type === "credential_refresh_required") { handleCredentialRefresh(event.data); return; }
|
||||
if (event.type === "agent_message") { refreshAgentMessageInbox(); return; }
|
||||
if (event.type === "session_title") {
|
||||
const title = getDataString(event.data, "title");
|
||||
if (title) {
|
||||
@@ -497,7 +551,7 @@ export function useWorkbenchAgent({
|
||||
|
||||
chatRef.current.setMessages((current) => applySessionStreamEvent(current, event));
|
||||
},
|
||||
[claimEnvelope, handleFrontendAction, onUiEnvelope, refreshSessionHistory]
|
||||
[refreshAgentMessageInbox, claimEnvelope, handleCredentialRefresh, handleFrontendAction, onUiEnvelope, refreshSessionHistory]
|
||||
);
|
||||
|
||||
const startSessionStreamSubscription = useCallback(
|
||||
@@ -557,6 +611,7 @@ export function useWorkbenchAgent({
|
||||
}
|
||||
|
||||
applySessionId(loadedSession.sessionId);
|
||||
setTaskDiscussion(null);
|
||||
setSessionPurpose(loadedSession.purpose);
|
||||
setSessionTaskId(loadedSession.taskId ?? null);
|
||||
setSessionTitle(loadedSession.title);
|
||||
@@ -570,6 +625,7 @@ export function useWorkbenchAgent({
|
||||
startSessionStreamSubscription(loadedSession.sessionId);
|
||||
}
|
||||
await refreshSessionHistory();
|
||||
return true;
|
||||
} catch (error) {
|
||||
showMapNotice({
|
||||
id: "agent-history-status",
|
||||
@@ -582,20 +638,34 @@ export function useWorkbenchAgent({
|
||||
[applySessionId, chat, refreshSessionHistory, startSessionStreamSubscription, stopSessionStreamSubscription, streaming]
|
||||
);
|
||||
|
||||
const openTaskCollaboration = useCallback(async (taskId: string, context?: AgentTaskDiscussionContext) => {
|
||||
const openTaskCollaboration = useCallback(async (taskId: string, discussion?: InvestigationDiscussion) => {
|
||||
const normalizedTaskId = taskId.trim();
|
||||
if (!normalizedTaskId) {
|
||||
throw new Error("工况任务 ID 不能为空。");
|
||||
}
|
||||
if (streaming) {
|
||||
throw new Error("Agent 正在处理上一条指令，完成后再打开任务协作。");
|
||||
}
|
||||
const normalizedContext = agentTaskDiscussionContextSchema.parse(discussion?.context ?? {});
|
||||
conversationScopeRef.current = "project";
|
||||
setConversationScope("project");
|
||||
const collaboration = await clientRef.current.createTaskCollaborationSession(normalizedTaskId);
|
||||
const collaborationSessionId = collaboration.session_id || collaboration.id;
|
||||
if (!collaborationSessionId) {
|
||||
throw new Error("Agent 未返回任务协作会话。");
|
||||
}
|
||||
await loadHistorySession(collaborationSessionId);
|
||||
taskDiscussionRef.current = { sessionId: collaborationSessionId, context: agentTaskDiscussionContextSchema.parse(context ?? {}) };
|
||||
const loaded = await loadHistorySession(collaborationSessionId);
|
||||
if (!loaded || sessionIdRef.current !== collaborationSessionId) {
|
||||
throw new Error("任务协作会话未能打开，请重试。");
|
||||
}
|
||||
setTaskDiscussion({ sessionId: collaborationSessionId, context: normalizedContext, reportGeneration: discussion?.reportGeneration });
|
||||
const draft = discussion?.prompt;
|
||||
if (draft?.trim()) {
|
||||
setPromptDraft((current) => current.trim() ? `${current}\n\n${draft.trim()}` : draft.trim());
|
||||
}
|
||||
setPromptFocusRequest((current) => current + 1);
|
||||
return collaborationSessionId;
|
||||
}, [loadHistorySession]);
|
||||
}, [loadHistorySession, streaming]);
|
||||
|
||||
const renameHistorySession = useCallback(
|
||||
async (nextSessionId: string, title: string) => {
|
||||
@@ -644,6 +714,7 @@ export function useWorkbenchAgent({
|
||||
setSessionTitle("新对话");
|
||||
setSessionPurpose("ordinary_chat");
|
||||
setSessionTaskId(null);
|
||||
setTaskDiscussion(null);
|
||||
setStatusLabel(nextStatusLabel);
|
||||
setConnectionStatus("online");
|
||||
setPermissionOverrides({});
|
||||
@@ -697,6 +768,13 @@ export function useWorkbenchAgent({
|
||||
resetDraftSession();
|
||||
}, [chat, resetDraftSession, stopSessionStreamSubscription]);
|
||||
|
||||
const changeConversationScope = useCallback(async (scope: "personal" | "project") => {
|
||||
if (scope === conversationScopeRef.current || streaming) return;
|
||||
await startNewSession();
|
||||
conversationScopeRef.current = scope;
|
||||
setConversationScope(scope);
|
||||
}, [startNewSession, streaming]);
|
||||
|
||||
const submitPrompt = useCallback(
|
||||
async (prompt: string) => {
|
||||
const normalizedPrompt = prompt.trim();
|
||||
@@ -705,7 +783,19 @@ export function useWorkbenchAgent({
|
||||
}
|
||||
|
||||
if (streaming) {
|
||||
showMapNotice({ tone: "warning", message: "Agent 正在处理上一条指令。" });
|
||||
try {
|
||||
if (!sessionIdRef.current) throw new Error("Agent 正在建立会话，请稍后追加消息。");
|
||||
const receipt = await clientRef.current.submitResponseInput(sessionIdRef.current, normalizedPrompt, inputDelivery, `msg_${crypto.randomUUID().replaceAll("-", "")}`);
|
||||
// A failed list refresh must not turn an accepted message into a
|
||||
// failed send and restore a draft that the user might submit twice.
|
||||
void refreshPendingInputs((current = []) => receipt.status === "pending"
|
||||
? [...current.filter((input) => input.id !== receipt.id), receipt]
|
||||
: current.filter((input) => input.id !== receipt.id), { revalidate: false })
|
||||
.then(() => refreshPendingInputs()).catch(() => {});
|
||||
} catch (error) {
|
||||
showMapNotice({ tone: "error", title: "追加消息失败", message: error instanceof Error ? error.message : "请稍后重试。" });
|
||||
throw error;
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -718,12 +808,12 @@ export function useWorkbenchAgent({
|
||||
session_id: sessionIdRef.current ?? undefined,
|
||||
model: selectedModel || undefined,
|
||||
approval_mode: approvalMode,
|
||||
task_context: taskDiscussionRef.current?.sessionId === sessionIdRef.current ? taskDiscussionRef.current.context : undefined
|
||||
task_context: taskDiscussion?.sessionId === sessionIdRef.current ? taskDiscussion.context : undefined
|
||||
}
|
||||
}
|
||||
);
|
||||
},
|
||||
[approvalMode, chat, selectedModel, streaming]
|
||||
[approvalMode, chat, inputDelivery, refreshPendingInputs, selectedModel, streaming, taskDiscussion]
|
||||
);
|
||||
|
||||
const triggerEvent = useCallback(
|
||||
@@ -795,10 +885,10 @@ export function useWorkbenchAgent({
|
||||
stopSessionStreamSubscription();
|
||||
const activeSessionId = sessionIdRef.current;
|
||||
if (activeSessionId) {
|
||||
void clientRef.current.abort(activeSessionId).catch(() => undefined);
|
||||
void clientRef.current.abort(activeSessionId).then(() => refreshPendingInputs()).catch(() => undefined);
|
||||
}
|
||||
markLastAssistantStreamDone(chatRef.current.messages, setStreamRenderState);
|
||||
}, [chat, stopSessionStreamSubscription]);
|
||||
}, [chat, refreshPendingInputs, stopSessionStreamSubscription]);
|
||||
|
||||
const replyPermission = useCallback(async (permission: AgentPermissionRequest, reply: AgentPermissionReply) => {
|
||||
setPermissionOverrides((current) => ({
|
||||
@@ -907,6 +997,11 @@ export function useWorkbenchAgent({
|
||||
|
||||
return {
|
||||
apiClient: clientRef.current,
|
||||
inputDelivery,
|
||||
setInputDelivery,
|
||||
pendingInputs,
|
||||
updateInputDelivery,
|
||||
agentMessageInbox,
|
||||
approvalMode,
|
||||
collapsePanel,
|
||||
expandPanel,
|
||||
@@ -917,6 +1012,8 @@ export function useWorkbenchAgent({
|
||||
openMobilePanel,
|
||||
closeMobilePanel,
|
||||
connectionStatus,
|
||||
conversationScope,
|
||||
changeConversationScope,
|
||||
panelCollapsing,
|
||||
panelOpen,
|
||||
personaState,
|
||||
@@ -927,15 +1024,19 @@ export function useWorkbenchAgent({
|
||||
renameHistorySession,
|
||||
deleteHistorySession,
|
||||
sessionHistory,
|
||||
generatePromptSuggestions,
|
||||
sessionHistoryLoading: sessionHistoryLoading || sessionHistoryValidating,
|
||||
sessionId,
|
||||
sessionPurpose,
|
||||
sessionTaskId,
|
||||
taskDiscussionContext: taskDiscussion?.sessionId === sessionId ? taskDiscussion.context : undefined,
|
||||
taskDiscussionReportGeneration: taskDiscussion?.sessionId === sessionId ? taskDiscussion.reportGeneration : undefined,
|
||||
sessionTitle,
|
||||
statusLabel,
|
||||
streamRenderState,
|
||||
selectedModel,
|
||||
promptDraft,
|
||||
promptFocusRequest,
|
||||
setApprovalMode,
|
||||
setPromptDraft,
|
||||
setSelectedModel,
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
import { act, cleanup, renderHook } from "@testing-library/react";
|
||||
import type { Map as MapLibreMap } from "maplibre-gl";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { searchPlaces, type PlaceSearchItem, type PlaceSearchResults } from "../api/place-search-client";
|
||||
import { locateGeocodingResult, markGeocodingLocation } from "../map/geocoding-location";
|
||||
import { useWorkbenchGeocoding } from "./use-workbench-geocoding";
|
||||
|
||||
vi.mock("../api/place-search-client", () => ({ searchPlaces: vi.fn() }));
|
||||
vi.mock("../map/geocoding-location", () => ({ locateGeocodingResult: vi.fn(), markGeocodingLocation: vi.fn() }));
|
||||
const location: PlaceSearchItem = { id: "place-1", name: "南桥路", address: "奉贤区南桥路", longitude: 121.44, latitude: 31.22, category: "道路名", district: "奉贤区" };
|
||||
const other = { ...location, id: "place-2", address: "奉贤区南桥路", longitude: 121.35, latitude: 30.77 };
|
||||
const results: PlaceSearchResults = { items: [location, other], total: 2, truncated: false, region: "奉贤区" };
|
||||
const remove = vi.fn();
|
||||
const options = {
|
||||
active: true, projectId: "project-1", mapRef: { current: {} as MapLibreMap }, mapReady: true,
|
||||
getPadding: () => ({ top: 48, right: 400, bottom: 48, left: 48 }), reducedMotion: true, onLocate: vi.fn()
|
||||
};
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
vi.mocked(markGeocodingLocation).mockReturnValue({ remove } as unknown as ReturnType<typeof markGeocodingLocation>);
|
||||
});
|
||||
afterEach(cleanup);
|
||||
|
||||
describe("place search selection lifecycle", () => {
|
||||
it("does not move until selection and replaces the previous marker", async () => {
|
||||
vi.mocked(searchPlaces).mockResolvedValue(results);
|
||||
const { result, rerender } = renderHook(useWorkbenchGeocoding, { initialProps: options });
|
||||
act(() => result.current.changeAddress("南桥路"));
|
||||
await act(() => result.current.search());
|
||||
expect(result.current.results).toEqual(results);
|
||||
expect(result.current.result).toBeNull();
|
||||
expect(result.current.choosing).toBe(true);
|
||||
expect(markGeocodingLocation).not.toHaveBeenCalled();
|
||||
act(() => result.current.select(location.id));
|
||||
expect(result.current.choosing).toBe(false);
|
||||
expect(locateGeocodingResult).toHaveBeenCalledWith(options.mapRef.current, location, options.getPadding(), true);
|
||||
act(() => result.current.select(other.id));
|
||||
expect(result.current.result).toEqual(other);
|
||||
expect(remove).toHaveBeenCalledOnce();
|
||||
expect(markGeocodingLocation).toHaveBeenLastCalledWith(options.mapRef.current, other);
|
||||
act(() => result.current.changeSelection());
|
||||
expect(result.current.choosing).toBe(true);
|
||||
expect(result.current.result).toEqual(other);
|
||||
const moves = vi.mocked(locateGeocodingResult).mock.calls.length;
|
||||
act(() => result.current.select(other.id));
|
||||
expect(result.current.choosing).toBe(false);
|
||||
expect(locateGeocodingResult).toHaveBeenCalledTimes(moves + 1);
|
||||
rerender({ ...options, active: false });
|
||||
expect(result.current.result).toEqual(other);
|
||||
act(() => result.current.clear());
|
||||
expect(remove).toHaveBeenCalledTimes(2);
|
||||
expect(result.current.results).toBeNull();
|
||||
});
|
||||
it.each(["edit", "close", "project"])("ignores a late search after %s", async (action) => {
|
||||
let resolve!: (value: PlaceSearchResults) => void;
|
||||
vi.mocked(searchPlaces).mockReturnValue(new Promise((done) => { resolve = done; }));
|
||||
const { result, rerender } = renderHook(useWorkbenchGeocoding, { initialProps: options });
|
||||
act(() => result.current.changeAddress("旧地址"));
|
||||
let pending!: Promise<void>;
|
||||
act(() => { pending = result.current.search(); });
|
||||
const signal = vi.mocked(searchPlaces).mock.calls[0][3];
|
||||
if (action === "edit") act(() => result.current.changeAddress("新地址"));
|
||||
if (action === "close") rerender({ ...options, active: false });
|
||||
if (action === "project") rerender({ ...options, projectId: "project-2" });
|
||||
expect(signal?.aborted).toBe(true);
|
||||
await act(async () => { resolve(results); await pending; });
|
||||
expect(result.current.results).toBeNull();
|
||||
expect(result.current.result).toBeNull();
|
||||
expect(result.current.pending).toBe(false);
|
||||
});
|
||||
it("recovers from errors and clears the map when editing the query", async () => {
|
||||
vi.mocked(searchPlaces).mockRejectedValueOnce(new Error("查询暂不可用")).mockResolvedValueOnce(results);
|
||||
const { result, unmount } = renderHook(() => useWorkbenchGeocoding(options));
|
||||
act(() => result.current.changeAddress("南桥路"));
|
||||
await act(() => result.current.search());
|
||||
expect(result.current.error).toBe("查询暂不可用");
|
||||
await act(() => result.current.search());
|
||||
expect(result.current.error).toBeNull();
|
||||
act(() => result.current.select(location.id));
|
||||
act(() => result.current.changeAddress("新地址"));
|
||||
expect(remove).toHaveBeenCalledOnce();
|
||||
expect(result.current.result).toBeNull();
|
||||
expect(result.current.results).toBeNull();
|
||||
unmount();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,104 @@
|
||||
import type { Map as MapLibreMap, PaddingOptions } from "maplibre-gl";
|
||||
import { useCallback, useEffect, useRef, useState, type RefObject } from "react";
|
||||
import type { AccessTokenProvider } from "@/shared/auth/keycloak-auth";
|
||||
import { searchPlaces, type PlaceSearchItem, type PlaceSearchResults } from "../api/place-search-client";
|
||||
import { locateGeocodingResult, markGeocodingLocation } from "../map/geocoding-location";
|
||||
|
||||
type GeocodingOptions = {
|
||||
active: boolean;
|
||||
projectId: string | null;
|
||||
getAccessToken?: AccessTokenProvider;
|
||||
mapRef: RefObject<MapLibreMap | null>;
|
||||
mapReady: boolean;
|
||||
getPadding: () => PaddingOptions;
|
||||
reducedMotion: boolean;
|
||||
onLocate: () => void;
|
||||
};
|
||||
|
||||
export function useWorkbenchGeocoding(options: GeocodingOptions) {
|
||||
const [address, setAddress] = useState("");
|
||||
const [result, setResult] = useState<PlaceSearchItem | null>(null);
|
||||
const [results, setResults] = useState<PlaceSearchResults | null>(null);
|
||||
const [choosing, setChoosing] = useState(true);
|
||||
const [pending, setPending] = useState(false);
|
||||
const [error, setError] = useState<string | null>(null);
|
||||
const requestRef = useRef<AbortController | null>(null);
|
||||
const optionsRef = useRef(options);
|
||||
optionsRef.current = options;
|
||||
|
||||
const cancel = useCallback(() => {
|
||||
requestRef.current?.abort();
|
||||
requestRef.current = null;
|
||||
setPending(false);
|
||||
}, []);
|
||||
const clear = useCallback(() => {
|
||||
cancel();
|
||||
setResult(null);
|
||||
setResults(null);
|
||||
setChoosing(true);
|
||||
setError(null);
|
||||
}, [cancel]);
|
||||
useEffect(() => {
|
||||
clear();
|
||||
setAddress("");
|
||||
return () => requestRef.current?.abort();
|
||||
}, [options.projectId, options.getAccessToken, clear]);
|
||||
useEffect(() => {
|
||||
if (!options.active) cancel();
|
||||
}, [options.active, cancel]);
|
||||
|
||||
const locate = useCallback(() => {
|
||||
const current = optionsRef.current;
|
||||
const map = current.mapRef.current;
|
||||
if (!result || !current.mapReady || !map) return;
|
||||
current.onLocate();
|
||||
locateGeocodingResult(map, result, current.getPadding(), current.reducedMotion);
|
||||
}, [result]);
|
||||
|
||||
useEffect(() => {
|
||||
const map = options.mapRef.current;
|
||||
if (!map || !options.mapReady || !result) return;
|
||||
const marker = markGeocodingLocation(map, result);
|
||||
locate();
|
||||
return () => { marker.remove(); };
|
||||
}, [options.mapRef, options.mapReady, result, locate]);
|
||||
|
||||
async function search() {
|
||||
clear();
|
||||
if (!address.trim()) { setError("请输入需要搜索的地址或地名"); return; }
|
||||
const controller = new AbortController();
|
||||
requestRef.current = controller;
|
||||
setPending(true);
|
||||
try {
|
||||
const value = await searchPlaces(address, options.projectId, options.getAccessToken, controller.signal);
|
||||
if (controller.signal.aborted || requestRef.current !== controller) return;
|
||||
setResults(value);
|
||||
} catch (failure) {
|
||||
if (controller.signal.aborted || requestRef.current !== controller) return;
|
||||
setError(failure instanceof Error && failure.name === "TimeoutError"
|
||||
? "地名搜索超时，请稍后重试"
|
||||
: failure instanceof Error ? failure.message : "地名搜索失败，请稍后重试");
|
||||
} finally {
|
||||
if (requestRef.current === controller) { requestRef.current = null; setPending(false); }
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
address, result, results, choosing, pending, error, locate, search,
|
||||
select: (id: string) => {
|
||||
if (pending) return;
|
||||
const item = results?.items.find((candidate) => candidate.id === id);
|
||||
if (item) {
|
||||
if (result?.id === id) locate();
|
||||
else setResult(item);
|
||||
setChoosing(false);
|
||||
}
|
||||
},
|
||||
changeSelection: () => setChoosing(true),
|
||||
enabled: options.mapReady,
|
||||
changeAddress: (value: string) => { clear(); setAddress(value); },
|
||||
clear: () => { clear(); setAddress(""); }
|
||||
};
|
||||
}
|
||||
|
||||
export type WorkbenchGeocoding = ReturnType<typeof useWorkbenchGeocoding>;
|
||||
@@ -0,0 +1,77 @@
|
||||
import { act, cleanup, renderHook } from "@testing-library/react";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { dismissMapNotice, showMapNotice } from "@/features/map/core";
|
||||
import { useWorkbenchSyncNotice } from "./use-workbench-sync-notice";
|
||||
|
||||
vi.mock("@/features/map/core", () => ({
|
||||
dismissMapNotice: vi.fn(),
|
||||
showMapNotice: vi.fn()
|
||||
}));
|
||||
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers();
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
cleanup();
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
describe("workbench reconnect notice", () => {
|
||||
it("keeps repeated brief reconnects silent without leaving a pending warning", () => {
|
||||
const { rerender } = renderHook((reconnecting) => useWorkbenchSyncNotice(reconnecting), {
|
||||
initialProps: false
|
||||
});
|
||||
|
||||
for (let cycle = 0; cycle < 3; cycle += 1) {
|
||||
rerender(true);
|
||||
act(() => vi.advanceTimersByTime(1_000));
|
||||
rerender(false);
|
||||
act(() => vi.advanceTimersByTime(300_000));
|
||||
}
|
||||
|
||||
expect(showMapNotice).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("warns once after a sustained disconnect and dismisses on recovery", () => {
|
||||
const { rerender } = renderHook((reconnecting) => useWorkbenchSyncNotice(reconnecting), {
|
||||
initialProps: true
|
||||
});
|
||||
act(() => vi.advanceTimersByTime(14_999));
|
||||
expect(showMapNotice).not.toHaveBeenCalled();
|
||||
// Snapshot refreshes and failed retries must not restart the waiting period.
|
||||
rerender(true);
|
||||
act(() => vi.advanceTimersByTime(1));
|
||||
expect(showMapNotice).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({
|
||||
id: "workbench-sync-reconnecting", title: "业务同步正在重连", duration: Infinity
|
||||
}));
|
||||
act(() => vi.advanceTimersByTime(60_000));
|
||||
expect(showMapNotice).toHaveBeenCalledOnce();
|
||||
|
||||
vi.mocked(dismissMapNotice).mockClear();
|
||||
rerender(false);
|
||||
expect(dismissMapNotice).toHaveBeenCalledWith("workbench-sync-reconnecting");
|
||||
});
|
||||
|
||||
it("starts a fresh waiting period for a new outage", () => {
|
||||
const { rerender } = renderHook((reconnecting) => useWorkbenchSyncNotice(reconnecting), {
|
||||
initialProps: true
|
||||
});
|
||||
act(() => vi.advanceTimersByTime(10_000));
|
||||
rerender(false);
|
||||
rerender(true);
|
||||
act(() => vi.advanceTimersByTime(5_000));
|
||||
expect(showMapNotice).not.toHaveBeenCalled();
|
||||
act(() => vi.advanceTimersByTime(10_000));
|
||||
expect(showMapNotice).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("cancels a pending warning when leaving the workbench", () => {
|
||||
const { unmount } = renderHook(() => useWorkbenchSyncNotice(true));
|
||||
act(() => vi.advanceTimersByTime(10_000));
|
||||
unmount();
|
||||
act(() => vi.advanceTimersByTime(20_000));
|
||||
expect(showMapNotice).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,30 @@
|
||||
import { useEffect } from "react";
|
||||
import { dismissMapNotice, showMapNotice } from "@/features/map/core";
|
||||
|
||||
const RECONNECT_NOTICE_DELAY_MS = 15_000;
|
||||
const RECONNECT_NOTICE_ID = "workbench-sync-reconnecting";
|
||||
|
||||
export function useWorkbenchSyncNotice(reconnecting: boolean) {
|
||||
useEffect(() => {
|
||||
if (!reconnecting) {
|
||||
dismissMapNotice(RECONNECT_NOTICE_ID);
|
||||
return;
|
||||
}
|
||||
|
||||
// Scheduled SSE renewal and brief outages normally recover within this window.
|
||||
const timer = window.setTimeout(() => {
|
||||
showMapNotice({
|
||||
id: RECONNECT_NOTICE_ID,
|
||||
tone: "warning",
|
||||
title: "业务同步正在重连",
|
||||
message: "当前数据可能陈旧，连接恢复后将自动更新。",
|
||||
duration: Infinity
|
||||
});
|
||||
}, RECONNECT_NOTICE_DELAY_MS);
|
||||
|
||||
return () => {
|
||||
window.clearTimeout(timer);
|
||||
dismissMapNotice(RECONNECT_NOTICE_ID);
|
||||
};
|
||||
}, [reconnecting]);
|
||||
}
|
||||
@@ -6,6 +6,7 @@ import {
|
||||
type WorkbenchEvent,
|
||||
type WorkbenchSyncSnapshot
|
||||
} from "../api/workbench-sync-client";
|
||||
import { WorkbenchEventChannel } from "./workbench-event-channel";
|
||||
|
||||
const RECONCILE_INTERVAL_MS = 60_000;
|
||||
const RUNTIME_TOPICS = new Set<WorkbenchEvent["topic"]>([
|
||||
@@ -30,6 +31,7 @@ export type WorkbenchSyncState = {
|
||||
detailResetGeneration: number;
|
||||
acknowledgeEventsThrough: (eventSequence: number) => void;
|
||||
refresh: () => Promise<void>;
|
||||
subscribePresentationEvents: WorkbenchEventChannel["subscribe"];
|
||||
};
|
||||
|
||||
export function shouldApplyWorkbenchSnapshot(input: {
|
||||
@@ -44,6 +46,7 @@ export function shouldApplyWorkbenchSnapshot(input: {
|
||||
}
|
||||
|
||||
export function useWorkbenchSync(getAccessToken?: AccessTokenProvider): WorkbenchSyncState {
|
||||
const channelRef = useRef(new WorkbenchEventChannel());
|
||||
const [snapshot, setSnapshot] = useState<WorkbenchSyncSnapshot | null>(null);
|
||||
const [events, setEvents] = useState<WorkbenchEvent[]>([]);
|
||||
const [status, setStatus] = useState<WorkbenchSyncState["status"]>("connecting");
|
||||
@@ -63,6 +66,7 @@ export function useWorkbenchSync(getAccessToken?: AccessTokenProvider): Workbenc
|
||||
}, []);
|
||||
|
||||
useEffect(() => {
|
||||
const channel = channelRef.current;
|
||||
const controller = new AbortController();
|
||||
let reconnectTimer: number | undefined;
|
||||
let reconcileTimer: number | undefined;
|
||||
@@ -135,13 +139,18 @@ export function useWorkbenchSync(getAccessToken?: AccessTokenProvider): Workbenc
|
||||
getAccessToken,
|
||||
signal: controller.signal,
|
||||
onOpen: () => {
|
||||
channel.publish({ type: "open" });
|
||||
statusRef.current = "live";
|
||||
setStatus("live");
|
||||
setError(null);
|
||||
retryMs = 1_000;
|
||||
},
|
||||
onReset: async () => { await loadSnapshot(true, true); },
|
||||
onReset: async () => {
|
||||
channel.publish({ type: "reset" });
|
||||
await loadSnapshot(true, true);
|
||||
},
|
||||
onEvents: (nextEvents) => {
|
||||
channel.publish({ type: "events", events: nextEvents });
|
||||
const unseen = nextEvents.filter((event) => event.event_seq > cursorRef.current);
|
||||
if (unseen.length === 0) return;
|
||||
cursorRef.current = unseen.at(-1)!.event_seq;
|
||||
@@ -155,11 +164,13 @@ export function useWorkbenchSync(getAccessToken?: AccessTokenProvider): Workbenc
|
||||
}
|
||||
});
|
||||
if (!controller.signal.aborted) {
|
||||
channel.publish({ type: "disconnect" });
|
||||
statusRef.current = "reconnecting";
|
||||
setStatus("reconnecting");
|
||||
}
|
||||
} catch (nextError) {
|
||||
if (controller.signal.aborted) return;
|
||||
channel.publish({ type: "disconnect" });
|
||||
statusRef.current = "reconnecting";
|
||||
setStatus("reconnecting");
|
||||
setError(nextError instanceof Error ? nextError.message : "工作台事件连接中断");
|
||||
@@ -175,6 +186,7 @@ export function useWorkbenchSync(getAccessToken?: AccessTokenProvider): Workbenc
|
||||
if (statusRef.current !== "live") void loadSnapshot().catch(() => undefined);
|
||||
}, RECONCILE_INTERVAL_MS);
|
||||
return () => {
|
||||
channel.publish({ type: "disconnect" });
|
||||
controller.abort();
|
||||
refreshSnapshotRef.current = null;
|
||||
if (reconnectTimer !== undefined) window.clearTimeout(reconnectTimer);
|
||||
@@ -190,6 +202,7 @@ export function useWorkbenchSync(getAccessToken?: AccessTokenProvider): Workbenc
|
||||
error,
|
||||
detailResetGeneration,
|
||||
acknowledgeEventsThrough,
|
||||
refresh
|
||||
refresh,
|
||||
subscribePresentationEvents: channelRef.current.subscribe
|
||||
};
|
||||
}
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import type { WorkbenchEvent } from "../api/workbench-sync-client";
|
||||
import { WorkbenchEventChannel, type WorkbenchStreamNotification } from "./workbench-event-channel";
|
||||
|
||||
describe("workbench wire event channel", () => {
|
||||
it("keeps independent observers live across business consumption, another observer failure and unsubscribe", () => {
|
||||
const channel = new WorkbenchEventChannel();
|
||||
const first = vi.fn();
|
||||
const unsubscribeFirst = channel.subscribe(first);
|
||||
expect(first).toHaveBeenCalledWith({ type: "disconnect" });
|
||||
channel.publish({ type: "open" });
|
||||
channel.subscribe((notification) => { if (notification.type === "events") throw new Error("observer failed"); });
|
||||
const second = vi.fn();
|
||||
channel.subscribe(second);
|
||||
expect(second).toHaveBeenCalledWith({ type: "open" });
|
||||
const events = [{ event_seq: 11 }, { event_seq: 12 }] as WorkbenchEvent[];
|
||||
const businessPending = [...events];
|
||||
const notification: WorkbenchStreamNotification = { type: "events", events };
|
||||
channel.publish(notification);
|
||||
businessPending.splice(0, businessPending.length);
|
||||
expect(first).toHaveBeenLastCalledWith(notification);
|
||||
expect(second).toHaveBeenLastCalledWith(notification);
|
||||
expect(events.map((event) => event.event_seq)).toEqual([11, 12]);
|
||||
const firstCallCount = first.mock.calls.length;
|
||||
unsubscribeFirst();
|
||||
const next: WorkbenchStreamNotification = { type: "events", events: [{ event_seq: 13 }] as WorkbenchEvent[] };
|
||||
channel.publish(next);
|
||||
expect(first).toHaveBeenCalledTimes(firstCallCount);
|
||||
expect(second).toHaveBeenLastCalledWith(next);
|
||||
channel.publish({ type: "disconnect" });
|
||||
expect(second).toHaveBeenLastCalledWith({ type: "disconnect" });
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,25 @@
|
||||
import type { WorkbenchEvent } from "../api/workbench-sync-client";
|
||||
|
||||
export type WorkbenchStreamNotification =
|
||||
| { type: "open" | "disconnect" | "reset" }
|
||||
| { type: "events"; events: WorkbenchEvent[] };
|
||||
|
||||
/** Presentation observes wire events before business acknowledgements or snapshot filtering. */
|
||||
export class WorkbenchEventChannel {
|
||||
private listeners = new Set<(notification: WorkbenchStreamNotification) => void>();
|
||||
private live = false;
|
||||
|
||||
subscribe = (listener: (notification: WorkbenchStreamNotification) => void) => {
|
||||
this.listeners.add(listener);
|
||||
listener({ type: this.live ? "open" : "disconnect" });
|
||||
return () => { this.listeners.delete(listener); };
|
||||
};
|
||||
|
||||
publish(notification: WorkbenchStreamNotification) {
|
||||
if (notification.type !== "events") this.live = notification.type === "open";
|
||||
for (const listener of this.listeners) {
|
||||
// One presentation observer must not interrupt authoritative business synchronization.
|
||||
try { listener(notification); } catch { /* Observers own their error presentation. */ }
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user