From a9d66536c3e59cac09aef234d53c9bf7a935c254 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Fri, 31 Jul 2026 21:00:56 +0800 Subject: [PATCH 1/2] 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 --- .dockerignore | 35 +++ .../components/PositionBusinessChainFlow.tsx | 35 +-- .../pages/PositionRoundtablePage.tsx | 277 ++++++++++++------ .../pages/BusinessChainEditorPage.tsx | 76 ++++- .../strategy-components/api/position-roles.ts | 13 +- frontend-web/vite.config.ts | 6 +- offline-backend-20260512/backend/CLAUDE.md | 20 +- offline-backend-20260512/backend/README.md | 10 +- .../app/gateway/routers/multi_agent.py | 20 ++ .../deerflow/persistence/thread_meta/base.py | 6 + .../deerflow/persistence/thread_meta/sql.py | 10 +- .../test_position_roundtable_sessions.py | 122 ++++++++ 12 files changed, 503 insertions(+), 127 deletions(-) create mode 100644 .dockerignore diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 00000000..e2541f15 --- /dev/null +++ b/.dockerignore @@ -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 diff --git a/frontend-web/src/position-roundtable/components/PositionBusinessChainFlow.tsx b/frontend-web/src/position-roundtable/components/PositionBusinessChainFlow.tsx index 3395bb87..dfded393 100644 --- a/frontend-web/src/position-roundtable/components/PositionBusinessChainFlow.tsx +++ b/frontend-web/src/position-roundtable/components/PositionBusinessChainFlow.tsx @@ -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)` }} > - {stages.map((stage, stageIndex) => { const stageWidth = Math.max( nodeWidth, @@ -182,12 +175,14 @@ function BusinessChainCanvas({ ); return (
- + {stageIndex > 0 && ( + + )}
{stage.seats.map((seat) => { const accent = NODE_COLORS[(colorIndex++ + 1) % NODE_COLORS.length]!; @@ -205,20 +200,6 @@ function BusinessChainCanvas({
); })} - - - -
); diff --git a/frontend-web/src/position-roundtable/pages/PositionRoundtablePage.tsx b/frontend-web/src/position-roundtable/pages/PositionRoundtablePage.tsx index 91c774de..df40752e 100644 --- a/frontend-web/src/position-roundtable/pages/PositionRoundtablePage.tsx +++ b/frontend-web/src/position-roundtable/pages/PositionRoundtablePage.tsx @@ -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(null); + const handleReturnToEmbeddingParent = useCallback(() => { + returnToEmbeddingParent(embeddingParentReturnRoute, embeddingBridgePortRef.current); + }, [embeddingParentReturnRoute]); + useEffect(() => { + const receiveBridgePort = (event: MessageEvent) => { + 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(null); const positionSessionRef = useRef(null); const sessionCreationRef = useRef | null>(null); @@ -2273,6 +2444,17 @@ export default function PositionRoundtablePage({
+ {isEmbeddingContext && ( + } + onClick={handleReturnToEmbeddingParent} + > + 返回 + + )} )} @@ -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({ @@ -4307,16 +4475,7 @@ function BusinessChainFlow({ { - 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)` }} > - 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" > - + {stageIndex > 0 && ( + + )}
{stage.seats.map((seat, seatIndex) => { const node = nodes.find( @@ -4499,38 +4636,6 @@ function ChainFlowCanvas({
); })} - - onSelectBuiltin("summary")} - /> - - onSelectBuiltin("action_plan")} - />
); diff --git a/frontend-web/src/roundtable-planning/pages/BusinessChainEditorPage.tsx b/frontend-web/src/roundtable-planning/pages/BusinessChainEditorPage.tsx index 2d4c3bc3..5c711392 100644 --- a/frontend-web/src/roundtable-planning/pages/BusinessChainEditorPage.tsx +++ b/frontend-web/src/roundtable-planning/pages/BusinessChainEditorPage.tsx @@ -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( + () => + 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 (
({ - label: position.name, - value: position.id, - })), - ]} + options={selectablePositionOptions} onChange={(value) => onPositionChange(value ? (String(value) as PositionId) : null) } diff --git a/frontend-web/src/strategy-components/api/position-roles.ts b/frontend-web/src/strategy-components/api/position-roles.ts index 67a03aa6..531f9b7a 100644 --- a/frontend-web/src/strategy-components/api/position-roles.ts +++ b/frontend-web/src/strategy-components/api/position-roles.ts @@ -158,12 +158,21 @@ export async function replacePositionRoles(roles: PositionRole[]): Promise` loopback. Only the recovery -snapshot layer is position-specific. Deployments must expose +deployment-sensitive `127.0.0.1:` 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`. diff --git a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py b/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py index f1a06b5c..1ef19359 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py +++ b/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py @@ -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 diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/base.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/base.py index 9d8f4560..a45afe33 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/base.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/base.py @@ -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 diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/sql.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/sql.py index ae749047..bb2a93b0 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/sql.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/sql.py @@ -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( diff --git a/offline-backend-20260512/backend/tests/test_position_roundtable_sessions.py b/offline-backend-20260512/backend/tests/test_position_roundtable_sessions.py index f39bfc2d..bebbd60d 100644 --- a/offline-backend-20260512/backend/tests/test_position_roundtable_sessions.py +++ b/offline-backend-20260512/backend/tests/test_position_roundtable_sessions.py @@ -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] From 37964d1f9568a001b4481d0fffc2edf7d1262f8c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Fri, 31 Jul 2026 21:09:54 +0800 Subject: [PATCH 2/2] =?UTF-8?q?=E4=BB=A3=E7=A0=81=E4=BF=AE=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tmp/wujie-repro.html | 89 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 89 insertions(+) create mode 100644 tmp/wujie-repro.html diff --git a/tmp/wujie-repro.html b/tmp/wujie-repro.html new file mode 100644 index 00000000..c5e1105e --- /dev/null +++ b/tmp/wujie-repro.html @@ -0,0 +1,89 @@ + + + + + + code-zk · Wujie 返回跳转复现 + + + +
+ code-zk · Wujie 返回跳转复现 + 子应用：localhost:5175/chMoreMode + + 准备加载 +
+
正在加载 Wujie 子应用…
+ + +