fix: avoid MySQL thread_meta refresh races and harden roundtable bootstrap

Skip post-commit ORM refresh on threads_meta creates so concurrent MySQL
read/write-split deployments no longer fail with Could not refresh instance;
clean up partial roundtable threads on checkpoint failure, and ship related
position-roundtable/frontend build fixes.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
董宏凯andCursor committed 2026-07-31 21:00:56 +08:00
1 parent e52261fc3d
commit a9d66536c3
12 files changed
+503 -127

No files matched your search

+35
View File
@@ -0,0 +1,35 @@
# Never send source-control history, local dependencies, runtime state, or
# credentials to the Docker daemon. The offline images are built from the
# repository root by deploy/1panel-offline/Dockerfile.*.
.git
.runtime
.history
tmp
dist
# Credentials and environment-specific configuration are supplied at runtime
# through the offline delivery package, never copied into an image layer.
**/.env
**/.env.*
offline-backend-20260512/backend/.deer-flow
offline-backend-20260512/backend/logs
offline-backend-20260512/backend/docker-data
offline-backend-20260512/backend/dist
offline-backend-20260512/backend/wheelhouse
offline-backend-20260512/backend/offline-wheels
offline-backend-20260512/backend/.venv
offline-backend-20260512/backend/.pytest_cache
offline-backend-20260512/backend/.ruff_cache
# Reproducibly install frontend dependencies inside the image instead.
frontend-web/node_modules
frontend-web/dist
frontend-web/.vite
frontend-web/.offline-pack-build
frontend-web/.offline-extract-test
frontend-web/node_modules-offline-*.tar.gz
# Editor and operating-system metadata.
**/__pycache__
**/*.py[cod]
**/.DS_Store
@@ -168,13 +168,6 @@ function BusinessChainCanvas({
className="flex min-h-full min-w-max flex-col items-center px-5 py-6"
style={{ transform: `translate(${view.tx}px, ${view.ty}px)` }}
>
<FlowNode
title="任务理解智能体"
description="任务理解岗"
accent={NODE_COLORS[0]}
completed
width={nodeWidth}
/>
{stages.map((stage, stageIndex) => {
const stageWidth = Math.max(
nodeWidth,
@@ -182,12 +175,14 @@ function BusinessChainCanvas({
);
return (
<div key={`${stage.goal ?? "stage"}-${stageIndex}`} className="flex flex-col items-center">
<FlowConnector
branches={stage.seats.length}
width={stageWidth}
nodeWidth={nodeWidth}
nodeGap={nodeGap}
/>
{stageIndex > 0 && (
<FlowConnector
branches={stage.seats.length}
width={stageWidth}
nodeWidth={nodeWidth}
nodeGap={nodeGap}
/>
)}
<div className="flex items-start justify-center" style={{ gap: nodeGap }}>
{stage.seats.map((seat) => {
const accent = NODE_COLORS[(colorIndex++ + 1) % NODE_COLORS.length]!;
@@ -205,20 +200,6 @@ function BusinessChainCanvas({
</div>
);
})}
<FlowConnector branches={1} width={nodeWidth} nodeWidth={nodeWidth} nodeGap={nodeGap} />
<FlowNode
title="方案总结智能体"
description="内置总结节点"
accent="#8b83f5"
width={nodeWidth}
/>
<FlowConnector branches={1} width={nodeWidth} nodeWidth={nodeWidth} nodeGap={nodeGap} />
<FlowNode
title="行动规划智能体"
description="内置规划节点"
accent="#78bce6"
width={nodeWidth}
/>
</div>
</div>
);
@@ -61,6 +61,8 @@ import {
DropdownMenuTrigger,
} from "@/components/ui/dropdown-menu";
import { useTheme } from "@/contexts/ThemeContext";
import { isEmbeddedHost } from "@/core/embed/frame-context";
import { emitWujieEvent, isWujieEnv } from "@/core/embed/wujie";
import { LayoutModeSwitcher, ThemeSwitcher } from "@/components/workspace/appearance-controls";
import { getAPIClient } from "@/core/api/api-client";
import { useAgent } from "@/core/agents";
@@ -167,6 +169,131 @@ type SandboxThreadSnapshot = {
signature: string;
};
type PositionRoundtableParentReturnMessage = {
source: "deerflow-position-roundtable";
type: "deerflow:position-roundtable:return";
payload: {
route: string | null;
fallback: "history-back";
};
};
type PositionRoundtableBridgeInitMessage = {
source: "code-zk-roundtable-host";
type: "code-zk:roundtable-bridge:init";
};
const POSITION_ROUNDTABLE_BRIDGE_SOURCE = "code-zk-roundtable-host";
const POSITION_ROUNDTABLE_BRIDGE_INIT_TYPE = "code-zk:roundtable-bridge:init";
const POSITION_ROUNDTABLE_BRIDGE_READY_TYPE = "deerflow:position-roundtable:bridge-ready";
function isEnabledEmbedFlag(value: string | null): boolean {
const normalized = value?.trim().toLowerCase();
return normalized === "1" || normalized === "true" || normalized === "yes";
}
/**
* Some hosts only retain URL parameters after the login redirect. Treat those
* established embed markers as an embedded context as well as checking the
* actual browser frame/Wujie runtime.
*/
function hasEmbeddedRouteHint(searchParams: URLSearchParams): boolean {
return (
searchParams.get("from") === "iframe" ||
searchParams.get("chatMode")?.trim().toLowerCase() === "iframe" ||
isEnabledEmbedFlag(searchParams.get("embed")) ||
isEnabledEmbedFlag(searchParams.get("hideChrome")) ||
isEnabledEmbedFlag(searchParams.get("hideHeader"))
);
}
/**
* `parentReturnTo` belongs to the embedding application, not this SPA. Keep
* it constrained to a route/hash path so the same-origin fallback can never
* turn an externally supplied URL into an arbitrary top-level navigation.
*/
function parentReturnRoute(value: string | null): string | null {
const route = value?.trim();
if (!route || route.length > 2048) return null;
if ((!route.startsWith("/") && !route.startsWith("#")) || route.startsWith("//") || route.includes("\\")) {
return null;
}
return route;
}
/**
* Request navigation from the embedding host. A real iframe parent is always
* the nearest integration boundary, even if an outer layer happens to be
* Wujie. This matters for Wujie host -> code-zk -> DeerFlow iframe: code-zk
* owns the postMessage contract and its own router. Wujie's event bus is only
* an additional fallback for a directly mounted Wujie child.
*/
function returnToEmbeddingParent(route: string | null, bridgePort?: MessagePort | null): void {
const message: PositionRoundtableParentReturnMessage = {
source: "deerflow-position-roundtable",
type: "deerflow:position-roundtable:return",
payload: { route, fallback: "history-back" },
};
if (typeof window === "undefined") return;
// In a Wujie host -> code-zk -> DeerFlow topology, the DOM iframe can be
// re-parented by Wujie. A port transferred by code-zk survives that DOM
// topology change and is the only channel guaranteed to reach code-zk.
if (bridgePort) {
try {
bridgePort.postMessage(message);
console.info("[position-roundtable] posted return through MessageChannel bridge");
return;
} catch {
// The owner may have unmounted/reloaded. Fall back to native bridges.
}
}
console.info("[position-roundtable] embedded return requested", {
route,
hasNativeParent: window.parent !== window,
wujie: isWujieEnv(),
});
if (window.parent === window) {
// Some Wujie execution modes expose no meaningful native parent window.
// In that direct-mount case the event bus is the only available bridge.
if (isWujieEnv()) {
emitWujieEvent("position-roundtable:return", message.payload);
console.info("[position-roundtable] emitted Wujie-only return event");
}
return;
}
try {
if (window.parent.location.origin === window.location.origin) {
if (route?.startsWith("#")) {
window.parent.location.hash = route;
} else if (route) {
window.parent.location.assign(route);
} else {
window.parent.history.back();
}
console.info("[position-roundtable] completed same-origin parent return");
return;
}
} catch {
// Cross-origin parent: fall through to the explicit postMessage contract.
}
window.parent.postMessage(message, "*");
console.info("[position-roundtable] posted return message to direct parent");
// A nested iframe must never skip the direct-parent notification merely
// because Wujie markers were inherited or injected. Emitting as well keeps
// directly mounted Wujie integrations compatible when their host listens.
if (isWujieEnv()) {
emitWujieEvent("position-roundtable:return", message.payload);
console.info("[position-roundtable] emitted supplemental Wujie return event");
}
}
/**
* URL parameters are externally supplied, so never cast the raw value into
* page state without checking it against the position directory first.
@@ -609,6 +736,50 @@ export default function PositionRoundtablePage({
: rawSessionIdFromUrl;
const forceNewSession =
!standaloneDemoMode && searchParams.get("new") === "1";
const embeddedHost = useMemo(() => isEmbeddedHost(), []);
const isEmbeddingContext = embeddedHost || hasEmbeddedRouteHint(searchParams);
const embeddingParentReturnRoute = useMemo(
() =>
parentReturnRoute(
searchParams.get("parentReturnTo") ??
searchParams.get("parentRoute") ??
searchParams.get("returnTo"),
),
[searchParamsString],
);
const embeddingBridgePortRef = useRef<MessagePort | null>(null);
const handleReturnToEmbeddingParent = useCallback(() => {
returnToEmbeddingParent(embeddingParentReturnRoute, embeddingBridgePortRef.current);
}, [embeddingParentReturnRoute]);
useEffect(() => {
const receiveBridgePort = (event: MessageEvent<PositionRoundtableBridgeInitMessage>) => {
const message = event.data;
const bridgePort = event.ports?.[0];
if (
message?.source !== POSITION_ROUNDTABLE_BRIDGE_SOURCE ||
message?.type !== POSITION_ROUNDTABLE_BRIDGE_INIT_TYPE ||
!bridgePort
) {
return;
}
embeddingBridgePortRef.current?.close();
embeddingBridgePortRef.current = bridgePort;
bridgePort.start?.();
bridgePort.postMessage({
source: "deerflow-position-roundtable",
type: POSITION_ROUNDTABLE_BRIDGE_READY_TYPE,
});
console.info("[position-roundtable] MessageChannel bridge connected");
};
window.addEventListener("message", receiveBridgePort);
return () => {
window.removeEventListener("message", receiveBridgePort);
embeddingBridgePortRef.current?.close();
embeddingBridgePortRef.current = null;
};
}, []);
const [positionSession, setPositionSession] = useState<PositionRoundtableSession | null>(null);
const positionSessionRef = useRef<PositionRoundtableSession | null>(null);
const sessionCreationRef = useRef<Promise<PositionRoundtableSession> | null>(null);
@@ -2273,6 +2444,17 @@ export default function PositionRoundtablePage({
</div>
<div className="flex shrink-0 items-center gap-2">
{isEmbeddingContext && (
<TButton
size="small"
variant="outline"
className="shrink-0 whitespace-nowrap"
icon={<ArrowLeft className="size-3.5" />}
onClick={handleReturnToEmbeddingParent}
>
返回
</TButton>
)}
<TButton
size="small"
variant="outline"
@@ -2443,14 +2625,9 @@ export default function PositionRoundtablePage({
<BusinessChainFlow
chain={activeChain}
nodes={positionSession?.nodes ?? []}
session={positionSession}
activeNodeKey={activeNodeKey}
onSelectIntent={
activePosition.isIntentPosition ? openIntentFlowDetail : undefined
}
onSelectNode={openNodeFlowDetail}
onSelectSeat={openSeatFlowDetail}
onSelectBuiltin={handleSelectBuiltinStage}
embedded
/>
)}
@@ -4215,23 +4392,17 @@ const FLOW_NODE_GAP = 10;
function BusinessChainFlow({
chain,
nodes,
session,
activeNodeKey,
onSelectIntent,
onSelectNode,
onSelectSeat,
onSelectBuiltin,
className,
embedded = false,
}: {
chain: BusinessChain | null;
nodes: PositionRoundtableNode[];
session: PositionRoundtableSession | null;
activeNodeKey: string | null;
onSelectIntent?: () => void;
onSelectNode: (node: PositionRoundtableNode) => void;
onSelectSeat?: (seat: BusinessChain["seats"][number], stageIndex: number, seatIndex: number) => void;
onSelectBuiltin: (kind: "summary" | "action_plan") => void;
className?: string;
embedded?: boolean;
}) {
@@ -4287,12 +4458,9 @@ function BusinessChainFlow({
<ChainFlowCanvas
chain={chain}
nodes={nodes}
session={session}
activeNodeKey={activeNodeKey}
onSelectIntent={onSelectIntent}
onSelectNode={onSelectNode}
onSelectSeat={onSelectSeat}
onSelectBuiltin={onSelectBuiltin}
className="min-h-0 flex-1"
/>
<Dialog open={flowDialogOpen} onOpenChange={setFlowDialogOpen}>
@@ -4307,16 +4475,7 @@ function BusinessChainFlow({
<ChainFlowCanvas
chain={chain}
nodes={nodes}
session={session}
activeNodeKey={activeNodeKey}
onSelectIntent={
onSelectIntent
? () => {
onSelectIntent();
setFlowDialogOpen(false);
}
: undefined
}
onSelectNode={(node) => {
onSelectNode(node);
setFlowDialogOpen(false);
@@ -4325,10 +4484,6 @@ function BusinessChainFlow({
onSelectSeat?.(seat, stageIndex, seatIndex);
setFlowDialogOpen(false);
}}
onSelectBuiltin={(kind) => {
onSelectBuiltin(kind);
setFlowDialogOpen(false);
}}
className="h-full"
expanded
/>
@@ -4344,23 +4499,17 @@ function BusinessChainFlow({
function ChainFlowCanvas({
chain,
nodes,
session,
activeNodeKey,
onSelectIntent,
onSelectNode,
onSelectSeat,
onSelectBuiltin,
className,
expanded = false,
}: {
chain: BusinessChain;
nodes: PositionRoundtableNode[];
session: PositionRoundtableSession | null;
activeNodeKey: string | null;
onSelectIntent?: () => void;
onSelectNode: (node: PositionRoundtableNode) => void;
onSelectSeat?: (seat: BusinessChain["seats"][number], stageIndex: number, seatIndex: number) => void;
onSelectBuiltin: (kind: "summary" | "action_plan") => void;
className?: string;
expanded?: boolean;
}) {
@@ -4376,12 +4525,6 @@ function ChainFlowCanvas({
} | null>(null);
const nodeWidth = expanded ? 190 : FLOW_NODE_WIDTH;
const nodeGap = expanded ? 16 : FLOW_NODE_GAP;
const allDone = allExecutableNodesDone(nodes);
const summaryStatus = builtinNodeStatus(session?.summarySnapshot, allDone);
const actionPlanStatus = builtinNodeStatus(
session?.actionPlanSnapshot,
snapshotStatus(session?.summarySnapshot) === "done",
);
const connectorColor =
theme === "dark" ? "rgba(143, 196, 255, 0.95)" : "rgba(148, 163, 184, 0.88)";
@@ -4439,14 +4582,6 @@ function ChainFlowCanvas({
className="flex min-h-full min-w-max flex-col items-center p-5"
style={{ transform: `translate(${view.tx}px, ${view.ty}px)` }}
>
<CompactChainNode
title="情报分析智能体"
description="情报分析岗"
accent="#3b82f6"
nodeStatus={nodes.length > 0 ? "done" : "ready"}
width={nodeWidth}
onClick={onSelectIntent}
/>
{stages.map((stage, stageIndex) => {
const stageWidth = Math.max(
nodeWidth,
@@ -4457,13 +4592,15 @@ function ChainFlowCanvas({
key={`${stage.goal ?? "stage"}-${stageIndex}`}
className="flex flex-col items-center"
>
<VerticalFlowConnector
seatCount={stage.seats.length}
width={stageWidth}
nodeWidth={nodeWidth}
nodeGap={nodeGap}
connectorColor={connectorColor}
/>
{stageIndex > 0 && (
<VerticalFlowConnector
seatCount={stage.seats.length}
width={stageWidth}
nodeWidth={nodeWidth}
nodeGap={nodeGap}
connectorColor={connectorColor}
/>
)}
<div className="flex items-start justify-center" style={{ gap: nodeGap }}>
{stage.seats.map((seat, seatIndex) => {
const node = nodes.find(
@@ -4499,38 +4636,6 @@ function ChainFlowCanvas({
</div>
);
})}
<VerticalFlowConnector
seatCount={1}
width={nodeWidth}
nodeWidth={nodeWidth}
nodeGap={nodeGap}
connectorColor={connectorColor}
/>
<CompactChainNode
title="方案总结智能体"
description="内置总结节点"
accent="#6366f1"
nodeStatus={summaryStatus}
width={nodeWidth}
titleAttr="对所有有效交付产物进行方案总结"
onClick={() => onSelectBuiltin("summary")}
/>
<VerticalFlowConnector
seatCount={1}
width={nodeWidth}
nodeWidth={nodeWidth}
nodeGap={nodeGap}
connectorColor={connectorColor}
/>
<CompactChainNode
title="行动规划智能体"
description="内置规划节点"
accent="#0ea5e9"
nodeStatus={actionPlanStatus}
width={nodeWidth}
titleAttr="基于方案总结和岗位产物拆解子任务"
onClick={() => onSelectBuiltin("action_plan")}
/>
</div>
</div>
);
@@ -82,8 +82,11 @@ import { listAgents } from "@/core/agents";
import { getStoredAuth } from "@/core/auth";
import {
POSITION_OPTIONS,
setRuntimePositionOptions,
type PositionOption,
type PositionId,
} from "@/position-roundtable/lib/positions";
import { usePositionRoles } from "@/strategy-components/api/position-roles";
import {
filterRecommendCandidates,
@@ -160,6 +163,33 @@ export default function BusinessChainEditorPage() {
// 岗位配置只由岗位协同页入口开启,普通圆桌编辑器完全不显示也不依赖该控件。
// `returnTo` 只接受站内绝对路径,避免把导航带往外部地址。
const positionConfigMode = searchParams.get("mode") === "position";
// This is the same directory the administrator edits in “岗位角色配置”.
// The editor used to read the static POSITION_OPTIONS array below, which
// meant a newly configured role never appeared in this seat dropdown.
const { data: configuredPositionRoles = [] } = usePositionRoles({
refreshOnMountOrFocus: positionConfigMode,
});
const positionOptions = useMemo<PositionOption[]>(
() =>
configuredPositionRoles.length > 0
? [...configuredPositionRoles]
.sort((left, right) => left.sortOrder - right.sortOrder)
.map((role) => ({
id: role.id,
name: role.name,
description: role.description,
isIntentPosition: role.roleType === "intent",
enabled: role.enabled,
isBuiltin: role.isBuiltin,
sortOrder: role.sortOrder,
}))
: [...POSITION_OPTIONS],
[configuredPositionRoles],
);
// Keep flow previews and historical seat labels on the same live directory.
useEffect(() => {
setRuntimePositionOptions(positionOptions);
}, [positionOptions]);
const requestedReturnTo = searchParams.get("returnTo");
const returnPath =
positionConfigMode && requestedReturnTo?.startsWith("/")
@@ -1244,6 +1274,7 @@ export default function BusinessChainEditorPage() {
onSeatSkillDirectiveChange={setSeatSkillDirectiveCb}
onSeatModeChange={setSeatModeCb}
positionConfigMode={positionConfigMode}
positionOptions={positionOptions}
onSeatPositionChange={setSeatPositionCb}
onStageGoalChange={setGoal}
goalOptimizingIdx={goalOptimizingIdx}
@@ -1426,6 +1457,7 @@ function StagesColumn({
onSeatSkillDirectiveChange,
onSeatModeChange,
positionConfigMode,
positionOptions,
onSeatPositionChange,
onStageGoalChange,
goalOptimizingIdx,
@@ -1447,6 +1479,8 @@ function StagesColumn({
onSeatModeChange: (agentId: string, mode: SeatMode | null) => void;
/** 仅岗位协同入口启用;普通圆桌编辑器不显示岗位控件。 */
positionConfigMode: boolean;
/** Live directory from the administrator's position-role configuration. */
positionOptions: readonly PositionOption[];
onSeatPositionChange: (agentId: string, positionId: PositionId | null) => void;
onStageGoalChange: (stageIdx: number, goal: string) => void;
/** 正在 AI 优化阶段目标的 stage 下标(null = 没有进行中的优化)。 */
@@ -1554,6 +1588,7 @@ function StagesColumn({
onSeatSkillDirectiveChange={onSeatSkillDirectiveChange}
onSeatModeChange={onSeatModeChange}
positionConfigMode={positionConfigMode}
positionOptions={positionOptions}
onSeatPositionChange={onSeatPositionChange}
onGoalChange={(goal) => onStageGoalChange(stageIdx, goal)}
goalOptimizing={goalOptimizingIdx === stageIdx}
@@ -1587,6 +1622,7 @@ function StageSection({
onSeatSkillDirectiveChange,
onSeatModeChange,
positionConfigMode,
positionOptions,
onSeatPositionChange,
onGoalChange,
goalOptimizing,
@@ -1609,6 +1645,7 @@ function StageSection({
onSeatSkillDirectiveChange: (agentId: string, enabled: boolean) => void;
onSeatModeChange: (agentId: string, mode: SeatMode | null) => void;
positionConfigMode: boolean;
positionOptions: readonly PositionOption[];
onSeatPositionChange: (agentId: string, positionId: PositionId | null) => void;
onGoalChange: (goal: string) => void;
/** 本 Stage 的阶段目标正在 AI 优化中。 */
@@ -1774,6 +1811,7 @@ function StageSection({
onSkillDirectiveChange={(enabled) => onSeatSkillDirectiveChange(seat.agent_id, enabled)}
onSeatModeChange={(mode) => onSeatModeChange(seat.agent_id, mode)}
showPositionConfig={positionConfigMode}
positionOptions={positionOptions}
onPositionChange={(positionId) => onSeatPositionChange(seat.agent_id, positionId)}
onDetail={() => onSeatDetail(seat.agent_id, seat.name)}
/>
@@ -1801,6 +1839,7 @@ function SeatStrip({
onSkillDirectiveChange,
onSeatModeChange,
showPositionConfig,
positionOptions,
onPositionChange,
onDetail,
}: {
@@ -1822,12 +1861,41 @@ function SeatStrip({
onSeatModeChange: (mode: SeatMode | null) => void;
/** 仅岗位协同入口展示的岗位配置。 */
showPositionConfig: boolean;
/** Current position directory, injected by the business-chain editor. */
positionOptions: readonly PositionOption[];
onPositionChange: (positionId: PositionId | null) => void;
onDetail: () => void;
}) {
const { attributes, listeners, setNodeRef, transform, transition, isDragging } =
useSortable({ id: SEAT_PREFIX + seat.agent_id });
const avatarType = getAvatarTypeFor(seat.agent_id, globalIndex);
const selectablePositionOptions = useMemo(() => {
const assignedId = seat.position_id ?? null;
const assigned = assignedId
? positionOptions.find((position) => position.id === assignedId)
: undefined;
const enabled = positionOptions.filter((position) => position.enabled !== false);
// Do not silently erase an archived configuration from a saved chain.
// It remains visible but cannot be selected for another seat.
const historical =
assignedId && (!assigned || assigned.enabled === false)
? [
{
label: assigned
? `${assigned.name}(已停用)`
: `${assignedId}(岗位已删除)`,
value: assignedId,
disabled: true,
},
]
: [];
return [
{ label: "未分配", value: "" },
...historical,
...enabled.map((position) => ({ label: position.name, value: position.id })),
];
}, [positionOptions, seat.position_id]);
return (
<div
ref={setNodeRef}
@@ -1880,13 +1948,7 @@ function SeatStrip({
<TSelect
size="small"
value={seat.position_id ?? ""}
options={[
{ label: "未分配", value: "" },
...POSITION_OPTIONS.map((position) => ({
label: position.name,
value: position.id,
})),
]}
options={selectablePositionOptions}
onChange={(value) =>
onPositionChange(value ? (String(value) as PositionId) : null)
}
@@ -158,12 +158,21 @@ export async function replacePositionRoles(roles: PositionRole[]): Promise<Posit
return saved;
}
export function usePositionRoles(options?: { enabled?: boolean }) {
export function usePositionRoles(options?: {
/** Whether this query should run. */
enabled?: boolean;
/**
* Consumers that edit business-chain seats need the latest shared directory
* whenever their page is opened or focused.
*/
refreshOnMountOrFocus?: boolean;
}) {
return useQuery({
queryKey: POSITION_ROLES_QUERY_KEY,
queryFn: listPositionRoles,
staleTime: 30_000,
refetchOnWindowFocus: false,
refetchOnMount: options?.refreshOnMountOrFocus ? "always" : true,
refetchOnWindowFocus: options?.refreshOnMountOrFocus ? "always" : false,
enabled: options?.enabled ?? true,
});
}
+5 -1
View File
@@ -45,7 +45,11 @@ export default defineConfig({
},
],
},
"base":"./",
// Keep the current portable-file default, while allowing container builds to
// emit absolute asset URLs. The latter is required for BrowserRouter pages
// behind Nginx: a request to /page/... must still load /assets/... rather
// than /page/assets/....
base: process.env.VITE_ASSET_BASE ?? "./",
build: {
cssMinify: "lightningcss",
target: "chrome100",
+19 -1
View File
@@ -127,7 +127,25 @@ Clients may retry the compatibility route only for a framework/proxy-level
`multi_agent._create_thread_direct`, exactly like multi-agent initialization;
do not restore the old HTTP loopback to `/api/threads`, because deployed
Gateway ports/proxies may make the automatic page initialization fail while
later manual turns still work. Streaming stays on the existing loopback path.
later manual turns still work. If checkpoint creation fails or the parallel seat
task is cancelled, the helper compensates by deleting any partial checkpoint and
metadata row (the cancellation branch runs under `asyncio.shield` so a second
cancellation cannot abort it half-way). Streaming stays on the existing loopback
path.
`ThreadMetaStore.create` never issues a post-commit `session.refresh()`. Every
`ThreadMetaRow` column is assigned before commit and the session factory uses
`expire_on_commit=False`, so the returned dict is built from the in-memory row.
The removed read-back ran on a *different* pooled connection than the INSERT,
which a MySQL read/write-splitting endpoint can route to a lagging replica — the
row comes back empty and SQLAlchemy raises `Could not refresh instance`, surfacing
as `500 "Failed to create thread"`. Concurrency makes that far more likely, so a
single-user deployment never sees it. No caller uses the return value, and all
seven call sites (`multi_agent._create_thread_direct`, `services.start_run`,
`routers/threads.create_thread`, `roundtable_inprocess_gateway`,
`scheduler/service`, `thread_shares`, `authz`) benefit. Keep new `create`
implementations read-back-free; `tests/test_position_roundtable_sessions.py::
test_thread_meta_create_never_reads_back_after_commit` guards this.
### Harness / App Split
+8 -2
View File
@@ -481,8 +481,14 @@ Its live intent conversation still uses the shared `/api/intent/init` and
`/api/intent/stream` endpoints used by the original multi-agent roundtable.
`/api/intent/init` creates its empty thread directly through the in-process
thread store/checkpointer, matching multi-agent initialization and avoiding a
deployment-sensitive `127.0.0.1:<gateway-port>` loopback. Only the recovery
snapshot layer is position-specific. Deployments must expose
deployment-sensitive `127.0.0.1:<gateway-port>` loopback. If initial checkpoint
creation fails or is cancelled, bootstrap removes both the partial checkpoint and
its thread metadata row. Thread metadata creation issues no post-commit ORM
read-back on any code path: all returned fields are assigned before commit, and
skipping the extra read keeps a MySQL read/write-splitting endpoint from sending
it to a lagging replica (which fails the request with
`Could not refresh instance`). Only the
recovery snapshot layer is position-specific. Deployments must expose
either `/api/position-roundtable/*` or the compatibility mirror
`/api/multi-agent/position-roundtable/*`; updated clients probe both and only
retry when the first response is the framework/proxy-level `404 Not Found`.
@@ -442,10 +442,30 @@ async def _create_thread_direct(request: Request, *, metadata: dict[str, Any] |
"parents": {},
"created_at": now,
}
async def _cleanup_partial_thread() -> None:
delete_checkpoint = getattr(checkpointer, "adelete_thread", None)
if callable(delete_checkpoint):
try:
await delete_checkpoint(thread_id)
except Exception:
logger.exception("[init-direct] partial checkpoint cleanup failed for %s", thread_id)
try:
await thread_store.delete(thread_id)
except Exception:
logger.exception("[init-direct] partial thread_meta cleanup failed for %s", thread_id)
try:
await checkpointer.aput(config, empty_checkpoint(), ckpt_metadata, {})
except asyncio.CancelledError:
# A sibling seat may fail while this task is between metadata commit and
# checkpoint creation. Compensate before propagating cancellation, under
# shield so a second cancellation cannot abort the cleanup half-way.
await asyncio.shield(_cleanup_partial_thread())
raise
except Exception:
logger.exception("[init-direct] checkpoint create failed for %s", thread_id)
await _cleanup_partial_thread()
raise HTTPException(status_code=500, detail="Failed to create thread")
return thread_id
@@ -30,6 +30,12 @@ class ThreadMetaStore(abc.ABC):
display_name: str | None = None,
metadata: dict | None = None,
) -> dict:
"""Create one metadata row.
Implementations must assemble the returned dict from the values passed
in, without a post-commit read-back: on MySQL such a read can be routed
to a lagging replica behind a read/write-splitting endpoint and fail.
"""
pass
@abc.abstractmethod
@@ -45,6 +45,7 @@ class ThreadMetaRepository(ThreadMetaStore):
assistant_id=assistant_id,
user_id=resolved_user_id,
display_name=display_name,
status="idle",
metadata_json=metadata or {},
created_at=now,
updated_at=now,
@@ -52,7 +53,14 @@ class ThreadMetaRepository(ThreadMetaStore):
async with self._sf() as session:
session.add(row)
await session.commit()
await session.refresh(row)
# Deliberately no session.refresh(): every column above is assigned
# explicitly and the session factory uses expire_on_commit=False, so
# the in-memory row is already complete. The re-SELECT would run on a
# different pooled connection than the INSERT, which a MySQL
# read/write-splitting endpoint can route to a lagging replica --
# the row comes back empty and SQLAlchemy raises
# "Could not refresh instance". Concurrency makes that far more
# likely, so a single-user deployment never sees it.
return self._row_to_dict(row)
async def get(
@@ -2,19 +2,24 @@
from __future__ import annotations
import asyncio
from types import SimpleNamespace
from typing import Any
import pytest
import pytest_asyncio
from fastapi import HTTPException
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from app.gateway import deps as gateway_deps
from app.gateway import services as gateway_services
from app.gateway.app import create_app
from app.gateway.routers import intent as intent_router
from app.gateway.routers import multi_agent
from app.gateway.routers import position_roundtable as position_roundtable_router
from deerflow.persistence.base import Base
from deerflow.persistence.position_roundtable import PositionRoundtableRepository
from deerflow.persistence.thread_meta.sql import ThreadMetaRepository
@pytest_asyncio.fixture
@@ -971,3 +976,120 @@ async def test_locked_node_cannot_bind_a_thread_before_its_stage_is_ready():
)
assert error.value.status_code == 409
assert "未解锁" in str(error.value.detail)
class _ThreadMetaSessionProbe:
def __init__(self, events: list[str]) -> None:
self._events = events
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb) -> None:
return None
def add(self, row) -> None:
self._events.append(f"add:{row.thread_id}")
async def commit(self) -> None:
self._events.append("commit")
async def refresh(self, row) -> None:
self._events.append(f"refresh:{row.thread_id}")
@pytest.mark.asyncio
async def test_thread_meta_create_never_reads_back_after_commit():
"""No caller may depend on a post-commit re-SELECT of the new row.
On MySQL behind a read/write-splitting endpoint that read can land on a
lagging replica, come back empty, and raise "Could not refresh instance".
"""
events: list[str] = []
repo = ThreadMetaRepository(lambda: _ThreadMetaSessionProbe(events)) # type: ignore[arg-type]
direct = await repo.create(
"roundtable-thread",
user_id=None,
metadata={"thread_type": "roundtable"},
)
ordinary = await repo.create("ordinary-thread", user_id=None, display_name="对话")
assert events == [
"add:roundtable-thread",
"commit",
"add:ordinary-thread",
"commit",
]
# The returned dict is assembled from the in-memory row, so it must still be
# complete for every column -- including ones no caller passed in.
assert direct["status"] == "idle"
assert direct["metadata"] == {"thread_type": "roundtable"}
assert direct["created_at"] and direct["updated_at"]
assert ordinary["status"] == "idle"
assert ordinary["display_name"] == "对话"
class _ThreadStoreProbe:
def __init__(self) -> None:
self.created: list[tuple[str, dict[str, Any]]] = []
self.deleted: list[str] = []
async def create(self, thread_id: str, **kwargs) -> dict[str, Any]:
self.created.append((thread_id, kwargs))
return {"thread_id": thread_id}
async def delete(self, thread_id: str) -> None:
self.deleted.append(thread_id)
class _FailingCheckpointer:
def __init__(self, exc: BaseException) -> None:
self._exc = exc
self.deleted: list[str] = []
async def aput(self, *args, **kwargs):
raise self._exc
async def adelete_thread(self, thread_id: str) -> None:
self.deleted.append(thread_id)
@pytest.mark.asyncio
async def test_direct_thread_creation_cleans_failed_checkpoint(monkeypatch):
store = _ThreadStoreProbe()
checkpointer = _FailingCheckpointer(RuntimeError("checkpoint unavailable"))
monkeypatch.setattr(gateway_deps, "get_thread_store", lambda request: store)
monkeypatch.setattr(gateway_deps, "get_checkpointer", lambda request: checkpointer)
with pytest.raises(HTTPException) as exc_info:
await multi_agent._create_thread_direct(
SimpleNamespace(),
metadata={"thread_type": "roundtable"},
)
assert exc_info.value.status_code == 500
assert len(store.created) == 1
thread_id, create_kwargs = store.created[0]
assert create_kwargs["metadata"] == {"thread_type": "roundtable"}
assert store.deleted == [thread_id]
assert checkpointer.deleted == [thread_id]
@pytest.mark.asyncio
async def test_direct_thread_creation_cleans_metadata_when_cancelled(monkeypatch):
store = _ThreadStoreProbe()
checkpointer = _FailingCheckpointer(asyncio.CancelledError())
monkeypatch.setattr(gateway_deps, "get_thread_store", lambda request: store)
monkeypatch.setattr(gateway_deps, "get_checkpointer", lambda request: checkpointer)
with pytest.raises(asyncio.CancelledError):
await multi_agent._create_thread_direct(
SimpleNamespace(),
metadata={"thread_type": "roundtable"},
)
thread_id, _ = store.created[0]
assert store.deleted == [thread_id]
assert checkpointer.deleted == [thread_id]