feat(workbench): refine dynamic task collaboration

2026-09-24 16:36:12 +08:00
parent b64d2b78bb
commit 87cc7785da
25 changed files with 1216 additions and 470 deletions
@@ -52,6 +52,7 @@ function task(taskType: OperationalTask["task_type"]): OperationalTask {
    latest_report: null,
    evidence: [],
    actions: [],
+    access_requests: [],
    permissions: {
      can_publish_artifacts: false,
      can_propose_actions: false,
@@ -59,7 +60,8 @@ function task(taskType: OperationalTask["task_type"]): OperationalTask {
      can_decide_work_orders: false,
      can_execute_work_orders: false,
      can_decide_commands: false,
-      can_execute_commands: false
+      can_execute_commands: false,
+      can_decide_access: false
    },
    work_order_snapshot: {
      work_order_id: "44444444-4444-4444-8444-444444444444",
@@ -5,6 +5,9 @@ import {
  createOperationalRequestId,
  createOperationalTask,
  decideOperationalTaskAction,
+  decideOperationalTaskAccess,
+  updateOperationalWorkOrderDraft,
+  confirmOperationalWorkOrderDispatch,
  executeOperationalTaskAction,
  fetchOperationalTask,
  fetchOperationalTaskEvidenceSnapshot,
@@ -72,6 +75,7 @@ export function useOperationalTasks(
  const [loading, setLoading] = useState(true);
  const [refreshGeneration, setRefreshGeneration] = useState(0);
  const forceRefreshRef = useRef(false);
+  const detailResetGenerationRef = useRef(sync.detailResetGeneration);

  const refresh = useCallback(() => {
    forceRefreshRef.current = true;
@@ -84,7 +88,9 @@ export function useOperationalTasks(
      const summaries = sync.snapshot?.tasks ?? [];
      try {
        const current = new Map(tasksRef.current.map((task) => [task.task_id, task]));
-        const changed = forceRefreshRef.current
+        const resetTaskDetails = forceRefreshRef.current
+          || detailResetGenerationRef.current !== sync.detailResetGeneration;
+        const changed = resetTaskDetails
          ? summaries
          : summaries.filter((summary) => current.get(summary.task_id)?.revision !== summary.revision);
        const details = await Promise.all(changed.map((task) =>
@@ -98,8 +104,11 @@ export function useOperationalTasks(
        });
        tasksRef.current = nextTasks;
        forceRefreshRef.current = false;
+        detailResetGenerationRef.current = sync.detailResetGeneration;
        setTasks(nextTasks);
-        setNotifications((sync.snapshot?.notifications ?? []).filter((item) => item.status === "unread"));
+        setNotifications((sync.snapshot?.notifications ?? []).filter(
+          (item) => item.lifecycle_status === "active" && item.status === "unread"
+        ));
        setAvailable(true);
        activatedRef.current = true;
        setError(null);
@@ -118,7 +127,7 @@ export function useOperationalTasks(
    return () => {
      controller.abort();
    };
-  }, [getAccessToken, refreshGeneration, sync.snapshot]);
+  }, [getAccessToken, refreshGeneration, sync.detailResetGeneration, sync.snapshot]);

  useEffect(() => {
    const events = syncEvents;
@@ -234,6 +243,45 @@ export function useOperationalTasks(
    return action;
  }, [getAccessToken]);

+  const updateWorkOrderDraft = useCallback(async (
+    taskId: string, actionId: string, expectedVersion: number, payload: Record<string, unknown>
+  ) => {
+    const action = await updateOperationalWorkOrderDraft(taskId, actionId, {
+      request_id: createOperationalRequestId(), expected_version: expectedVersion, payload
+    }, getAccessToken);
+    setTasks((current) => current.map((item) => item.task_id === taskId ? {
+      ...item, actions: item.actions.map((entry) => entry.action_id === actionId ? action : entry)
+    } : item));
+    return action;
+  }, [getAccessToken]);
+
+  const confirmWorkOrderDispatch = useCallback(async (
+    taskId: string, actionId: string, expectedVersion: number
+  ) => {
+    const action = await confirmOperationalWorkOrderDispatch(taskId, actionId, {
+      request_id: createOperationalRequestId(), expected_version: expectedVersion
+    }, getAccessToken);
+    setTasks((current) => current.map((item) => item.task_id === taskId ? {
+      ...item, actions: item.actions.map((entry) => entry.action_id === actionId ? action : entry)
+    } : item));
+    return action;
+  }, [getAccessToken]);
+
+  const decideAccess = useCallback(async (
+    taskId: string, accessRequestId: string, expectedVersion: number,
+    decision: "approved" | "rejected", approvedScopes: string[]
+  ) => {
+    const request = await decideOperationalTaskAccess(taskId, accessRequestId, {
+      request_id: createOperationalRequestId(), expected_version: expectedVersion,
+      decision, approved_scopes: approvedScopes
+    }, getAccessToken);
+    setTasks((current) => current.map((item) => item.task_id === taskId ? {
+      ...item, access_requests: item.access_requests.map((entry) => entry.access_request_id === accessRequestId ? request : entry)
+    } : item));
+    refresh();
+    return request;
+  }, [getAccessToken, refresh]);
+
  const acknowledgeNotification = useCallback(async (notificationId: string, version: number) => {
    await acknowledgeOperationalTaskNotification(
      notificationId,
@@ -261,6 +309,9 @@ export function useOperationalTasks(
    decideAction,
    executeAction,
    reconcileAction,
+    updateWorkOrderDraft,
+    confirmWorkOrderDispatch,
+    decideAccess,
    acknowledgeNotification,
    loadEvidenceSnapshot
  };
@@ -89,6 +89,7 @@ export function useWorkbenchAgent({
  const [mobilePanelCollapsing, setMobilePanelCollapsing] = useState(false);
  const [sessionId, setSessionId] = useState<string | null>(null);
  const [sessionPurpose, setSessionPurpose] = useState<AgentSessionPurpose>("ordinary_chat");
+  const [sessionTaskId, setSessionTaskId] = useState<string | null>(null);
  const [registry, setRegistry] = useState<UIRegistry | null>(null);
  const [sessionTitle, setSessionTitle] = useState("新对话");
  const [statusLabel, setStatusLabel] = useState("Agent 后端连接中");
@@ -131,6 +132,7 @@ export function useWorkbenchAgent({
  const clearSessionId = useCallback(() => {
    sessionIdRef.current = null;
    setSessionId(null);
+    setSessionTaskId(null);
  }, []);

  const resolveRenderRef = useCallback((renderRef: string, nextSessionId: string) => {
@@ -549,6 +551,7 @@ export function useWorkbenchAgent({

        applySessionId(loadedSession.sessionId);
        setSessionPurpose(loadedSession.purpose);
+        setSessionTaskId(loadedSession.taskId ?? null);
        setSessionTitle(loadedSession.title);
        setStatusLabel("Agent 历史会话已打开");
        setPermissionOverrides({});
@@ -632,6 +635,7 @@ export function useWorkbenchAgent({
      clearSessionId();
      setSessionTitle("新对话");
      setSessionPurpose("ordinary_chat");
+      setSessionTaskId(null);
      setStatusLabel(nextStatusLabel);
      setConnectionStatus("online");
      setPermissionOverrides({});
@@ -917,6 +921,7 @@ export function useWorkbenchAgent({
    sessionHistoryLoading: sessionHistoryLoading || sessionHistoryValidating,
    sessionId,
    sessionPurpose,
+    sessionTaskId,
    sessionTitle,
    statusLabel,
    streamRenderState,
@@ -0,0 +1,31 @@
+import { describe, expect, it } from "vitest";
+import { shouldApplyWorkbenchSnapshot } from "./use-workbench-sync";
+
+describe("workbench snapshot ordering", () => {
+  it("rejects an older concurrent response even when its cursor is valid", () => {
+    expect(shouldApplyWorkbenchSnapshot({
+      requestGeneration: 1,
+      latestRequestGeneration: 2,
+      eventCursor: 120,
+      currentCursor: 100
+    })).toBe(false);
+  });
+
+  it("rejects a snapshot whose cursor trails already consumed events", () => {
+    expect(shouldApplyWorkbenchSnapshot({
+      requestGeneration: 2,
+      latestRequestGeneration: 2,
+      eventCursor: 99,
+      currentCursor: 100
+    })).toBe(false);
+  });
+
+  it("accepts only the latest response without cursor rollback", () => {
+    expect(shouldApplyWorkbenchSnapshot({
+      requestGeneration: 2,
+      latestRequestGeneration: 2,
+      eventCursor: 100,
+      currentCursor: 100
+    })).toBe(true);
+  });
+});
@@ -19,26 +19,38 @@ export type WorkbenchSyncState = {
  events: WorkbenchEvent[];
  status: "connecting" | "live" | "reconnecting";
  error: string | null;
+  detailResetGeneration: number;
  acknowledgeEventsThrough: (eventSequence: number) => void;
  refresh: () => Promise<void>;
 };

+export function shouldApplyWorkbenchSnapshot(input: {
+  requestGeneration: number;
+  latestRequestGeneration: number;
+  eventCursor: number;
+  currentCursor: number;
+}) {
+  return input.requestGeneration === input.latestRequestGeneration
+    && input.eventCursor >= input.currentCursor;
+}
+
 export function useWorkbenchSync(getAccessToken?: AccessTokenProvider): WorkbenchSyncState {
  const [snapshot, setSnapshot] = useState<WorkbenchSyncSnapshot | null>(null);
  const [events, setEvents] = useState<WorkbenchEvent[]>([]);
  const [status, setStatus] = useState<WorkbenchSyncState["status"]>("connecting");
  const [error, setError] = useState<string | null>(null);
+  const [detailResetGeneration, setDetailResetGeneration] = useState(0);
  const cursorRef = useRef(0);
  const statusRef = useRef<WorkbenchSyncState["status"]>("connecting");
  const runtimeRefreshRef = useRef<number | undefined>(undefined);
-  const refreshSnapshotRef = useRef<(() => Promise<void>) | null>(null);
+  const refreshSnapshotRef = useRef<((resetTaskDetails?: boolean) => Promise<void>) | null>(null);
  const acknowledgeEventsThrough = useCallback((eventSequence: number) => {
    setEvents((current) => current.filter((event) => event.event_seq > eventSequence));
  }, []);
  const refresh = useCallback(async () => {
    const refreshSnapshot = refreshSnapshotRef.current;
    if (!refreshSnapshot) throw new Error("工作台同步尚未就绪");
-    await refreshSnapshot();
+    await refreshSnapshot(true);
  }, []);

  useEffect(() => {
@@ -47,12 +59,24 @@ export function useWorkbenchSync(getAccessToken?: AccessTokenProvider): Workbenc
    let reconcileTimer: number | undefined;
    let retryMs = 1_000;
    let initialized = false;
+    let snapshotRequestGeneration = 0;

-    const loadSnapshot = async () => {
+    const loadSnapshot = async (resetTaskDetails = false) => {
+      const requestGeneration = ++snapshotRequestGeneration;
      const next = await fetchWorkbenchSyncSnapshot(getAccessToken, controller.signal);
-      if (controller.signal.aborted) return;
+      if (
+        controller.signal.aborted
+        || !shouldApplyWorkbenchSnapshot({
+          requestGeneration,
+          latestRequestGeneration: snapshotRequestGeneration,
+          eventCursor: next.eventCursor,
+          currentCursor: cursorRef.current
+        })
+      ) return;
      cursorRef.current = next.eventCursor;
+      if (resetTaskDetails) setEvents([]);
      setSnapshot(next);
+      if (resetTaskDetails) setDetailResetGeneration((current) => current + 1);
      setError(null);
      initialized = true;
    };
@@ -69,7 +93,7 @@ export function useWorkbenchSync(getAccessToken?: AccessTokenProvider): Workbenc
          after: cursorRef.current,
          getAccessToken,
          signal: controller.signal,
-          onReset: loadSnapshot,
+          onReset: () => loadSnapshot(true),
          onEvents: (nextEvents) => {
            const unseen = nextEvents.filter((event) => event.event_seq > cursorRef.current);
            if (unseen.length === 0) return;
@@ -116,5 +140,13 @@ export function useWorkbenchSync(getAccessToken?: AccessTokenProvider): Workbenc
    };
  }, [getAccessToken]);

-  return { snapshot, events, status, error, acknowledgeEventsThrough, refresh };
+  return {
+    snapshot,
+    events,
+    status,
+    error,
+    detailResetGeneration,
+    acknowledgeEventsThrough,
+    refresh
+  };
 }