Files
deerflow-code/frontend-web/src/position-roundtable/api/sessions.ts
T
董宏凯andCursor 75963c9ad9 feat: 岗位会商命令幂等与圆桌后台作业租约调度
岗位会商写操作引入 command_id 幂等与版本锁,前端支持同键重试;圆桌后台作业改为 DB 租约 dispatcher、去重与草稿 outbox 补偿,并持久化 intent turns。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-14 11:08:12 +08:00

737 lines
25 KiB
TypeScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
/** Client for the independent position-collaboration Session / Node API.
*
* Messages themselves stay in LangGraph threads. These endpoints only persist
* task/intent snapshots, frozen chain topology, node states and indexes that
* let the UI recover the correct thread after a refresh.
*/
import { apiFetch } from "@/core/api/fetch-client";
import { getBackendBaseURL } from "@/core/config";
/**
* 节点状态由岗位会商服务推进;前端只据此决定可否发送、展示进度,
* 不应在组件内自行把节点改成 done 或 ready。
*/
export type PositionNodeStatus =
| "locked"
| "ready"
| "running"
| "done"
| "stale"
| "rejected"
| "error";
/**
* SQL-owned recovery copy for a chat surface. LangGraph remains the execution
* source, but this snapshot keeps history and an unsent composer draft
* recoverable when a deployment replaces the checkpoint or sandbox volume.
*/
export interface PositionConversationSnapshot {
messages: Array<Record<string, unknown>>;
timelineEntries?: Array<Record<string, unknown>>;
inputDraft?: string;
savedAt?: string;
}
/**
* 冻结业务链中的一个席位。
*
* `threadId` 指向真实的 LangGraph 对话;节点表只承担“席位 ↔ 线程”索引、
* 上游版本与交付清单的职责,绝不保存完整聊天消息。
*/
export interface PositionRoundtableNode {
id: string;
sessionId: string;
nodeKey: string;
stageIndex: number;
seatIndex: number;
agentId: string;
positionId: string | null;
threadId: string | null;
status: PositionNodeStatus;
latestAnswer: string | null;
artifactManifest: Array<Record<string, unknown>>;
conversationSnapshot: PositionConversationSnapshot | null;
revision: number;
upstreamRevision: Record<string, unknown> | null;
rejectionCount: number;
lastRejection: Record<string, unknown> | null;
invalidatedBy: string | null;
createdAt?: string;
updatedAt?: string;
}
/**
* 岗位会商会话的可恢复快照。
*
* 任务、意图与链条在启动后被分别冻结,避免外部任务或链条配置发生变化时
* 改写已经开始的协同过程。`nodes` 只在详情接口中携带。
*/
export interface PositionRoundtableSession {
id: string;
/** Present only when the workspace was opened with a route `taskId`. */
externalTaskId: string | null;
/** True only for sessions intentionally shared by an explicit route taskId. */
taskScoped: boolean;
taskSnapshot: Record<string, unknown>;
chainId: string | null;
chainSnapshot: Record<string, unknown> | null;
intentThreadId: string | null;
intentSnapshot: Record<string, unknown> | null;
intentConversation: PositionConversationSnapshot | null;
summaryThreadId: string | null;
summarySnapshot: Record<string, unknown> | null;
summaryConversation: PositionConversationSnapshot | null;
actionPlanThreadId: string | null;
actionPlanSnapshot: Record<string, unknown> | null;
actionPlanConversation: PositionConversationSnapshot | null;
status: "intent_pending" | "active" | "completed" | "archived";
createdAt?: string;
updatedAt?: string;
nodes?: PositionRoundtableNode[];
}
interface ServerNode {
id: string;
session_id: string;
node_key: string;
stage_index: number;
seat_index: number;
agent_id: string;
position_id: string | null;
thread_id: string | null;
status: PositionNodeStatus;
latest_answer: string | null;
artifact_manifest?: Array<Record<string, unknown>>;
conversation_snapshot?: PositionConversationSnapshot | null;
revision?: number;
upstream_revision?: Record<string, unknown> | null;
rejection_count?: number;
last_rejection?: Record<string, unknown> | null;
invalidated_by?: string | null;
created_at?: string;
updated_at?: string;
}
interface ServerSession {
id: string;
external_task_id: string | null;
task_scoped?: boolean;
task_snapshot?: Record<string, unknown>;
chain_id: string | null;
chain_snapshot?: Record<string, unknown> | null;
intent_thread_id: string | null;
intent_snapshot?: Record<string, unknown> | null;
intent_conversation?: PositionConversationSnapshot | null;
summary_thread_id?: string | null;
summary_snapshot?: Record<string, unknown> | null;
summary_conversation?: PositionConversationSnapshot | null;
action_plan_thread_id?: string | null;
action_plan_snapshot?: Record<string, unknown> | null;
action_plan_conversation?: PositionConversationSnapshot | null;
status: PositionRoundtableSession["status"];
created_at?: string;
updated_at?: string;
nodes?: ServerNode[];
}
const PRIMARY_API_PATH = "/api/position-roundtable";
const COMPAT_API_PATH = "/api/multi-agent/position-roundtable";
const BASE = () => `${getBackendBaseURL()}${PRIMARY_API_PATH}`;
/**
* 生成一次写操作的幂等键(command_id)。
*
* 后端命令式状态机用 ``UNIQUE(session_id, node_key, command_id)`` 保证同一 command_id
* 只执行一次。把它在**一次调用入口**生成并固定进请求体,使得 `positionApiFetch` 的
* 主/兼容路由回退重试复用同一 command_id —— 网络层重发绝不会把同一操作执行两遍
* (不会重复 +revision / 重复解锁下游 / 重复驳回)。每次新的用户操作则取新键。
*/
function newCommandId(): string {
if (typeof crypto !== "undefined" && typeof crypto.randomUUID === "function") {
return crypto.randomUUID().replace(/-/g, "");
}
return `${Date.now().toString(36)}${Math.random().toString(36).slice(2, 12)}`;
}
/**
* 用**同一个 command_id** 重试一个写操作至多 ``attempts`` 次(指数退避)。
*
* 后端命令幂等表保证同 command_id + 同参数只执行一次:即便第一次其实已提交成功、
* 只是响应在网络层丢失,重试也只会重放上次结果,而不会重复推进状态机(重复
* +revision / 重复解锁下游)。因此「提交产物给下一阶段」这类关键写操作可以放心
* 重试,把偶发网络错误/超时变成自愈,而不是让下一阶段永远拿不到上游产物。
*
* 返回 ``{ ok, result?, error? }``;``ok=false`` 时 ``error`` 是最后一次失败。
*/
export async function retryWithSameCommandId<T>(
operation: (commandId: string) => Promise<T>,
options?: { attempts?: number; baseDelayMs?: number },
): Promise<{ ok: true; result: T } | { ok: false; error: unknown }> {
const attempts = Math.max(1, options?.attempts ?? 3);
const baseDelay = options?.baseDelayMs ?? 400;
const commandId = newCommandId();
let lastError: unknown = null;
for (let attempt = 0; attempt < attempts; attempt += 1) {
try {
return { ok: true, result: await operation(commandId) };
} catch (error) {
lastError = error;
if (attempt < attempts - 1) {
await new Promise((resolve) => setTimeout(resolve, baseDelay * 2 ** attempt));
}
}
}
return { ok: false, error: lastError };
}
/**
* Some intranet gateways only forward the long-established `/api/multi-agent`
* namespace. The position workspace was added later under its own top-level
* prefix, so a frontend-only release can work locally but receive FastAPI's
* route-level `{"detail":"Not Found"}` after deployment.
*
* Retry only that exact "route is absent" response against the compatibility
* namespace. A domain 404 such as "Position roundtable session not found"
* remains a real 404 and is never hidden by this routing fallback.
*/
let resolvedApiPath: typeof PRIMARY_API_PATH | typeof COMPAT_API_PATH | null = null;
function replaceApiPath(url: string, apiPath: string): string {
return url.replace(PRIMARY_API_PATH, apiPath);
}
async function isMissingRoute(response: Response): Promise<boolean> {
if (response.status !== 404) return false;
const raw = await response.clone().text().catch(() => "");
try {
const parsed = JSON.parse(raw) as { detail?: unknown };
return parsed?.detail === "Not Found";
} catch {
const normalized = raw.trim();
return (
normalized === "Not Found" ||
/<title>\s*404 Not Found\s*<\/title>/i.test(normalized)
);
}
}
async function positionApiFetch(
input: string,
init?: RequestInit,
): Promise<Response> {
if (resolvedApiPath === COMPAT_API_PATH) {
const compatibilityResponse = await apiFetch(
replaceApiPath(input, COMPAT_API_PATH),
init,
);
if (!(await isMissingRoute(compatibilityResponse))) {
return compatibilityResponse;
}
// A gateway may be reloaded while the SPA remains open. Re-probe the
// canonical route instead of pinning the tab to a now-stale route choice.
resolvedApiPath = null;
}
const primaryResponse = await apiFetch(input, init);
if (!(await isMissingRoute(primaryResponse))) {
resolvedApiPath = PRIMARY_API_PATH;
return primaryResponse;
}
const compatibilityResponse = await apiFetch(
replaceApiPath(input, COMPAT_API_PATH),
init,
);
if (!(await isMissingRoute(compatibilityResponse))) {
resolvedApiPath = COMPAT_API_PATH;
return compatibilityResponse;
}
// Preserve the canonical response/error when neither route exists. This
// keeps diagnostics stable and identifies an outdated backend release.
return primaryResponse;
}
async function readJson<T>(response: Response, label: string): Promise<T> {
if (!response.ok) {
if (await isMissingRoute(response)) {
throw new Error(
`${label} failed (${response.status}): 岗位会商会话接口尚未部署或未被网关转发;` +
`已尝试 ${PRIMARY_API_PATH} 与 ${COMPAT_API_PATH}`,
);
}
const detail = await response.text().catch(() => "");
throw new Error(`${label} failed (${response.status}): ${detail}`);
}
return response.json() as Promise<T>;
}
/**
* A missing Session is a recoverable navigation state, not a generic gateway
* failure. It commonly occurs when another tab (or the history menu) deletes
* the Session that is still present in the current URL.
*/
export class PositionSessionNotFoundError extends Error {
readonly sessionId: string;
constructor(sessionId: string) {
super(`Position roundtable session not found: ${sessionId}`);
this.name = "PositionSessionNotFoundError";
this.sessionId = sessionId;
}
}
export function isPositionSessionNotFoundError(
error: unknown,
): error is PositionSessionNotFoundError {
return error instanceof PositionSessionNotFoundError;
}
/**
* API 使用 snake_case,页面使用 camelCase。所有字段映射集中在这里,
* 组件禁止直接依赖后端字段名,避免 API 演进时遗漏状态字段。
*/
function toNode(node: ServerNode): PositionRoundtableNode {
return {
id: node.id,
sessionId: node.session_id,
nodeKey: node.node_key,
stageIndex: node.stage_index,
seatIndex: node.seat_index,
agentId: node.agent_id,
positionId: node.position_id,
threadId: node.thread_id,
status: node.status,
latestAnswer: node.latest_answer,
artifactManifest: node.artifact_manifest ?? [],
conversationSnapshot: node.conversation_snapshot ?? null,
revision: node.revision ?? 0,
upstreamRevision: node.upstream_revision ?? null,
rejectionCount: node.rejection_count ?? 0,
lastRejection: node.last_rejection ?? null,
invalidatedBy: node.invalidated_by ?? null,
createdAt: node.created_at,
updatedAt: node.updated_at,
};
}
/** 同上:会话快照在这一层完成默认值填充与字段转换。 */
function toSession(session: ServerSession): PositionRoundtableSession {
const result: PositionRoundtableSession = {
id: session.id,
externalTaskId: session.external_task_id,
taskScoped: session.task_scoped ?? false,
taskSnapshot: session.task_snapshot ?? {},
chainId: session.chain_id,
chainSnapshot: session.chain_snapshot ?? null,
intentThreadId: session.intent_thread_id,
intentSnapshot: session.intent_snapshot ?? null,
intentConversation: session.intent_conversation ?? null,
summaryThreadId: session.summary_thread_id ?? null,
summarySnapshot: session.summary_snapshot ?? null,
summaryConversation: session.summary_conversation ?? null,
actionPlanThreadId: session.action_plan_thread_id ?? null,
actionPlanSnapshot: session.action_plan_snapshot ?? null,
actionPlanConversation: session.action_plan_conversation ?? null,
status: session.status,
createdAt: session.created_at,
updatedAt: session.updated_at,
};
if (Array.isArray(session.nodes)) {
result.nodes = session.nodes.map(toNode);
}
return result;
}
export async function listPositionSessions(externalTaskId?: string): Promise<PositionRoundtableSession[]> {
const params = externalTaskId ? `?external_task_id=${encodeURIComponent(externalTaskId)}` : "";
const result = await readJson<ServerSession[]>(
await positionApiFetch(`${BASE()}/sessions${params}`),
"list position sessions",
);
return result.map(toSession);
}
export async function createPositionSession(input: {
externalTaskId?: string;
taskSnapshot: Record<string, unknown>;
}): Promise<PositionRoundtableSession> {
const result = await readJson<ServerSession>(
await positionApiFetch(`${BASE()}/sessions`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
external_task_id: input.externalTaskId || null,
task_snapshot: input.taskSnapshot,
}),
}),
"create position session",
);
return toSession(result);
}
export async function getPositionSession(sessionId: string): Promise<PositionRoundtableSession> {
const response = await positionApiFetch(`${BASE()}/sessions/${encodeURIComponent(sessionId)}`);
if (response.status === 404 && !(await isMissingRoute(response))) {
// Consume the body so the connection can be reused, then expose a typed
// error that the page can resolve by selecting another history record.
await response.text().catch(() => "");
throw new PositionSessionNotFoundError(sessionId);
}
const result = await readJson<ServerSession>(response, "get position session");
return toSession(result);
}
export async function updatePositionTaskSnapshot(
sessionId: string,
taskSnapshot: Record<string, unknown>,
): Promise<PositionRoundtableSession> {
const result = await readJson<ServerSession>(
await positionApiFetch(`${BASE()}/sessions/${encodeURIComponent(sessionId)}/task`, {
method: "PUT",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ task_snapshot: taskSnapshot }),
}),
"update position task",
);
return toSession(result);
}
export async function savePositionIntent(
sessionId: string,
input: {
intentThreadId?: string | null;
intentSnapshot: object | null;
conversationSnapshot?: PositionConversationSnapshot | null;
},
): Promise<PositionRoundtableSession> {
const result = await readJson<ServerSession>(
await positionApiFetch(`${BASE()}/sessions/${encodeURIComponent(sessionId)}/intent`, {
method: "PUT",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
intent_thread_id: input.intentThreadId || null,
intent_snapshot: input.intentSnapshot,
conversation_snapshot: input.conversationSnapshot ?? null,
}),
}),
"save position intent",
);
return toSession(result);
}
export async function activatePositionSession(
sessionId: string,
chainId: string,
): Promise<PositionRoundtableSession> {
// 启动即冻结链条快照并创建节点;后续页面不得用当前编辑中的链条替换它。
const result = await readJson<ServerSession>(
await positionApiFetch(`${BASE()}/sessions/${encodeURIComponent(sessionId)}/activate`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ chain_id: chainId, command_id: newCommandId() }),
}),
"activate position session",
);
return toSession(result);
}
export async function archivePositionSession(sessionId: string): Promise<PositionRoundtableSession> {
const result = await readJson<ServerSession>(
await positionApiFetch(`${BASE()}/sessions/${encodeURIComponent(sessionId)}/archive`, {
method: "POST",
}),
"archive position session",
);
return toSession(result);
}
export async function deletePositionSession(sessionId: string): Promise<void> {
const response = await positionApiFetch(`${BASE()}/sessions/${encodeURIComponent(sessionId)}`, {
method: "DELETE",
});
// DELETE is deliberately idempotent from the UI's perspective. A stale
// history row that is already gone still needs to be removed locally.
if (response.status === 404 && !(await isMissingRoute(response))) {
await response.text().catch(() => "");
return;
}
await readJson<{ deleted: boolean }>(response, "delete position session");
}
export async function bindPositionNodeThread(
sessionId: string,
nodeKey: string,
threadId: string,
commandId?: string,
): Promise<PositionRoundtableNode> {
const result = await readJson<ServerNode>(
await positionApiFetch(
`${BASE()}/sessions/${encodeURIComponent(sessionId)}/nodes/${encodeURIComponent(nodeKey)}/thread`,
{
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ thread_id: threadId, command_id: commandId ?? newCommandId() }),
},
),
"bind position node thread",
);
return toNode(result);
}
export async function savePositionNodeConversation(
sessionId: string,
nodeKey: string,
input: {
threadId?: string | null;
conversationSnapshot: PositionConversationSnapshot;
commandId?: string;
},
): Promise<PositionRoundtableNode> {
const result = await readJson<ServerNode>(
await positionApiFetch(
`${BASE()}/sessions/${encodeURIComponent(sessionId)}/nodes/${encodeURIComponent(nodeKey)}/state`,
{
method: "PUT",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
thread_id: input.threadId || null,
conversation_snapshot: input.conversationSnapshot,
command_id: input.commandId ?? newCommandId(),
}),
},
),
"save position node conversation",
);
return toNode(result);
}
export async function savePositionBuiltinConversation(
sessionId: string,
kind: "summary" | "action_plan",
input: {
threadId?: string | null;
conversationSnapshot: PositionConversationSnapshot;
},
): Promise<PositionRoundtableSession> {
const path = kind === "summary" ? "summary/state" : "action-plan/state";
const result = await readJson<ServerSession>(
await positionApiFetch(`${BASE()}/sessions/${encodeURIComponent(sessionId)}/${path}`, {
method: "PUT",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
thread_id: input.threadId || null,
conversation_snapshot: input.conversationSnapshot,
}),
}),
`save position ${kind} conversation`,
);
return toSession(result);
}
export interface PositionDemoSeedPayload {
demo_key: string;
session: Record<string, unknown>;
nodes: Array<Record<string, unknown>>;
}
/**
* Import the canonical frontend demo fixture into the normal SQL tables.
*
* The backend makes this operation idempotent per account and demo key. It
* returns the same Session shape as a real workflow, so no demo-only rendering
* branch is needed after a successful import.
*/
export async function seedPositionDemoSession(
payload: PositionDemoSeedPayload,
): Promise<PositionRoundtableSession> {
const result = await readJson<ServerSession>(
await positionApiFetch(`${BASE()}/demo/seed`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(payload),
}),
"seed position demo session",
);
return toSession(result);
}
/**
* 在普通节点发送前取得受长度限制的隐藏交接上下文。
*
* 该上下文会作为 `task_external_context` 发送给 LangGraph,而不是渲染在聊天
* 消息列表中;这样既能让节点承接上游,又不会把整段交接材料重复展示给用户。
*/
export async function getPositionNodeContext(
sessionId: string,
nodeKey: string,
): Promise<Record<string, unknown>> {
return readJson<Record<string, unknown>>(
await positionApiFetch(
`${BASE()}/sessions/${encodeURIComponent(sessionId)}/nodes/${encodeURIComponent(nodeKey)}/context`,
{ method: "POST" },
),
"build position node context",
);
}
export async function completePositionNodeTurn(
sessionId: string,
nodeKey: string,
input: {
latestAnswer: string;
artifactManifest?: Array<Record<string, unknown>>;
upstreamRevision?: Record<string, unknown> | null;
/** Retry-stable idempotency key; omit to auto-generate per call. */
commandId?: string;
},
): Promise<PositionRoundtableNode> {
// 后端依据 artifactManifest 判断是否有 Markdown 交付物,并在需要时解锁下游。
const result = await readJson<ServerNode>(
await positionApiFetch(
`${BASE()}/sessions/${encodeURIComponent(sessionId)}/nodes/${encodeURIComponent(nodeKey)}/complete-turn`,
{
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
latest_answer: input.latestAnswer,
artifact_manifest: input.artifactManifest ?? [],
upstream_revision: input.upstreamRevision ?? null,
command_id: input.commandId ?? newCommandId(),
}),
},
),
"complete position node turn",
);
return toNode(result);
}
export async function rejectPositionNode(
sessionId: string,
nodeKey: string,
input: { reason?: string; title?: string; commandId?: string },
): Promise<PositionRoundtableSession> {
const result = await readJson<ServerSession>(
await positionApiFetch(
`${BASE()}/sessions/${encodeURIComponent(sessionId)}/nodes/${encodeURIComponent(nodeKey)}/reject`,
{
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
// ``title`` is retained for older callers. The current position
// workspace deliberately sends only an optional reason.
...(input.title ? { title: input.title } : {}),
reason: input.reason ?? "",
command_id: input.commandId ?? newCommandId(),
}),
},
),
"reject position node",
);
return toSession(result);
}
export interface PositionBuiltinRunPreparation {
agentId: string;
displayName: string;
threadId: string | null;
prompt: string;
sourceRevisions: Record<string, number>;
}
interface ServerBuiltinRunPreparation {
agent_id: string;
display_name: string;
thread_id?: string | null;
prompt: string;
source_revisions?: Record<string, number>;
}
function toBuiltinRunPreparation(value: ServerBuiltinRunPreparation): PositionBuiltinRunPreparation {
return {
agentId: value.agent_id,
displayName: value.display_name,
threadId: value.thread_id ?? null,
prompt: value.prompt,
sourceRevisions: value.source_revisions ?? {},
};
}
type BuiltinKind = "summary" | "action_plan";
async function prepareBuiltinRun(
sessionId: string,
kind: BuiltinKind,
message = "",
): Promise<PositionBuiltinRunPreparation> {
// prepare 阶段固定本轮的上游版本与隐藏提示词,避免总结/行动规划引用漂移数据。
const path = kind === "summary" ? "summary/prepare" : "action-plan/prepare";
const result = await readJson<ServerBuiltinRunPreparation>(
await positionApiFetch(`${BASE()}/sessions/${encodeURIComponent(sessionId)}/${path}`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ message }),
}),
`prepare position ${kind}`,
);
return toBuiltinRunPreparation(result);
}
async function completeBuiltinRun(
sessionId: string,
kind: BuiltinKind,
input: {
threadId: string;
content: string;
artifactManifest: Array<Record<string, unknown>>;
sourceRevisions: Record<string, number>;
model?: string;
},
): Promise<PositionRoundtableSession> {
// complete 阶段会校验 sourceRevisions;上游已变化时不能把旧结果覆盖为最新交付。
const path = kind === "summary" ? "summary/complete" : "action-plan/complete";
const result = await readJson<ServerSession>(
await positionApiFetch(`${BASE()}/sessions/${encodeURIComponent(sessionId)}/${path}`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
thread_id: input.threadId,
content: input.content,
artifact_manifest: input.artifactManifest,
source_revisions: input.sourceRevisions,
model: input.model || null,
}),
}),
`complete position ${kind}`,
);
return toSession(result);
}
export function preparePositionSummary(
sessionId: string,
message = "",
): Promise<PositionBuiltinRunPreparation> {
return prepareBuiltinRun(sessionId, "summary", message);
}
export function completePositionSummary(
sessionId: string,
input: Parameters<typeof completeBuiltinRun>[2],
): Promise<PositionRoundtableSession> {
return completeBuiltinRun(sessionId, "summary", input);
}
export function preparePositionActionPlan(
sessionId: string,
message = "",
): Promise<PositionBuiltinRunPreparation> {
return prepareBuiltinRun(sessionId, "action_plan", message);
}
export function completePositionActionPlan(
sessionId: string,
input: Parameters<typeof completeBuiltinRun>[2],
): Promise<PositionRoundtableSession> {
return completeBuiltinRun(sessionId, "action_plan", input);
}