Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 4b2af3d34a |
@@ -86,19 +86,11 @@ const MIN_HEIGHT = 400;
|
|||||||
// Margin kept between the window and the viewport edges while dragging.
|
// Margin kept between the window and the viewport edges while dragging.
|
||||||
const EDGE_MARGIN = 8;
|
const EDGE_MARGIN = 8;
|
||||||
|
|
||||||
// #184 phase 1.5 / #430: backstop for the degraded-poll fallback. The poll is
|
// #184 phase 1.5: hard cap on the degraded-poll fallback. The poll is armed when
|
||||||
// armed when a resume attempt could not attach to the live run and disarmed by the
|
// a resume attempt could not attach to the live run and disarmed by the thread on
|
||||||
// thread on settle / local stream; this cap is the ONLY backstop against an endless
|
// settle / local stream; this cap is the ONLY backstop against an endless tick
|
||||||
// tick (a stuck 'streaming' row before the boot-sweep, or a user-tail 204 with no
|
// (a stuck 'streaming' row before the boot-sweep, or a user-tail 204 with no run).
|
||||||
// run).
|
const DEGRADED_POLL_MAX_MS = 10 * 60_000;
|
||||||
//
|
|
||||||
// #430: measured from RUN ACTIVITY, not from arm-time. A real autonomous run takes
|
|
||||||
// 11-25 min — longer than a fixed 10-min-from-start cap, which used to cut the poll
|
|
||||||
// off mid-run. Instead we cap on INACTIVITY: keep polling as long as the run is
|
|
||||||
// still making progress (its persisted rows keep changing), and only give up after
|
|
||||||
// this long with NO new activity. A genuinely stuck run produces no row changes, so
|
|
||||||
// the idle cap still bounds it; a long-but-progressing run polls to completion.
|
|
||||||
const DEGRADED_POLL_IDLE_MAX_MS = 10 * 60_000;
|
|
||||||
|
|
||||||
/** Compact token formatter: 1.2M / 3.4k / 950. */
|
/** Compact token formatter: 1.2M / 3.4k / 950. */
|
||||||
function formatTokens(n: number): string {
|
function formatTokens(n: number): string {
|
||||||
@@ -262,12 +254,9 @@ export default function AiChatWindow() {
|
|||||||
// onResumeFallback(true); the thread disarms it on settle / local stream. The
|
// onResumeFallback(true); the thread disarms it on settle / local stream. The
|
||||||
// window only OWNS the timer (armedAtRef stamps when it was armed for the cap).
|
// window only OWNS the timer (armedAtRef stamps when it was armed for the cap).
|
||||||
const [degradedPoll, setDegradedPoll] = useState(false);
|
const [degradedPoll, setDegradedPoll] = useState(false);
|
||||||
// #430: timestamp of the LAST run activity while the poll is armed — stamped on
|
const armedAtRef = useRef(0);
|
||||||
// arm and re-stamped whenever the polled rows change (see the effect below). The
|
|
||||||
// idle cap is measured from this, so a long-but-progressing run keeps polling.
|
|
||||||
const lastActivityAtRef = useRef(0);
|
|
||||||
const onResumeFallback = useCallback((active: boolean): void => {
|
const onResumeFallback = useCallback((active: boolean): void => {
|
||||||
if (active) lastActivityAtRef.current = Date.now();
|
if (active) armedAtRef.current = Date.now();
|
||||||
setDegradedPoll(active);
|
setDegradedPoll(active);
|
||||||
}, []);
|
}, []);
|
||||||
// Reset the degraded poll whenever the open chat changes: it is scoped to the
|
// Reset the degraded poll whenever the open chat changes: it is scoped to the
|
||||||
@@ -280,28 +269,18 @@ export default function AiChatWindow() {
|
|||||||
useAiChatMessagesQuery(
|
useAiChatMessagesQuery(
|
||||||
activeChatId ?? undefined,
|
activeChatId ?? undefined,
|
||||||
// DELIBERATELY DUMB (invariant 8 / task 2.4): poll every 2.5s while armed
|
// DELIBERATELY DUMB (invariant 8 / task 2.4): poll every 2.5s while armed
|
||||||
// and while the run is still active (#430: under the INACTIVITY cap, not a
|
// and under the 10-min cap; otherwise off. NO error checks (TanStack v5
|
||||||
// fixed-from-start cap); otherwise off. NO error checks (TanStack v5 resets
|
// resets fetchFailureCount each fetch, so consecutive errors are not
|
||||||
// fetchFailureCount each fetch, so consecutive errors are not expressible —
|
// expressible — and the poll must survive a server restart) and NO tail
|
||||||
// and the poll must survive a server restart) and NO tail checks (the
|
// checks (the settled/local-stream semantics live in ChatThread, which
|
||||||
// settled/local-stream semantics live in ChatThread, which disarms via
|
// disarms via onResumeFallback(false)). The time cap is the only backstop.
|
||||||
// onResumeFallback(false)). The idle cap is the only backstop.
|
|
||||||
() =>
|
() =>
|
||||||
degradedPoll === true &&
|
degradedPoll === true &&
|
||||||
Date.now() - lastActivityAtRef.current < DEGRADED_POLL_IDLE_MAX_MS
|
Date.now() - armedAtRef.current < DEGRADED_POLL_MAX_MS
|
||||||
? 2500
|
? 2500
|
||||||
: false,
|
: false,
|
||||||
);
|
);
|
||||||
|
|
||||||
// #430: re-stamp the activity clock whenever the polled rows change while the
|
|
||||||
// poll is armed. TanStack keeps the same `messageRows` reference across refetches
|
|
||||||
// that return deep-equal data (structural sharing), so a new reference means the
|
|
||||||
// run genuinely progressed — which extends the inactivity cap above. A stuck run
|
|
||||||
// yields no reference change, so the cap eventually fires and stops the poll.
|
|
||||||
useEffect(() => {
|
|
||||||
if (degradedPoll) lastActivityAtRef.current = Date.now();
|
|
||||||
}, [degradedPoll, messageRows]);
|
|
||||||
|
|
||||||
// #184 reconnect-and-live-follow. Whether detached agent runs are enabled for
|
// #184 reconnect-and-live-follow. Whether detached agent runs are enabled for
|
||||||
// this workspace. When the feature is off no runs are ever created, so the
|
// this workspace. When the feature is off no runs are ever created, so the
|
||||||
// resume attempt would only ever 204; gating ChatThread's resume on it avoids a
|
// resume attempt would only ever 204; gating ChatThread's resume on it avoids a
|
||||||
|
|||||||
@@ -739,170 +739,3 @@ function renderResumable(initialRows: IAiChatMessageRow[]) {
|
|||||||
act(() => view.rerender(<Wrapper rows={rows} />));
|
act(() => view.rerender(<Wrapper rows={rows} />));
|
||||||
return { rerender, onResumeFallback };
|
return { rerender, onResumeFallback };
|
||||||
}
|
}
|
||||||
|
|
||||||
// #430: auto-reconnect to a DETACHED run after a LIVE SSE disconnect. The mount
|
|
||||||
// path only resumes on mount/reload; these cover the missing trigger — a live
|
|
||||||
// `isDisconnect` on onFinish must (backoff-)re-attach WITHOUT a reload, pin+strip
|
|
||||||
// the live row to avoid duplicates, fall back to the degraded poll on a 204, and
|
|
||||||
// exhaust to a manual Retry.
|
|
||||||
describe("ChatThread — live reconnect after isDisconnect (#430)", () => {
|
|
||||||
// A LIVE local turn that just dropped: the settled tail existed before, and the
|
|
||||||
// partial assistant row lives only in `messages` (not persisted as a tail).
|
|
||||||
const settledTail = () => [
|
|
||||||
row("u1", "user", undefined, "hi"),
|
|
||||||
row("a1", "assistant", "succeeded", "done"),
|
|
||||||
];
|
|
||||||
// The partial assistant message onFinish hands us for the dropped LIVE turn.
|
|
||||||
const liveMsg = {
|
|
||||||
id: "a2",
|
|
||||||
role: "assistant",
|
|
||||||
parts: [{ type: "text", text: "partial live answer" }],
|
|
||||||
};
|
|
||||||
|
|
||||||
beforeEach(() => {
|
|
||||||
resetState();
|
|
||||||
// status "ready": with a live disconnect the mock is not streaming, so the
|
|
||||||
// status==="streaming" auto-clear effect stays out of the way.
|
|
||||||
h.state.status = "ready";
|
|
||||||
vi.useFakeTimers();
|
|
||||||
});
|
|
||||||
afterEach(() => {
|
|
||||||
vi.useRealTimers();
|
|
||||||
cleanup();
|
|
||||||
});
|
|
||||||
|
|
||||||
// Render a NON-resuming mount (settled tail -> no mount resume) with autonomous
|
|
||||||
// runs on, then simulate a live disconnect via onFinish.
|
|
||||||
function renderLiveThenDisconnect() {
|
|
||||||
const view = renderThread({
|
|
||||||
autonomousRunsEnabled: true,
|
|
||||||
initialRows: settledTail(),
|
|
||||||
});
|
|
||||||
// The settled tail must NOT have triggered a mount resume.
|
|
||||||
expect(h.state.resumeStream).not.toHaveBeenCalled();
|
|
||||||
act(() => {
|
|
||||||
h.state.onFinish?.({
|
|
||||||
message: liveMsg,
|
|
||||||
isAbort: false,
|
|
||||||
isDisconnect: true,
|
|
||||||
isError: false,
|
|
||||||
});
|
|
||||||
});
|
|
||||||
return view;
|
|
||||||
}
|
|
||||||
|
|
||||||
// Fire the pending (scheduled) attempt for `attempt` (backoff = 1s,2s,4s,...).
|
|
||||||
function advanceToAttempt(attempt: number) {
|
|
||||||
act(() => {
|
|
||||||
vi.advanceTimersByTime(1000 * 2 ** (attempt - 1));
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
// Simulate the reconnect GET returning 204 (nothing live) so the transport's
|
|
||||||
// no-active-stream recovery runs.
|
|
||||||
async function reconnect204() {
|
|
||||||
vi.stubGlobal(
|
|
||||||
"fetch",
|
|
||||||
vi.fn().mockResolvedValue({ status: 204, ok: false }),
|
|
||||||
);
|
|
||||||
await act(async () => {
|
|
||||||
await h.state.transport!.fetch!("http://x", { method: "GET" });
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
// Simulate the reconnect GET returning a live 2xx stream.
|
|
||||||
async function reconnect200() {
|
|
||||||
vi.stubGlobal(
|
|
||||||
"fetch",
|
|
||||||
vi.fn().mockResolvedValue({ status: 200, ok: true }),
|
|
||||||
);
|
|
||||||
await act(async () => {
|
|
||||||
await h.state.transport!.fetch!("http://x", { method: "GET" });
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
it("calls resumeStream POST-mount (a live disconnect triggers a backoff reconnect)", () => {
|
|
||||||
renderLiveThenDisconnect();
|
|
||||||
// The banner shows immediately; the attach itself fires after the first backoff.
|
|
||||||
expect(screen.getByText(/reconnecting/i)).toBeTruthy();
|
|
||||||
expect(h.state.resumeStream).not.toHaveBeenCalled();
|
|
||||||
advanceToAttempt(1);
|
|
||||||
// resumeStream is now called AFTER mount — the bug was it only ever fired once
|
|
||||||
// on mount. The reconnect URL pins expect=live&anchor to OUR run.
|
|
||||||
expect(h.state.resumeStream).toHaveBeenCalledTimes(1);
|
|
||||||
expect(h.state.transport!.prepareReconnectToStreamRequest!().api).toBe(
|
|
||||||
"/api/ai-chat/runs/c1/stream?expect=live&anchor=a2",
|
|
||||||
);
|
|
||||||
});
|
|
||||||
|
|
||||||
it("strips the pinned live row before replay so content is NOT duplicated", () => {
|
|
||||||
renderLiveThenDisconnect();
|
|
||||||
advanceToAttempt(1);
|
|
||||||
// The attempt strips the anchor row from the store (the live replay rebuilds
|
|
||||||
// it). Apply the setMessages updater to prove it removes exactly the anchor.
|
|
||||||
const updater = h.state.setMessages.mock.calls.at(-1)![0] as (
|
|
||||||
prev: { id: string }[],
|
|
||||||
) => { id: string }[];
|
|
||||||
expect(updater([{ id: "u1" }, { id: "a2" }])).toEqual([{ id: "u1" }]);
|
|
||||||
});
|
|
||||||
|
|
||||||
it("a live re-attach (2xx) clears the reconnect banner", async () => {
|
|
||||||
renderLiveThenDisconnect();
|
|
||||||
advanceToAttempt(1);
|
|
||||||
await reconnect200();
|
|
||||||
expect(screen.queryByText(/reconnecting/i)).toBeNull();
|
|
||||||
});
|
|
||||||
|
|
||||||
it("a 204 arms the degraded poll and backs off to the next attempt", async () => {
|
|
||||||
const { onResumeFallback } = renderLiveThenDisconnect();
|
|
||||||
advanceToAttempt(1);
|
|
||||||
expect(h.state.resumeStream).toHaveBeenCalledTimes(1);
|
|
||||||
await reconnect204();
|
|
||||||
// Fallback engaged: the degraded poll is armed (204 -> onNoActiveStream).
|
|
||||||
expect(onResumeFallback).toHaveBeenCalledWith(true);
|
|
||||||
// Still reconnecting — the banner advanced to attempt 2/5.
|
|
||||||
expect(screen.getByText(/reconnecting.*2\/5/i)).toBeTruthy();
|
|
||||||
// The next backoff fires attempt 2 (another resumeStream).
|
|
||||||
advanceToAttempt(2);
|
|
||||||
expect(h.state.resumeStream).toHaveBeenCalledTimes(2);
|
|
||||||
});
|
|
||||||
|
|
||||||
it("exhausts the attempt limit into a manual Retry, which restarts the sequence", async () => {
|
|
||||||
renderLiveThenDisconnect();
|
|
||||||
// Drive all 5 attempts, each failing with a 204.
|
|
||||||
for (let n = 1; n <= 5; n++) {
|
|
||||||
advanceToAttempt(n);
|
|
||||||
expect(h.state.resumeStream).toHaveBeenCalledTimes(n);
|
|
||||||
await reconnect204();
|
|
||||||
}
|
|
||||||
// The 5th 204 exhausted the cap -> the manual Retry replaces the banner.
|
|
||||||
expect(screen.queryByText(/reconnecting/i)).toBeNull();
|
|
||||||
const retry = screen.getByText("Retry");
|
|
||||||
expect(retry).toBeTruthy();
|
|
||||||
// Retry fires attempt 1 immediately (no backoff) — a 6th resumeStream.
|
|
||||||
act(() => {
|
|
||||||
fireEvent.click(retry);
|
|
||||||
});
|
|
||||||
expect(h.state.resumeStream).toHaveBeenCalledTimes(6);
|
|
||||||
expect(screen.getByText(/reconnecting/i)).toBeTruthy();
|
|
||||||
});
|
|
||||||
|
|
||||||
it("does NOT reconnect when autonomous runs are disabled", () => {
|
|
||||||
renderThread({ autonomousRunsEnabled: false, initialRows: settledTail() });
|
|
||||||
act(() => {
|
|
||||||
h.state.onFinish?.({
|
|
||||||
message: liveMsg,
|
|
||||||
isAbort: false,
|
|
||||||
isDisconnect: true,
|
|
||||||
isError: false,
|
|
||||||
});
|
|
||||||
});
|
|
||||||
expect(screen.queryByText(/reconnecting/i)).toBeNull();
|
|
||||||
// The terminal "connection lost" notice is shown instead (unchanged behavior).
|
|
||||||
expect(
|
|
||||||
screen.getByText("Connection lost — the answer was interrupted."),
|
|
||||||
).toBeTruthy();
|
|
||||||
advanceToAttempt(1);
|
|
||||||
expect(h.state.resumeStream).not.toHaveBeenCalled();
|
|
||||||
});
|
|
||||||
});
|
|
||||||
|
|||||||
@@ -1,17 +1,7 @@
|
|||||||
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
|
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
|
||||||
import { useQueryClient } from "@tanstack/react-query";
|
import { useQueryClient } from "@tanstack/react-query";
|
||||||
import { generateId } from "ai";
|
import { generateId } from "ai";
|
||||||
import {
|
import { ActionIcon, Box, Group, Stack, Text, Tooltip } from "@mantine/core";
|
||||||
ActionIcon,
|
|
||||||
Alert,
|
|
||||||
Box,
|
|
||||||
Button,
|
|
||||||
Group,
|
|
||||||
Loader,
|
|
||||||
Stack,
|
|
||||||
Text,
|
|
||||||
Tooltip,
|
|
||||||
} from "@mantine/core";
|
|
||||||
import {
|
import {
|
||||||
IconClockHour4,
|
IconClockHour4,
|
||||||
IconPlayerPlayFilled,
|
IconPlayerPlayFilled,
|
||||||
@@ -61,15 +51,6 @@ import classes from "@/features/ai-chat/components/ai-chat.module.css";
|
|||||||
// from the token rate.
|
// from the token rate.
|
||||||
const STREAM_THROTTLE_MS = 50;
|
const STREAM_THROTTLE_MS = 50;
|
||||||
|
|
||||||
// #430: auto-reconnect after a LIVE SSE disconnect of a DETACHED (autonomous) run.
|
|
||||||
// The run keeps executing server-side, so instead of a dead "Lost connection"
|
|
||||||
// banner we re-attach to the live tail through the SAME resumable machinery the
|
|
||||||
// mount path uses. Attempts back off exponentially and are capped; on exhaustion
|
|
||||||
// the user gets a manual Retry (the degraded poll keeps catching up underneath).
|
|
||||||
const RECONNECT_MAX_ATTEMPTS = 5;
|
|
||||||
// Backoff before attempt N (1-based): 1s, 2s, 4s, 8s, 16s.
|
|
||||||
const RECONNECT_BASE_DELAY_MS = 1000;
|
|
||||||
|
|
||||||
/** The page the user is currently viewing, sent as chat context. */
|
/** The page the user is currently viewing, sent as chat context. */
|
||||||
export interface OpenPageContext {
|
export interface OpenPageContext {
|
||||||
id: string;
|
id: string;
|
||||||
@@ -194,10 +175,6 @@ export default function ChatThread({
|
|||||||
const reconcileTailRef = useRef(false);
|
const reconcileTailRef = useRef(false);
|
||||||
const noStreamHandledRef = useRef(false);
|
const noStreamHandledRef = useRef(false);
|
||||||
const onNoActiveStreamRef = useRef<(() => void) | null>(null);
|
const onNoActiveStreamRef = useRef<(() => void) | null>(null);
|
||||||
// #430: called from the transport's reconnect-GET success branch when a live
|
|
||||||
// stream re-attached (2xx, not 204) — clears the reconnect banner. Kept in a ref
|
|
||||||
// because the transport's fetch closure (useMemo([])) reads it live.
|
|
||||||
const onReconnectAttachedRef = useRef<(() => void) | null>(null);
|
|
||||||
// Live mount flag. The attach GET and the resumed `onFinish` are async and can
|
// Live mount flag. The attach GET and the resumed `onFinish` are async and can
|
||||||
// land AFTER this thread unmounts (the parent remounts per chat via `key`); with
|
// land AFTER this thread unmounts (the parent remounts per chat via `key`); with
|
||||||
// chatIdRef then pointing at the NEW chat, an ungated late callback would arm a
|
// chatIdRef then pointing at the NEW chat, an ungated late callback would arm a
|
||||||
@@ -401,10 +378,6 @@ export default function ChatThread({
|
|||||||
// NOT drop the in-progress row or stop tracking the durable run.
|
// NOT drop the in-progress row or stop tracking the durable run.
|
||||||
if (response.status === 204 || !response.ok)
|
if (response.status === 204 || !response.ok)
|
||||||
onNoActiveStreamRef.current?.();
|
onNoActiveStreamRef.current?.();
|
||||||
// #430: a 2xx stream re-attached (live tail or finished-replay). Signal
|
|
||||||
// the reconnect controller to clear its banner. No-op outside an active
|
|
||||||
// reconnect sequence (e.g. the mount attach), so it is safe here.
|
|
||||||
else onReconnectAttachedRef.current?.();
|
|
||||||
return response;
|
return response;
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
// Network throw: same no-onFinish recovery, then rethrow so the SDK
|
// Network throw: same no-onFinish recovery, then rethrow so the SDK
|
||||||
@@ -508,31 +481,6 @@ export default function ChatThread({
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// (2b) #430: a LIVE (non-resumed) detached run whose SSE just dropped. The
|
|
||||||
// server run keeps executing, so instead of a dead "Lost connection" banner
|
|
||||||
// start a reconnect sequence: pin the CURRENT streaming assistant row as the
|
|
||||||
// strip/anchor (the live tail is the already-shown partial in `messages`, not
|
|
||||||
// a persistent row) and re-attach to the live tail via the resumable machinery.
|
|
||||||
const startedReconnect =
|
|
||||||
isDisconnect &&
|
|
||||||
!wasResumed &&
|
|
||||||
autonomousRunsEnabled === true &&
|
|
||||||
mountedRef.current &&
|
|
||||||
message?.role === "assistant" &&
|
|
||||||
typeof message.id === "string";
|
|
||||||
if (startedReconnect) {
|
|
||||||
beginReconnect({
|
|
||||||
id: message.id,
|
|
||||||
role: "assistant",
|
|
||||||
content: "",
|
|
||||||
status: "streaming",
|
|
||||||
createdAt: new Date().toISOString(),
|
|
||||||
// Preserve the partial parts so a 204 restore (onNoActiveStream) re-shows
|
|
||||||
// what was on screen while the degraded poll catches the run up to
|
|
||||||
// terminal (rowToUiMessage prefers metadata.parts).
|
|
||||||
metadata: { parts: message.parts },
|
|
||||||
});
|
|
||||||
}
|
|
||||||
// (3) Standard branches.
|
// (3) Standard branches.
|
||||||
// Forward the authoritative server chatId (streamed on the assistant
|
// Forward the authoritative server chatId (streamed on the assistant
|
||||||
// message metadata) so the parent adopts the REAL created chat id for a new
|
// message metadata) so the parent adopts the REAL created chat id for a new
|
||||||
@@ -542,11 +490,9 @@ export default function ChatThread({
|
|||||||
onTurnFinished(extractServerChatId(message), threadKey);
|
onTurnFinished(extractServerChatId(message), threadKey);
|
||||||
// Show a neutral "stopped" marker for an aborted turn; the red error banner
|
// Show a neutral "stopped" marker for an aborted turn; the red error banner
|
||||||
// (via `error`) already covers isError, and a clean finish clears any marker.
|
// (via `error`) already covers isError, and a clean finish clears any marker.
|
||||||
// On a live disconnect that STARTED a reconnect, suppress the terminal
|
|
||||||
// "connection lost" notice — the reconnect banner takes over (#430).
|
|
||||||
if (isError) setStopNotice(null);
|
if (isError) setStopNotice(null);
|
||||||
else if (isAbort) setStopNotice("manual");
|
else if (isAbort) setStopNotice("manual");
|
||||||
else if (isDisconnect) setStopNotice(startedReconnect ? null : "disconnect");
|
else if (isDisconnect) setStopNotice("disconnect");
|
||||||
else setStopNotice(null);
|
else setStopNotice(null);
|
||||||
// A resumed turn NEVER flushes the queue (invariant 7): skip BOTH the
|
// A resumed turn NEVER flushes the queue (invariant 7): skip BOTH the
|
||||||
// flush-on-abort branch and the plain flush. The local streamer is the only
|
// flush-on-abort branch and the plain flush. The local streamer is the only
|
||||||
@@ -633,106 +579,6 @@ export default function ChatThread({
|
|||||||
|
|
||||||
const isStreaming = status === "submitted" || status === "streaming";
|
const isStreaming = status === "submitted" || status === "streaming";
|
||||||
|
|
||||||
// #430: live-disconnect reconnect controller. `null` = idle; `{ trying, attempt }`
|
|
||||||
// = a backoff sequence is running (drives the "reconnecting… (N/max)" banner);
|
|
||||||
// `{ failed }` = attempts exhausted (drives the manual Retry). Mirrored into a ref
|
|
||||||
// so the transport/onNoActiveStream closures branch on the LIVE value.
|
|
||||||
type ReconnectState =
|
|
||||||
| null
|
|
||||||
| { phase: "trying"; attempt: number }
|
|
||||||
| { phase: "failed" };
|
|
||||||
const [reconnectState, setReconnectState] = useState<ReconnectState>(null);
|
|
||||||
const reconnectStateRef = useRef<ReconnectState>(null);
|
|
||||||
const setReconnectStatePair = useCallback((s: ReconnectState) => {
|
|
||||||
reconnectStateRef.current = s;
|
|
||||||
setReconnectState(s);
|
|
||||||
}, []);
|
|
||||||
const reconnectTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null);
|
|
||||||
const clearReconnectTimer = useCallback(() => {
|
|
||||||
if (reconnectTimerRef.current) {
|
|
||||||
clearTimeout(reconnectTimerRef.current);
|
|
||||||
reconnectTimerRef.current = null;
|
|
||||||
}
|
|
||||||
}, []);
|
|
||||||
|
|
||||||
// One reconnect attempt — MIRRORS the mount strip/anchor path for the LIVE case.
|
|
||||||
// beginReconnect pinned strippedRowRef/stripRef to the run's assistant row, so:
|
|
||||||
// - remove that row from the store (the mount path strips it from the SEED; here
|
|
||||||
// it is already shown, so filter it out) — the live replay's `text-start` then
|
|
||||||
// rebuilds it without DUPLICATING parts (the main dedup risk, #430);
|
|
||||||
// - reset the one-shot 204 guard so onNoActiveStream can fire for THIS attempt;
|
|
||||||
// - mark the turn resumed (invariant 7/8) so onFinish runs the recovery block and
|
|
||||||
// never flushes the queue;
|
|
||||||
// - resumeStream() -> prepareReconnectToStreamRequest builds
|
|
||||||
// ?expect=live&anchor=<pinned id>, pinning the replay to OUR run (invariant 6).
|
|
||||||
const attemptReconnectOnce = useCallback(
|
|
||||||
(attempt: number) => {
|
|
||||||
if (!mountedRef.current) return;
|
|
||||||
const anchor = strippedRowRef.current;
|
|
||||||
if (anchor) {
|
|
||||||
setMessages((prev) => prev.filter((m) => m.id !== anchor.id));
|
|
||||||
}
|
|
||||||
noStreamHandledRef.current = false;
|
|
||||||
setResumedTurnPair(true);
|
|
||||||
setReconnectStatePair({ phase: "trying", attempt });
|
|
||||||
void resumeStream();
|
|
||||||
},
|
|
||||||
[setMessages, setResumedTurnPair, setReconnectStatePair, resumeStream],
|
|
||||||
);
|
|
||||||
|
|
||||||
// Schedule attempt `attempt` after an exponential backoff.
|
|
||||||
const scheduleReconnectAttempt = useCallback(
|
|
||||||
(attempt: number) => {
|
|
||||||
clearReconnectTimer();
|
|
||||||
setReconnectStatePair({ phase: "trying", attempt });
|
|
||||||
reconnectTimerRef.current = setTimeout(
|
|
||||||
() => attemptReconnectOnce(attempt),
|
|
||||||
RECONNECT_BASE_DELAY_MS * 2 ** (attempt - 1),
|
|
||||||
);
|
|
||||||
},
|
|
||||||
[clearReconnectTimer, setReconnectStatePair, attemptReconnectOnce],
|
|
||||||
);
|
|
||||||
|
|
||||||
// Start a fresh reconnect sequence, pinning `anchorRow` (the live run's assistant
|
|
||||||
// row) as the strip/anchor reused by every attempt.
|
|
||||||
const beginReconnect = useCallback(
|
|
||||||
(anchorRow: IAiChatMessageRow) => {
|
|
||||||
if (!autonomousRunsEnabled || !mountedRef.current) return;
|
|
||||||
strippedRowRef.current = anchorRow;
|
|
||||||
stripRef.current = true;
|
|
||||||
scheduleReconnectAttempt(1);
|
|
||||||
},
|
|
||||||
[autonomousRunsEnabled, scheduleReconnectAttempt],
|
|
||||||
);
|
|
||||||
|
|
||||||
// Manual Retry (shown once attempts are exhausted): restart at attempt 1 and fire
|
|
||||||
// immediately (the user asked for it now — no backoff).
|
|
||||||
const retryReconnect = useCallback(() => {
|
|
||||||
clearReconnectTimer();
|
|
||||||
attemptReconnectOnce(1);
|
|
||||||
}, [clearReconnectTimer, attemptReconnectOnce]);
|
|
||||||
|
|
||||||
// Live SSE re-attached (the reconnect GET returned a 2xx stream): clear the
|
|
||||||
// banner + any pending backoff. No-op outside a sequence (e.g. the mount attach).
|
|
||||||
const onReconnectAttached = useCallback(() => {
|
|
||||||
if (!mountedRef.current || !reconnectStateRef.current) return;
|
|
||||||
clearReconnectTimer();
|
|
||||||
setReconnectStatePair(null);
|
|
||||||
}, [clearReconnectTimer, setReconnectStatePair]);
|
|
||||||
onReconnectAttachedRef.current = onReconnectAttached;
|
|
||||||
|
|
||||||
// The reconnect GET could not attach (204 / error). onNoActiveStream has already
|
|
||||||
// armed the degraded poll (the robust fallback that drives the row to terminal
|
|
||||||
// from the DB), so this only decides the LIVE-attach retry: back off and try
|
|
||||||
// again up to the cap, else surface the manual Retry.
|
|
||||||
const onReconnectNoStream = useCallback(() => {
|
|
||||||
const s = reconnectStateRef.current;
|
|
||||||
if (s?.phase !== "trying") return;
|
|
||||||
if (s.attempt < RECONNECT_MAX_ATTEMPTS)
|
|
||||||
scheduleReconnectAttempt(s.attempt + 1);
|
|
||||||
else setReconnectStatePair({ phase: "failed" });
|
|
||||||
}, [scheduleReconnectAttempt, setReconnectStatePair]);
|
|
||||||
|
|
||||||
// 204-handler (`onNoActiveStream`): the attach returned 204 — nothing live to
|
// 204-handler (`onNoActiveStream`): the attach returned 204 — nothing live to
|
||||||
// resume (overflow / begin-failure / after retention / anchor-mismatch). One-
|
// resume (overflow / begin-failure / after retention / anchor-mismatch). One-
|
||||||
// shot via noStreamHandledRef (we do NOT null onNoActiveStreamRef). Exactly four
|
// shot via noStreamHandledRef (we do NOT null onNoActiveStreamRef). Exactly four
|
||||||
@@ -764,17 +610,7 @@ export default function ChatThread({
|
|||||||
// (d) 204 means onFinish will NOT fire — clear the suppression flag so it
|
// (d) 204 means onFinish will NOT fire — clear the suppression flag so it
|
||||||
// cannot swallow the NEXT local turn's queue flush.
|
// cannot swallow the NEXT local turn's queue flush.
|
||||||
setResumedTurnPair(false);
|
setResumedTurnPair(false);
|
||||||
// (e) #430: if this 204/error landed during a live-disconnect reconnect
|
}, [setMessages, queryClient, onResumeFallback, setResumedTurnPair]);
|
||||||
// sequence, back off and retry the live attach (or give up to the manual
|
|
||||||
// Retry). The degraded poll armed in (c) is the fallback either way.
|
|
||||||
onReconnectNoStream();
|
|
||||||
}, [
|
|
||||||
setMessages,
|
|
||||||
queryClient,
|
|
||||||
onResumeFallback,
|
|
||||||
setResumedTurnPair,
|
|
||||||
onReconnectNoStream,
|
|
||||||
]);
|
|
||||||
onNoActiveStreamRef.current = onNoActiveStream;
|
onNoActiveStreamRef.current = onNoActiveStream;
|
||||||
|
|
||||||
// Mount effect: kick off the resume attempt for a non-settled tail. Marking the
|
// Mount effect: kick off the resume attempt for a non-settled tail. Marking the
|
||||||
@@ -792,9 +628,6 @@ export default function ChatThread({
|
|||||||
return () => {
|
return () => {
|
||||||
mountedRef.current = false;
|
mountedRef.current = false;
|
||||||
attachAbortRef.current?.abort();
|
attachAbortRef.current?.abort();
|
||||||
// #430: drop any pending reconnect backoff so it can't fire against the next
|
|
||||||
// chat this thread's refs are reused for.
|
|
||||||
if (reconnectTimerRef.current) clearTimeout(reconnectTimerRef.current);
|
|
||||||
};
|
};
|
||||||
// Mount-only by design; the parent remounts per chat via `key`.
|
// Mount-only by design; the parent remounts per chat via `key`.
|
||||||
// eslint-disable-next-line react-hooks/exhaustive-deps
|
// eslint-disable-next-line react-hooks/exhaustive-deps
|
||||||
@@ -833,27 +666,12 @@ export default function ChatThread({
|
|||||||
if (tail.status !== "streaming") {
|
if (tail.status !== "streaming") {
|
||||||
reconcileTailRef.current = false;
|
reconcileTailRef.current = false;
|
||||||
onResumeFallback?.(false);
|
onResumeFallback?.(false);
|
||||||
// #430: the run reached its terminal state via the degraded poll — there is
|
|
||||||
// no live tail left to reconnect to, so drop any reconnect banner / Retry.
|
|
||||||
clearReconnectTimer();
|
|
||||||
setReconnectStatePair(null);
|
|
||||||
}
|
}
|
||||||
// onResumeFallback intentionally omitted (parent-stable callback); deps are
|
// onResumeFallback intentionally omitted (parent-stable callback); deps are
|
||||||
// fixed by the resume design.
|
// fixed by the resume design.
|
||||||
// eslint-disable-next-line react-hooks/exhaustive-deps
|
// eslint-disable-next-line react-hooks/exhaustive-deps
|
||||||
}, [initialRows, isStreaming, setMessages]);
|
}, [initialRows, isStreaming, setMessages]);
|
||||||
|
|
||||||
// #430: a real stream is live again — the reconnect re-attached to the live tail
|
|
||||||
// (status -> "streaming") OR the user started a new local turn. Either way clear
|
|
||||||
// the reconnect banner + any pending backoff. Gated on "streaming" (not the
|
|
||||||
// broader "submitted") so a still-pending attach GET does not clear prematurely.
|
|
||||||
useEffect(() => {
|
|
||||||
if (status === "streaming") {
|
|
||||||
clearReconnectTimer();
|
|
||||||
setReconnectStatePair(null);
|
|
||||||
}
|
|
||||||
}, [status, clearReconnectTimer, setReconnectStatePair]);
|
|
||||||
|
|
||||||
// "Send now" on a queued message: interrupt the current turn and immediately
|
// "Send now" on a queued message: interrupt the current turn and immediately
|
||||||
// send THIS message, keeping the agent's partial output. Other queued messages
|
// send THIS message, keeping the agent's partial output. Other queued messages
|
||||||
// stay queued and flush normally after the new turn. Reuses the existing
|
// stay queued and flush normally after the new turn. Reuses the existing
|
||||||
@@ -901,9 +719,6 @@ export default function ChatThread({
|
|||||||
// observer's Stop would otherwise leave the attach fetch running.
|
// observer's Stop would otherwise leave the attach fetch running.
|
||||||
attachAbortRef.current?.abort();
|
attachAbortRef.current?.abort();
|
||||||
stop();
|
stop();
|
||||||
// #430: pressing Stop also cancels an in-progress reconnect sequence.
|
|
||||||
clearReconnectTimer();
|
|
||||||
setReconnectStatePair(null);
|
|
||||||
if (!autonomousRunsEnabled) return;
|
if (!autonomousRunsEnabled) return;
|
||||||
if (chatIdRef.current) {
|
if (chatIdRef.current) {
|
||||||
onServerStop?.(chatIdRef.current);
|
onServerStop?.(chatIdRef.current);
|
||||||
@@ -925,13 +740,7 @@ export default function ChatThread({
|
|||||||
// for this fix. Documented so a future change can address the abort-ordering.
|
// for this fix. Documented so a future change can address the abort-ordering.
|
||||||
stopPendingRef.current = true;
|
stopPendingRef.current = true;
|
||||||
}
|
}
|
||||||
}, [
|
}, [stop, autonomousRunsEnabled, onServerStop]);
|
||||||
stop,
|
|
||||||
autonomousRunsEnabled,
|
|
||||||
onServerStop,
|
|
||||||
clearReconnectTimer,
|
|
||||||
setReconnectStatePair,
|
|
||||||
]);
|
|
||||||
|
|
||||||
// Clear the stopped marker as soon as a new turn begins streaming, and drop any
|
// Clear the stopped marker as soon as a new turn begins streaming, and drop any
|
||||||
// stale "Send now" interrupt flags. On the legit interrupt path both refs are
|
// stale "Send now" interrupt flags. On the legit interrupt path both refs are
|
||||||
@@ -1016,43 +825,6 @@ export default function ChatThread({
|
|||||||
detail={errorView.detail}
|
detail={errorView.detail}
|
||||||
mb="xs"
|
mb="xs"
|
||||||
/>
|
/>
|
||||||
) : reconnectState ? (
|
|
||||||
// #430: while auto-reconnecting to a detached run's live tail, show progress
|
|
||||||
// instead of a dead "Lost connection" banner; once attempts are exhausted,
|
|
||||||
// offer a manual Retry (the degraded poll keeps catching up underneath).
|
|
||||||
<Alert
|
|
||||||
variant="light"
|
|
||||||
color="gray"
|
|
||||||
p="xs"
|
|
||||||
mb="xs"
|
|
||||||
style={{ flexShrink: 0 }}
|
|
||||||
>
|
|
||||||
<Group gap={8} wrap="nowrap" align="center">
|
|
||||||
{reconnectState.phase === "trying" ? (
|
|
||||||
<>
|
|
||||||
<Loader size={14} color="gray" style={{ flex: "none" }} />
|
|
||||||
<Text size="sm" lh={1.3} c="dimmed">
|
|
||||||
{t("Connection lost — reconnecting…")}
|
|
||||||
{` (${reconnectState.attempt}/${RECONNECT_MAX_ATTEMPTS})`}
|
|
||||||
</Text>
|
|
||||||
</>
|
|
||||||
) : (
|
|
||||||
<>
|
|
||||||
<Text size="sm" lh={1.3} c="dimmed" style={{ flex: 1 }}>
|
|
||||||
{t("Couldn't reconnect to the answer.")}
|
|
||||||
</Text>
|
|
||||||
<Button
|
|
||||||
size="compact-xs"
|
|
||||||
variant="light"
|
|
||||||
color="gray"
|
|
||||||
onClick={retryReconnect}
|
|
||||||
>
|
|
||||||
{t("Retry")}
|
|
||||||
</Button>
|
|
||||||
</>
|
|
||||||
)}
|
|
||||||
</Group>
|
|
||||||
</Alert>
|
|
||||||
) : stopNotice ? (
|
) : stopNotice ? (
|
||||||
<ChatStoppedNotice
|
<ChatStoppedNotice
|
||||||
text={
|
text={
|
||||||
|
|||||||
@@ -49,14 +49,19 @@ export default function FootnoteDefinitionView(props: NodeViewProps) {
|
|||||||
className={classes.definition}
|
className={classes.definition}
|
||||||
style={{ ["--footnote-number" as any]: `"${number}"` }}
|
style={{ ["--footnote-number" as any]: `"${number}"` }}
|
||||||
>
|
>
|
||||||
{/* #146: contentDOM MUST be the first child — non-editable chrome before
|
{/* #146: contentDOM MUST be the first child — a non-editable marker before
|
||||||
it makes click hit-testing snap the caret above. Content first; the
|
it makes click hit-testing snap the caret above. Content first; the
|
||||||
back-link follows in DOM and is placed on the right via CSS flex. The
|
marker + back-link follow in DOM and are placed left/right via CSS
|
||||||
decorative "N." number is rendered inline via the .definitionContent
|
flex `order`. The second #146 mitigation lives in
|
||||||
::before rule (from the --footnote-number var), so no marker element
|
|
||||||
precedes the content. The second #146 mitigation lives in
|
|
||||||
editor-paste-handler.tsx (reflowAfterPaste). */}
|
editor-paste-handler.tsx (reflowAfterPaste). */}
|
||||||
<NodeViewContent className={classes.definitionContent} />
|
<NodeViewContent className={classes.definitionContent} />
|
||||||
|
<span
|
||||||
|
className={classes.definitionMarker}
|
||||||
|
contentEditable={false}
|
||||||
|
aria-hidden="true"
|
||||||
|
>
|
||||||
|
{number}.
|
||||||
|
</span>
|
||||||
{refCount > 1 ? (
|
{refCount > 1 ? (
|
||||||
// Multiple references -> ↩ followed by one lettered link per occurrence.
|
// Multiple references -> ↩ followed by one lettered link per occurrence.
|
||||||
<span
|
<span
|
||||||
|
|||||||
@@ -81,34 +81,34 @@
|
|||||||
.definition {
|
.definition {
|
||||||
display: flex;
|
display: flex;
|
||||||
align-items: flex-start;
|
align-items: flex-start;
|
||||||
/* Tight spacing between the content and the trailing ↩ back-link. */
|
/* Tight number→text spacing (~one space) so it reads like "1. text"
|
||||||
gap: 0.3em;
|
instead of leaving a wide gap after the period. */
|
||||||
|
gap: 0.4em;
|
||||||
padding: 2px 0;
|
padding: 2px 0;
|
||||||
/* Footnotes read smaller than body text (16px). Matches .listHeading. */
|
|
||||||
font-size: var(--mantine-font-size-sm);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/* The "N." number is decorative (from the --footnote-number CSS var on the
|
.definitionMarker {
|
||||||
wrapper, never in the document model) and is rendered inline at the start of
|
order: -1; /* keep the "N." marker on the LEFT though it follows content in DOM (#146) */
|
||||||
the first content line via ::before. This keeps text and wrapped lines flush
|
flex: 0 0 auto;
|
||||||
to the left margin — no hanging indent — while the editable contentDOM stays
|
min-width: 1.5em;
|
||||||
the FIRST DOM child (#146). */
|
/* Right-align within the narrow column so the period sits next to the text
|
||||||
|
and multi-digit numbers (10, 11, …) stay aligned on their right edge. */
|
||||||
|
text-align: right;
|
||||||
|
font-variant-numeric: tabular-nums;
|
||||||
|
color: var(--mantine-color-dimmed);
|
||||||
|
user-select: none;
|
||||||
|
}
|
||||||
|
|
||||||
.definitionContent {
|
.definitionContent {
|
||||||
flex: 1 1 auto;
|
flex: 1 1 auto;
|
||||||
min-width: 0;
|
min-width: 0;
|
||||||
}
|
}
|
||||||
|
|
||||||
.definitionContent > :first-child::before {
|
/* The inner editable paragraph inherits `.ProseMirror p { margin: 0.5em 0 }`,
|
||||||
content: var(--footnote-number, "?") ". ";
|
which pushes the first text line ~0.5em below the "N." marker (aligned to
|
||||||
color: var(--mantine-color-dimmed);
|
flex-start), making the number float above the text. Drop the outer margins
|
||||||
font-variant-numeric: tabular-nums;
|
so the marker and the first line share the same top edge — same approach
|
||||||
user-select: none;
|
used for callouts in core.css. */
|
||||||
}
|
|
||||||
|
|
||||||
/* The inner editable paragraph inherits `.ProseMirror p { margin: 0.5em 0 }`.
|
|
||||||
Drop the outer margins so the definition sits tight to the heading above and
|
|
||||||
the ::before number aligns with the top of the row — same approach used for
|
|
||||||
callouts in core.css. */
|
|
||||||
.definitionContent > :first-child {
|
.definitionContent > :first-child {
|
||||||
margin-top: 0;
|
margin-top: 0;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,140 +0,0 @@
|
|||||||
/**
|
|
||||||
* gitmost #401 — regression test for the connect-vs-unload race in
|
|
||||||
* @hocuspocus/server 3.4.4 (patched via patches/@hocuspocus__server@3.4.4.patch).
|
|
||||||
*
|
|
||||||
* The race (unpatched): when the last client disconnects, storeDocumentHooks'
|
|
||||||
* `finally` schedules an async `unloadDocument`. That unload runs its
|
|
||||||
* `beforeUnloadDocument` hooks asynchronously and, meanwhile, records an
|
|
||||||
* in-flight promise in `this.unloadingDocuments`. In the original 3.4.4
|
|
||||||
* `createDocument`, a NEW connection arriving in that window falls straight
|
|
||||||
* through to the `loadingDocuments`/`documents` checks — it never consults
|
|
||||||
* `unloadingDocuments`. So the new connection can start loading (or reuse) a
|
|
||||||
* document while the old instance is still being torn down; the re-check inside
|
|
||||||
* unload (`shouldUnloadDocument`, which sees 0 connections because async auth
|
|
||||||
* hooks have not registered the new connection yet) then deletes/destroys the
|
|
||||||
* doc out from under the freshly-connected client → orphaned Document → later
|
|
||||||
* redis-sync takes the "doc not loaded" path → sync never completes → the
|
|
||||||
* provider hangs until its ~25s timeout.
|
|
||||||
*
|
|
||||||
* The patch: `createDocument` first awaits any in-flight
|
|
||||||
* `unloadingDocuments.get(name)` before proceeding. Once that settles, the
|
|
||||||
* decision is deterministic — either the doc was fully unloaded (gone from
|
|
||||||
* `documents`, so a clean fresh load) or the unload aborted (healthy doc still
|
|
||||||
* in `documents`, reused). The new connection can never hand-shake onto an
|
|
||||||
* about-to-be-destroyed Document.
|
|
||||||
*
|
|
||||||
* These tests exercise the REAL patched `Hocuspocus.createDocument` (the class
|
|
||||||
* is directly constructible) by seeding `unloadingDocuments` with a controllable
|
|
||||||
* in-flight unload and observing that createDocument waits for it.
|
|
||||||
*/
|
|
||||||
import { Hocuspocus } from '@hocuspocus/server';
|
|
||||||
|
|
||||||
// A promise we can resolve on demand, to model an unload that is mid-flight.
|
|
||||||
function deferred<T = void>() {
|
|
||||||
let resolve!: (v: T) => void;
|
|
||||||
const promise = new Promise<T>((r) => {
|
|
||||||
resolve = r;
|
|
||||||
});
|
|
||||||
return { promise, resolve };
|
|
||||||
}
|
|
||||||
|
|
||||||
describe('gitmost #401 — hocuspocus createDocument awaits in-flight unload', () => {
|
|
||||||
it('does NOT start loading a new doc until the in-flight unload settles, then loads fresh', async () => {
|
|
||||||
const hp = new Hocuspocus();
|
|
||||||
const name = 'page.race';
|
|
||||||
|
|
||||||
// Observe loadDocument: on the unpatched code it is invoked synchronously
|
|
||||||
// within createDocument (before the unload settles); on the patched code it
|
|
||||||
// must be deferred until unloadingDocuments resolves.
|
|
||||||
const freshDoc = { name, __fresh: true } as any;
|
|
||||||
const loadSpy = jest
|
|
||||||
.spyOn(hp as any, 'loadDocument')
|
|
||||||
.mockResolvedValue(freshDoc);
|
|
||||||
|
|
||||||
// Model an unload in progress: an entry sits in unloadingDocuments and, when
|
|
||||||
// it completes, it removes the doc from `documents` (a real full unload).
|
|
||||||
const unload = deferred();
|
|
||||||
(hp as any).documents.set(name, { name, __dying: true });
|
|
||||||
(hp as any).unloadingDocuments.set(
|
|
||||||
name,
|
|
||||||
unload.promise.then(() => {
|
|
||||||
(hp as any).documents.delete(name);
|
|
||||||
}),
|
|
||||||
);
|
|
||||||
|
|
||||||
// Kick off a new connection's createDocument but do not await it yet.
|
|
||||||
const createPromise = (hp as any).createDocument(
|
|
||||||
name,
|
|
||||||
{},
|
|
||||||
'socket-1',
|
|
||||||
{ isAuthenticated: true, readOnly: false },
|
|
||||||
{},
|
|
||||||
);
|
|
||||||
|
|
||||||
// Let all currently-schedulable microtasks run. The patched createDocument is
|
|
||||||
// now parked on `await unloadingDocuments.get(name)`, so loadDocument must
|
|
||||||
// NOT have been called yet, and it must NOT have returned the dying doc.
|
|
||||||
await Promise.resolve();
|
|
||||||
await Promise.resolve();
|
|
||||||
expect(loadSpy).not.toHaveBeenCalled();
|
|
||||||
|
|
||||||
// The unload completes (doc removed from `documents`).
|
|
||||||
unload.resolve();
|
|
||||||
|
|
||||||
// createDocument now proceeds: sees no existing doc → fresh load.
|
|
||||||
const doc = await createPromise;
|
|
||||||
expect(loadSpy).toHaveBeenCalledTimes(1);
|
|
||||||
expect(doc).toBe(freshDoc);
|
|
||||||
// The freshly-loaded doc is the one registered — never the dying instance.
|
|
||||||
expect((hp as any).documents.get(name)).toBe(freshDoc);
|
|
||||||
});
|
|
||||||
|
|
||||||
it('reuses the live doc when the in-flight unload aborts (doc left in documents)', async () => {
|
|
||||||
const hp = new Hocuspocus();
|
|
||||||
const name = 'page.abort';
|
|
||||||
|
|
||||||
const loadSpy = jest.spyOn(hp as any, 'loadDocument');
|
|
||||||
|
|
||||||
// Model an unload that ABORTS (e.g. a new connection reappeared before the
|
|
||||||
// sync re-check): it settles WITHOUT deleting the doc from `documents`.
|
|
||||||
const unload = deferred();
|
|
||||||
const liveDoc = { name, __live: true } as any;
|
|
||||||
(hp as any).documents.set(name, liveDoc);
|
|
||||||
(hp as any).unloadingDocuments.set(name, unload.promise); // no-op unload
|
|
||||||
|
|
||||||
const createPromise = (hp as any).createDocument(
|
|
||||||
name,
|
|
||||||
{},
|
|
||||||
'socket-2',
|
|
||||||
{ isAuthenticated: true, readOnly: false },
|
|
||||||
{},
|
|
||||||
);
|
|
||||||
|
|
||||||
unload.resolve();
|
|
||||||
const doc = await createPromise;
|
|
||||||
|
|
||||||
// The still-present live doc is reused; no fresh load happened.
|
|
||||||
expect(doc).toBe(liveDoc);
|
|
||||||
expect(loadSpy).not.toHaveBeenCalled();
|
|
||||||
});
|
|
||||||
|
|
||||||
it('no in-flight unload → behaves normally (fresh load)', async () => {
|
|
||||||
const hp = new Hocuspocus();
|
|
||||||
const name = 'page.normal';
|
|
||||||
const freshDoc = { name } as any;
|
|
||||||
const loadSpy = jest
|
|
||||||
.spyOn(hp as any, 'loadDocument')
|
|
||||||
.mockResolvedValue(freshDoc);
|
|
||||||
|
|
||||||
const doc = await (hp as any).createDocument(
|
|
||||||
name,
|
|
||||||
{},
|
|
||||||
'socket-3',
|
|
||||||
{ isAuthenticated: true, readOnly: false },
|
|
||||||
{},
|
|
||||||
);
|
|
||||||
|
|
||||||
expect(loadSpy).toHaveBeenCalledTimes(1);
|
|
||||||
expect(doc).toBe(freshDoc);
|
|
||||||
});
|
|
||||||
});
|
|
||||||
@@ -1,130 +0,0 @@
|
|||||||
/**
|
|
||||||
* gitmost #401 fix 2 — onLoadDocument applies the DB state directly into the
|
|
||||||
* hook's target document and returns undefined (instead of building a NEW Y.Doc
|
|
||||||
* and returning it, which made hocuspocus re-encode+apply the whole state a
|
|
||||||
* SECOND time on every cold load).
|
|
||||||
*
|
|
||||||
* These tests assert:
|
|
||||||
* - the hook mutates `data.document` in place so its content equals the DB doc,
|
|
||||||
* - onLoadDocument returns undefined (so hocuspocus keeps the mutated doc and
|
|
||||||
* does NOT run its own applyUpdate(encodeStateAsUpdate(...)) merge),
|
|
||||||
* - both the raw-ydoc branch and the json→ydoc conversion branch behave so.
|
|
||||||
*
|
|
||||||
* Returning undefined is the observable signal that the double-encode is gone
|
|
||||||
* (the old code returned a new Y.Doc, which made hocuspocus re-encode+apply the
|
|
||||||
* state a second time); we assert that contract rather than counting internal
|
|
||||||
* encode calls, which is brittle given the encodes inside toYdoc and the test's
|
|
||||||
* own `expected` fixtures.
|
|
||||||
*/
|
|
||||||
import * as Y from 'yjs';
|
|
||||||
import { Document } from '@hocuspocus/server';
|
|
||||||
import { TiptapTransformer } from '@hocuspocus/transformer';
|
|
||||||
import { PersistenceExtension } from './persistence.extension';
|
|
||||||
import { tiptapExtensions } from '../collaboration.util';
|
|
||||||
|
|
||||||
// A fresh hocuspocus Document (extends Y.Doc, adds isEmpty()) as hocuspocus
|
|
||||||
// hands to onLoadDocument on a cold load.
|
|
||||||
const freshDoc = () => new Document(`page.${PAGE_ID}`, {});
|
|
||||||
|
|
||||||
const PAGE_ID = '550e8400-e29b-41d4-a716-446655440000';
|
|
||||||
|
|
||||||
const doc = (text: string) => ({
|
|
||||||
type: 'doc',
|
|
||||||
content: [{ type: 'paragraph', content: [{ type: 'text', text }] }],
|
|
||||||
});
|
|
||||||
|
|
||||||
const jsonOf = (ydoc: Y.Doc) =>
|
|
||||||
TiptapTransformer.fromYdoc(ydoc, 'default');
|
|
||||||
|
|
||||||
describe('PersistenceExtension.onLoadDocument — #401 fix 2 (apply-into-hook-doc)', () => {
|
|
||||||
let ext: PersistenceExtension;
|
|
||||||
let pageRepo: { findById: jest.Mock };
|
|
||||||
|
|
||||||
beforeEach(() => {
|
|
||||||
pageRepo = { findById: jest.fn() };
|
|
||||||
ext = new PersistenceExtension(
|
|
||||||
pageRepo as any,
|
|
||||||
{} as any,
|
|
||||||
{} as any,
|
|
||||||
{} as any,
|
|
||||||
{} as any,
|
|
||||||
{} as any,
|
|
||||||
{} as any,
|
|
||||||
{} as any,
|
|
||||||
);
|
|
||||||
jest.spyOn(ext['logger'], 'debug').mockImplementation(() => undefined);
|
|
||||||
jest.spyOn(ext['logger'], 'warn').mockImplementation(() => undefined);
|
|
||||||
});
|
|
||||||
|
|
||||||
const load = (document: Document) =>
|
|
||||||
ext.onLoadDocument({ documentName: `page.${PAGE_ID}`, document } as any);
|
|
||||||
|
|
||||||
it('raw ydoc branch: mutates the hook doc to the DB state and returns undefined', async () => {
|
|
||||||
// Source doc representing the persisted ydoc state.
|
|
||||||
const source = TiptapTransformer.toYdoc(
|
|
||||||
doc('DB CONTENT'),
|
|
||||||
'default',
|
|
||||||
tiptapExtensions,
|
|
||||||
);
|
|
||||||
const dbState = Buffer.from(Y.encodeStateAsUpdate(source));
|
|
||||||
pageRepo.findById.mockResolvedValue({ id: PAGE_ID, ydoc: dbState });
|
|
||||||
|
|
||||||
// The hook target is a fresh empty doc (as hocuspocus supplies on cold load).
|
|
||||||
const target = freshDoc();
|
|
||||||
const result = await load(target);
|
|
||||||
|
|
||||||
// Return undefined so hocuspocus keeps `target` as-is (no second merge).
|
|
||||||
expect(result).toBeUndefined();
|
|
||||||
// The hook document now carries the DB content.
|
|
||||||
expect(jsonOf(target)).toEqual(jsonOf(source));
|
|
||||||
});
|
|
||||||
|
|
||||||
it('json→ydoc branch: converts page.content into the hook doc and returns undefined', async () => {
|
|
||||||
pageRepo.findById.mockResolvedValue({
|
|
||||||
id: PAGE_ID,
|
|
||||||
ydoc: null,
|
|
||||||
content: doc('JSON CONTENT'),
|
|
||||||
});
|
|
||||||
|
|
||||||
const target = freshDoc();
|
|
||||||
const result = await load(target);
|
|
||||||
|
|
||||||
// Returning undefined is what keeps hocuspocus from re-encoding+applying the
|
|
||||||
// state a second time (the old code returned the doc, forcing that extra
|
|
||||||
// encode). We assert the observable contract here — the return value and the
|
|
||||||
// resulting content — rather than counting internal encode calls, which is
|
|
||||||
// brittle: toYdoc and the `expected` build below both encode too.
|
|
||||||
expect(result).toBeUndefined();
|
|
||||||
|
|
||||||
// The converted content landed in the hook document.
|
|
||||||
const expected = TiptapTransformer.toYdoc(
|
|
||||||
doc('JSON CONTENT'),
|
|
||||||
'default',
|
|
||||||
tiptapExtensions,
|
|
||||||
);
|
|
||||||
expect(jsonOf(target)).toEqual(jsonOf(expected));
|
|
||||||
});
|
|
||||||
|
|
||||||
it('live doc already non-empty: early return, no DB read', async () => {
|
|
||||||
// A hocuspocus Document carrying live content (isEmpty('default') === false).
|
|
||||||
const target = freshDoc();
|
|
||||||
const live = TiptapTransformer.toYdoc(
|
|
||||||
doc('LIVE'),
|
|
||||||
'default',
|
|
||||||
tiptapExtensions,
|
|
||||||
);
|
|
||||||
Y.applyUpdate(target, Y.encodeStateAsUpdate(live));
|
|
||||||
|
|
||||||
const result = await load(target);
|
|
||||||
expect(result).toBeUndefined();
|
|
||||||
expect(pageRepo.findById).not.toHaveBeenCalled();
|
|
||||||
});
|
|
||||||
|
|
||||||
it('no persisted state: leaves the fresh empty doc untouched, returns undefined', async () => {
|
|
||||||
pageRepo.findById.mockResolvedValue({ id: PAGE_ID, ydoc: null, content: null });
|
|
||||||
const target = freshDoc();
|
|
||||||
const result = await load(target);
|
|
||||||
expect(result).toBeUndefined();
|
|
||||||
expect(target.isEmpty('default')).toBe(true);
|
|
||||||
});
|
|
||||||
});
|
|
||||||
@@ -171,21 +171,15 @@ export class PersistenceExtension implements Extension {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
// #401 fix 2 — apply the DB state DIRECTLY into the hook's target document
|
|
||||||
// (`document` === `data.document`) and return undefined. When onLoadDocument
|
|
||||||
// returns undefined, hocuspocus keeps the mutated hook document as-is; only
|
|
||||||
// when the hook RETURNS a Y.Doc does hocuspocus re-`applyUpdate(document,
|
|
||||||
// encodeStateAsUpdate(returned))` — a second full encode+apply of the whole
|
|
||||||
// (e.g. 315KB) state on every cold load. Mutating in place performs a single
|
|
||||||
// apply and avoids the throwaway `new Y.Doc()` allocation.
|
|
||||||
if (page.ydoc) {
|
if (page.ydoc) {
|
||||||
this.logger.debug(`ydoc loaded from db: ${pageId}`);
|
this.logger.debug(`ydoc loaded from db: ${pageId}`);
|
||||||
|
|
||||||
|
const doc = new Y.Doc();
|
||||||
const dbState = new Uint8Array(page.ydoc);
|
const dbState = new Uint8Array(page.ydoc);
|
||||||
|
|
||||||
Y.applyUpdate(document, dbState);
|
Y.applyUpdate(doc, dbState);
|
||||||
observeCollabLoad(dbState.length, (performance.now() - startedAt) / 1000);
|
observeCollabLoad(dbState.length, (performance.now() - startedAt) / 1000);
|
||||||
return;
|
return doc;
|
||||||
}
|
}
|
||||||
|
|
||||||
// if no ydoc state in db convert json in page.content to Ydoc.
|
// if no ydoc state in db convert json in page.content to Ydoc.
|
||||||
@@ -198,23 +192,18 @@ export class PersistenceExtension implements Extension {
|
|||||||
tiptapExtensions,
|
tiptapExtensions,
|
||||||
);
|
);
|
||||||
|
|
||||||
// Encode the converted doc ONCE, reuse the bytes for both the size label
|
// Reuse this single encode for the size label (do NOT add a second one).
|
||||||
// and the single apply into the hook document (previously this encode's
|
|
||||||
// result was returned and hocuspocus re-encoded+applied it a second time).
|
|
||||||
const encoded = Y.encodeStateAsUpdate(ydoc);
|
const encoded = Y.encodeStateAsUpdate(ydoc);
|
||||||
Y.applyUpdate(document, encoded);
|
|
||||||
observeCollabLoad(
|
observeCollabLoad(
|
||||||
encoded.byteLength,
|
encoded.byteLength,
|
||||||
(performance.now() - startedAt) / 1000,
|
(performance.now() - startedAt) / 1000,
|
||||||
);
|
);
|
||||||
return;
|
return ydoc;
|
||||||
}
|
}
|
||||||
|
|
||||||
// No persisted state: the hook document is already a fresh empty Y.Doc, so
|
|
||||||
// leave it untouched and return undefined (no re-encode of an empty doc).
|
|
||||||
this.logger.debug(`creating fresh ydoc: ${pageId}`);
|
this.logger.debug(`creating fresh ydoc: ${pageId}`);
|
||||||
observeCollabLoad(0, (performance.now() - startedAt) / 1000);
|
observeCollabLoad(0, (performance.now() - startedAt) / 1000);
|
||||||
return;
|
return new Y.Doc();
|
||||||
}
|
}
|
||||||
|
|
||||||
async onStoreDocument(data: onStoreDocumentPayload) {
|
async onStoreDocument(data: onStoreDocumentPayload) {
|
||||||
|
|||||||
@@ -17,24 +17,10 @@ import { Injectable, Logger, OnModuleDestroy } from '@nestjs/common';
|
|||||||
/** How long a finished entry is retained for late attach (replay + immediate end). */
|
/** How long a finished entry is retained for late attach (replay + immediate end). */
|
||||||
export const RUN_STREAM_RETAIN_FINISHED_MS = 30_000;
|
export const RUN_STREAM_RETAIN_FINISHED_MS = 30_000;
|
||||||
|
|
||||||
/**
|
/** Per-run replay buffer cap. Past this the buffer is dropped (attach -> 204). */
|
||||||
* Per-run replay buffer cap. Past this the buffer is dropped (attach -> 204, and
|
export const RUN_STREAM_MAX_BUFFER_BYTES = 4 * 1024 * 1024;
|
||||||
* the client falls back to its restore + degraded-poll path, #430).
|
|
||||||
*
|
|
||||||
* Raised from 4MB to 32MB (#430): marathon autonomous runs (11-25 min observed)
|
|
||||||
* stream far more than 4MB of SSE frames, so a live disconnect mid-run would find
|
|
||||||
* an already-overflowed buffer and could only degrade-poll instead of re-attaching
|
|
||||||
* to the live tail. 32MB comfortably covers those runs while staying bounded.
|
|
||||||
*
|
|
||||||
* Memory cost: this is the WORST-CASE retained size PER ACTIVE run (the buffer is
|
|
||||||
* freed on finish + retention, or dropped immediately on overflow). With the small
|
|
||||||
* number of concurrent autonomous runs a single workspace realistically has, 32MB
|
|
||||||
* each is an acceptable ceiling; the overflow->204->degraded-poll fallback remains
|
|
||||||
* the backstop for anything larger, so correctness never depends on this bound.
|
|
||||||
*/
|
|
||||||
export const RUN_STREAM_MAX_BUFFER_BYTES = 32 * 1024 * 1024;
|
|
||||||
|
|
||||||
// 2x the replay cap: a just-written full-replay burst alone can never trip the
|
// 2x the replay cap: a just-written 4MB replay burst alone can never trip the
|
||||||
// per-subscriber cap (see controller); only a genuinely stalled socket can.
|
// per-subscriber cap (see controller); only a genuinely stalled socket can.
|
||||||
export const SUBSCRIBER_MAX_BUFFERED_BYTES = 2 * RUN_STREAM_MAX_BUFFER_BYTES;
|
export const SUBSCRIBER_MAX_BUFFERED_BYTES = 2 * RUN_STREAM_MAX_BUFFER_BYTES;
|
||||||
|
|
||||||
|
|||||||
@@ -2,7 +2,6 @@ import {
|
|||||||
AiChatStreamRegistryService,
|
AiChatStreamRegistryService,
|
||||||
RUN_STREAM_MAX_BUFFER_BYTES,
|
RUN_STREAM_MAX_BUFFER_BYTES,
|
||||||
RUN_STREAM_RETAIN_FINISHED_MS,
|
RUN_STREAM_RETAIN_FINISHED_MS,
|
||||||
SUBSCRIBER_MAX_BUFFERED_BYTES,
|
|
||||||
RunStreamCallbacks,
|
RunStreamCallbacks,
|
||||||
} from './ai-chat-stream-registry.service';
|
} from './ai-chat-stream-registry.service';
|
||||||
|
|
||||||
@@ -211,10 +210,9 @@ describe('AiChatStreamRegistryService', () => {
|
|||||||
const att = (await registry.attach(CHAT, false, undefined, c.cb))!;
|
const att = (await registry.attach(CHAT, false, undefined, c.cb))!;
|
||||||
att.start();
|
att.start();
|
||||||
|
|
||||||
// Cap-relative so it survives a buffer-cap change (#430): a quarter-cap frame
|
const oneMb = 'x'.repeat(1024 * 1024);
|
||||||
// means 5 frames comfortably exceed the replay cap; the last one crosses.
|
// 5 x 1MB = 5MB > 4MB cap; the 5th frame is the one that crosses.
|
||||||
const chunk = 'x'.repeat(Math.floor(RUN_STREAM_MAX_BUFFER_BYTES / 4));
|
for (let i = 0; i < 5; i++) src.push(oneMb + i);
|
||||||
for (let i = 0; i < 5; i++) src.push(chunk + i);
|
|
||||||
await flush();
|
await flush();
|
||||||
|
|
||||||
const entry = (registry as any).entries.get(CHAT);
|
const entry = (registry as any).entries.get(CHAT);
|
||||||
@@ -222,7 +220,7 @@ describe('AiChatStreamRegistryService', () => {
|
|||||||
expect(entry.bytes).toBeGreaterThan(RUN_STREAM_MAX_BUFFER_BYTES);
|
expect(entry.bytes).toBeGreaterThan(RUN_STREAM_MAX_BUFFER_BYTES);
|
||||||
// The live subscriber received ALL 5 frames, including the crossing one.
|
// The live subscriber received ALL 5 frames, including the crossing one.
|
||||||
expect(c.frames).toHaveLength(5);
|
expect(c.frames).toHaveLength(5);
|
||||||
expect(c.frames[4]).toBe(chunk + 4);
|
expect(c.frames[4]).toBe(oneMb + 4);
|
||||||
|
|
||||||
// A NEW attach after overflow gets null (replay buffer is gone).
|
// A NEW attach after overflow gets null (replay buffer is gone).
|
||||||
const c2 = collector();
|
const c2 = collector();
|
||||||
@@ -242,11 +240,9 @@ describe('AiChatStreamRegistryService', () => {
|
|||||||
const attB = (await registry.attach(CHAT, false, undefined, b.cb))!;
|
const attB = (await registry.attach(CHAT, false, undefined, b.cb))!;
|
||||||
attB.start();
|
attB.start();
|
||||||
|
|
||||||
// Cap-relative so it survives a buffer-cap change (#430): a quarter-of-the-
|
const oneMb = 'x'.repeat(1024 * 1024);
|
||||||
// per-subscriber-cap frame means 5 frames exceed A's paused-pending cap while
|
// 9 x 1MB = 9MB > 8MB per-subscriber cap; A's pending overflows, B streams live.
|
||||||
// B streams every frame live.
|
for (let i = 0; i < 9; i++) src.push(oneMb + i);
|
||||||
const chunk = 'x'.repeat(Math.floor(SUBSCRIBER_MAX_BUFFERED_BYTES / 4));
|
|
||||||
for (let i = 0; i < 5; i++) src.push(chunk + i);
|
|
||||||
await flush();
|
await flush();
|
||||||
|
|
||||||
const entry = (registry as any).entries.get(CHAT);
|
const entry = (registry as any).entries.get(CHAT);
|
||||||
@@ -254,7 +250,7 @@ describe('AiChatStreamRegistryService', () => {
|
|||||||
expect(entry.subscribers.size).toBe(1);
|
expect(entry.subscribers.size).toBe(1);
|
||||||
expect(a.frames).toEqual([]); // paused + overflowed: nothing was delivered
|
expect(a.frames).toEqual([]); // paused + overflowed: nothing was delivered
|
||||||
// B received every frame live (delivery unaffected by A's overflow).
|
// B received every frame live (delivery unaffected by A's overflow).
|
||||||
expect(b.frames).toHaveLength(5);
|
expect(b.frames).toHaveLength(9);
|
||||||
|
|
||||||
// A's start() (arriving late) degrades to an immediate end, not a partial replay.
|
// A's start() (arriving late) degrades to an immediate end, not a partial replay.
|
||||||
attA.start();
|
attA.start();
|
||||||
|
|||||||
@@ -148,53 +148,6 @@ describe('assistantParts', () => {
|
|||||||
expect(toolPart).not.toHaveProperty('output');
|
expect(toolPart).not.toHaveProperty('output');
|
||||||
});
|
});
|
||||||
|
|
||||||
it('replays the REAL error text for a THROWN tool (tool-error part)', () => {
|
|
||||||
const steps = [
|
|
||||||
{
|
|
||||||
text: '',
|
|
||||||
toolCalls: [
|
|
||||||
{ toolCallId: 'c7', toolName: 'editPageText', input: { id: 'p1' } },
|
|
||||||
],
|
|
||||||
// A thrown tool is a `tool-error` content part; toolResults holds only
|
|
||||||
// successes and stays empty for this call.
|
|
||||||
toolResults: [],
|
|
||||||
content: [
|
|
||||||
{
|
|
||||||
type: 'tool-error',
|
|
||||||
toolCallId: 'c7',
|
|
||||||
toolName: 'editPageText',
|
|
||||||
input: { id: 'p1' },
|
|
||||||
error: new Error('page is locked'),
|
|
||||||
},
|
|
||||||
],
|
|
||||||
},
|
|
||||||
];
|
|
||||||
const parts = assistantParts(steps, '') as AnyPart[];
|
|
||||||
const toolPart = parts.find((p) => p.type === 'tool-editPageText');
|
|
||||||
expect(toolPart).toBeDefined();
|
|
||||||
expect(toolPart!.state).toBe('output-error');
|
|
||||||
// The REAL error is replayed, NOT the 'Tool call did not complete.' placeholder.
|
|
||||||
expect(toolPart!.errorText).toBe('page is locked');
|
|
||||||
expect(toolPart).not.toHaveProperty('output');
|
|
||||||
});
|
|
||||||
|
|
||||||
it('keeps the placeholder ONLY for a call with neither result nor tool-error', () => {
|
|
||||||
const steps = [
|
|
||||||
{
|
|
||||||
text: '',
|
|
||||||
toolCalls: [
|
|
||||||
{ toolCallId: 'c8', toolName: 'insertNode', input: { node: {} } },
|
|
||||||
],
|
|
||||||
toolResults: [],
|
|
||||||
content: [], // aborted mid-step: no result AND no tool-error
|
|
||||||
},
|
|
||||||
];
|
|
||||||
const parts = assistantParts(steps, '') as AnyPart[];
|
|
||||||
const toolPart = parts.find((p) => p.type === 'tool-insertNode');
|
|
||||||
expect(toolPart!.state).toBe('output-error');
|
|
||||||
expect(toolPart!.errorText).toBe('Tool call did not complete.');
|
|
||||||
});
|
|
||||||
|
|
||||||
it('skips malformed tool-calls (missing toolName or toolCallId)', () => {
|
it('skips malformed tool-calls (missing toolName or toolCallId)', () => {
|
||||||
const steps = [
|
const steps = [
|
||||||
{
|
{
|
||||||
@@ -242,45 +195,6 @@ describe('serializeSteps', () => {
|
|||||||
expect(trace[0]).toEqual({ toolName: 'getPage', input: { id: 'p1' } });
|
expect(trace[0]).toEqual({ toolName: 'getPage', input: { id: 'p1' } });
|
||||||
expect(trace[1]).toEqual({ toolName: 'getPage', output: { title: 'T' } });
|
expect(trace[1]).toEqual({ toolName: 'getPage', output: { title: 'T' } });
|
||||||
});
|
});
|
||||||
|
|
||||||
it('records a THROWN tool failure (tool-error part) with its error message', () => {
|
|
||||||
const trace = serializeSteps([
|
|
||||||
{
|
|
||||||
toolCalls: [{ toolName: 'editPageText', input: { id: 'p1' } }],
|
|
||||||
toolResults: [],
|
|
||||||
content: [
|
|
||||||
{
|
|
||||||
type: 'tool-error',
|
|
||||||
toolName: 'editPageText',
|
|
||||||
error: new Error('page is locked'),
|
|
||||||
},
|
|
||||||
],
|
|
||||||
},
|
|
||||||
]) as Array<Record<string, unknown>>;
|
|
||||||
// The call element is followed by a paired error element (mirroring how a
|
|
||||||
// successful result is appended), so the failure survives in the trace.
|
|
||||||
expect(trace).toHaveLength(2);
|
|
||||||
expect(trace[0]).toEqual({ toolName: 'editPageText', input: { id: 'p1' } });
|
|
||||||
expect(trace[1]).toEqual({
|
|
||||||
toolName: 'editPageText',
|
|
||||||
error: 'page is locked',
|
|
||||||
});
|
|
||||||
});
|
|
||||||
|
|
||||||
it('truncates a very long tool-error message to the tool-output limit', () => {
|
|
||||||
const long = 'x'.repeat(5000);
|
|
||||||
const trace = serializeSteps([
|
|
||||||
{
|
|
||||||
toolCalls: [{ toolName: 'editPageText', input: {} }],
|
|
||||||
toolResults: [],
|
|
||||||
content: [{ type: 'tool-error', toolName: 'editPageText', error: long }],
|
|
||||||
},
|
|
||||||
]) as Array<Record<string, unknown>>;
|
|
||||||
const errorText = trace[1].error as string;
|
|
||||||
// Truncated (not the full 5000 chars) and carries the omission marker.
|
|
||||||
expect(errorText.length).toBeLessThan(long.length);
|
|
||||||
expect(errorText).toContain('chars omitted');
|
|
||||||
});
|
|
||||||
});
|
});
|
||||||
|
|
||||||
describe('rowToUiMessage', () => {
|
describe('rowToUiMessage', () => {
|
||||||
|
|||||||
@@ -1637,17 +1637,6 @@ type StepLike = {
|
|||||||
toolName?: string;
|
toolName?: string;
|
||||||
output?: unknown;
|
output?: unknown;
|
||||||
}>;
|
}>;
|
||||||
// ai@6.0.134: a tool that THREW surfaces as a `tool-error` content part
|
|
||||||
// ({ type:'tool-error', toolCallId, toolName, input, error }), NOT as a
|
|
||||||
// `toolResults` entry (which holds only successes). Read from here so failed
|
|
||||||
// calls are persisted with their real error instead of being dropped.
|
|
||||||
content?: ReadonlyArray<{
|
|
||||||
type?: string;
|
|
||||||
toolCallId?: string;
|
|
||||||
toolName?: string;
|
|
||||||
input?: unknown;
|
|
||||||
error?: unknown;
|
|
||||||
}>;
|
|
||||||
};
|
};
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -1750,26 +1739,6 @@ function compactValue(value: unknown, depth: number): unknown {
|
|||||||
return value;
|
return value;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* Extract a bounded string message from a `tool-error` part's `error` field for
|
|
||||||
* persistence and history replay. The field may be an `Error`, a string, or an
|
|
||||||
* arbitrary object, so pull a message robustly. The result is passed through
|
|
||||||
* `compactValue` so a very long error honors the SAME truncation limits the file
|
|
||||||
* already applies to tool outputs (no new limit is introduced here).
|
|
||||||
*/
|
|
||||||
function normalizeToolError(error: unknown): string {
|
|
||||||
const message =
|
|
||||||
error instanceof Error
|
|
||||||
? error.message
|
|
||||||
: typeof error === 'string'
|
|
||||||
? error
|
|
||||||
: error != null &&
|
|
||||||
typeof (error as { message?: unknown }).message === 'string'
|
|
||||||
? (error as { message: string }).message
|
|
||||||
: String(error);
|
|
||||||
return compactValue(message, 0) as string;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Rebuild the FULL UIMessage `parts` for an assistant turn from the SDK steps,
|
* Rebuild the FULL UIMessage `parts` for an assistant turn from the SDK steps,
|
||||||
* so multi-turn history replays prior tool-calls/results to the model (not just
|
* so multi-turn history replays prior tool-calls/results to the model (not just
|
||||||
@@ -1802,14 +1771,6 @@ export function assistantParts(
|
|||||||
for (const r of step.toolResults ?? []) {
|
for (const r of step.toolResults ?? []) {
|
||||||
if (r.toolCallId) resultsById.set(r.toolCallId, r.output);
|
if (r.toolCallId) resultsById.set(r.toolCallId, r.output);
|
||||||
}
|
}
|
||||||
// Index this step's THROWN tool failures (ai@6 `tool-error` content parts)
|
|
||||||
// by tool call id, so a call that failed replays with its real error text.
|
|
||||||
const errorsById = new Map<string, unknown>();
|
|
||||||
for (const part of step.content ?? []) {
|
|
||||||
if (part.type === 'tool-error' && part.toolCallId) {
|
|
||||||
errorsById.set(part.toolCallId, part.error);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
for (const call of step.toolCalls ?? []) {
|
for (const call of step.toolCalls ?? []) {
|
||||||
if (!call.toolName || !call.toolCallId) continue;
|
if (!call.toolName || !call.toolCallId) continue;
|
||||||
const hasResult = resultsById.has(call.toolCallId);
|
const hasResult = resultsById.has(call.toolCallId);
|
||||||
@@ -1822,21 +1783,9 @@ export function assistantParts(
|
|||||||
input: call.input,
|
input: call.input,
|
||||||
output: compactToolOutput(resultsById.get(call.toolCallId)),
|
output: compactToolOutput(resultsById.get(call.toolCallId)),
|
||||||
});
|
});
|
||||||
} else if (errorsById.has(call.toolCallId)) {
|
|
||||||
// The tool THREW: replay the REAL error so the model on the next turn
|
|
||||||
// knows WHY the call failed (and does not blindly repeat it). An
|
|
||||||
// output-error round-trips through convertToModelMessages as a balanced
|
|
||||||
// tool-call + tool-result, keeping the rebuilt history valid.
|
|
||||||
parts.push({
|
|
||||||
type: `tool-${call.toolName}`,
|
|
||||||
toolCallId: call.toolCallId,
|
|
||||||
state: 'output-error',
|
|
||||||
input: call.input,
|
|
||||||
errorText: normalizeToolError(errorsById.get(call.toolCallId)),
|
|
||||||
});
|
|
||||||
} else {
|
} else {
|
||||||
// No paired result AND no tool-error (e.g. aborted mid-step). Persisting
|
// No paired result (e.g. aborted mid-step). Persisting a bare
|
||||||
// a bare tool-call (input-available) would replay as an unpaired call and
|
// tool-call (input-available) would replay as an unpaired call and
|
||||||
// throw MissingToolResultsError on the next turn (convertToModelMessages
|
// throw MissingToolResultsError on the next turn (convertToModelMessages
|
||||||
// emits no tool-result for it). Emit a SYNTHETIC paired result instead:
|
// emits no tool-result for it). Emit a SYNTHETIC paired result instead:
|
||||||
// an output-error round-trips through convertToModelMessages as a
|
// an output-error round-trips through convertToModelMessages as a
|
||||||
@@ -2072,19 +2021,10 @@ export function serializeSteps(
|
|||||||
steps: ReadonlyArray<{
|
steps: ReadonlyArray<{
|
||||||
toolCalls?: ReadonlyArray<{ toolName?: string; input?: unknown }>;
|
toolCalls?: ReadonlyArray<{ toolName?: string; input?: unknown }>;
|
||||||
toolResults?: ReadonlyArray<{ toolName?: string; output?: unknown }>;
|
toolResults?: ReadonlyArray<{ toolName?: string; output?: unknown }>;
|
||||||
content?: ReadonlyArray<{
|
|
||||||
type?: string;
|
|
||||||
toolName?: string;
|
|
||||||
error?: unknown;
|
|
||||||
}>;
|
|
||||||
}>,
|
}>,
|
||||||
): unknown {
|
): unknown {
|
||||||
const calls: Array<{
|
const calls: Array<{ toolName?: string; input?: unknown; output?: unknown }> =
|
||||||
toolName?: string;
|
[];
|
||||||
input?: unknown;
|
|
||||||
output?: unknown;
|
|
||||||
error?: string;
|
|
||||||
}> = [];
|
|
||||||
for (const step of steps ?? []) {
|
for (const step of steps ?? []) {
|
||||||
for (const call of step.toolCalls ?? []) {
|
for (const call of step.toolCalls ?? []) {
|
||||||
calls.push({ toolName: call.toolName, input: call.input });
|
calls.push({ toolName: call.toolName, input: call.input });
|
||||||
@@ -2092,18 +2032,6 @@ export function serializeSteps(
|
|||||||
for (const r of step.toolResults ?? []) {
|
for (const r of step.toolResults ?? []) {
|
||||||
calls.push({ toolName: r.toolName, output: compactToolOutput(r.output) });
|
calls.push({ toolName: r.toolName, output: compactToolOutput(r.output) });
|
||||||
}
|
}
|
||||||
// ai@6 surfaces a THROWN tool failure as a `tool-error` content part, NOT as
|
|
||||||
// a `toolResults` entry. Record it as its own paired element (mirroring how a
|
|
||||||
// successful result is appended) so the failure and its reason survive in the
|
|
||||||
// trace instead of leaving an orphaned call with no result.
|
|
||||||
for (const part of step.content ?? []) {
|
|
||||||
if (part.type === 'tool-error') {
|
|
||||||
calls.push({
|
|
||||||
toolName: part.toolName,
|
|
||||||
error: normalizeToolError(part.error),
|
|
||||||
});
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
return calls.length > 0 ? calls : null;
|
return calls.length > 0 ? calls : null;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -17,7 +17,7 @@ import {
|
|||||||
resolveCurrentPageResult,
|
resolveCurrentPageResult,
|
||||||
type SelectionContext,
|
type SelectionContext,
|
||||||
} from './current-page.util';
|
} from './current-page.util';
|
||||||
import { parseNodeArg } from './parse-node-arg';
|
import { parseNodeArg } from '@docmost/prosemirror-markdown';
|
||||||
import { modelFriendlyInput } from './model-friendly-input';
|
import { modelFriendlyInput } from './model-friendly-input';
|
||||||
import { SandboxStore } from '../../../integrations/sandbox/sandbox.store';
|
import { SandboxStore } from '../../../integrations/sandbox/sandbox.store';
|
||||||
import {
|
import {
|
||||||
@@ -729,35 +729,6 @@ export class AiChatToolsService {
|
|||||||
}),
|
}),
|
||||||
),
|
),
|
||||||
|
|
||||||
// Schema + description live in @docmost/mcp's SHARED_TOOL_SPECS (#423).
|
|
||||||
// meta.hash in the result is the baseHash drawioUpdate requires.
|
|
||||||
drawioGet: sharedTool(
|
|
||||||
sharedToolSpecs.drawioGet,
|
|
||||||
async ({ pageId, node, format }) =>
|
|
||||||
await client.drawioGet(pageId, node, format ?? 'xml'),
|
|
||||||
),
|
|
||||||
|
|
||||||
// Schema + description live in @docmost/mcp's SHARED_TOOL_SPECS (#423).
|
|
||||||
// The flat schema fields are regrouped into the client's `where` object.
|
|
||||||
drawioCreate: sharedTool(
|
|
||||||
sharedToolSpecs.drawioCreate,
|
|
||||||
async ({ pageId, xml, position, anchorNodeId, anchorText, title }) =>
|
|
||||||
await client.drawioCreate(
|
|
||||||
pageId,
|
|
||||||
{ position, anchorNodeId, anchorText },
|
|
||||||
xml,
|
|
||||||
title,
|
|
||||||
),
|
|
||||||
),
|
|
||||||
|
|
||||||
// Schema + description live in @docmost/mcp's SHARED_TOOL_SPECS (#423).
|
|
||||||
// baseHash is the optimistic lock: mismatch => structured conflict error.
|
|
||||||
drawioUpdate: sharedTool(
|
|
||||||
sharedToolSpecs.drawioUpdate,
|
|
||||||
async ({ pageId, node, xml, baseHash }) =>
|
|
||||||
await client.drawioUpdate(pageId, node, xml, baseHash),
|
|
||||||
),
|
|
||||||
|
|
||||||
// Schema + description now live in @docmost/mcp's SHARED_TOOL_SPECS (#294).
|
// Schema + description now live in @docmost/mcp's SHARED_TOOL_SPECS (#294).
|
||||||
// The table reference parameter was unified to `table` (was `tableRef`).
|
// The table reference parameter was unified to `table` (was `tableRef`).
|
||||||
tableInsertRow: sharedTool(
|
tableInsertRow: sharedTool(
|
||||||
|
|||||||
@@ -168,32 +168,6 @@ export interface DocmostClientLike {
|
|||||||
url: string,
|
url: string,
|
||||||
opts?: { align?: 'left' | 'center' | 'right'; alt?: string },
|
opts?: { align?: 'left' | 'center' | 'right'; alt?: string },
|
||||||
): Promise<Record<string, unknown>>;
|
): Promise<Record<string, unknown>>;
|
||||||
// --- draw.io diagrams (#423, stage 1) ---
|
|
||||||
// Read a diagram as decoded mxGraph XML (default) or the raw .drawio.svg.
|
|
||||||
// meta.hash is the optimistic-lock key drawioUpdate expects as baseHash.
|
|
||||||
drawioGet(
|
|
||||||
pageId: string,
|
|
||||||
node: string,
|
|
||||||
format?: 'xml' | 'svg',
|
|
||||||
): Promise<Record<string, unknown>>;
|
|
||||||
// Lint mxGraph XML, build the .drawio.svg attachment and insert a drawio node.
|
|
||||||
drawioCreate(
|
|
||||||
pageId: string,
|
|
||||||
where: {
|
|
||||||
position: 'before' | 'after' | 'append';
|
|
||||||
anchorNodeId?: string;
|
|
||||||
anchorText?: string;
|
|
||||||
},
|
|
||||||
xml: string,
|
|
||||||
title?: string,
|
|
||||||
): Promise<Record<string, unknown>>;
|
|
||||||
// Optimistic-locked full replacement of a diagram (baseHash from drawioGet).
|
|
||||||
drawioUpdate(
|
|
||||||
pageId: string,
|
|
||||||
node: string,
|
|
||||||
xml: string,
|
|
||||||
baseHash: string,
|
|
||||||
): Promise<Record<string, unknown>>;
|
|
||||||
tableInsertRow(
|
tableInsertRow(
|
||||||
pageId: string,
|
pageId: string,
|
||||||
tableRef: string,
|
tableRef: string,
|
||||||
|
|||||||
@@ -1,10 +1,10 @@
|
|||||||
import { parseNodeArg } from './parse-node-arg';
|
import { parseNodeArg } from '@docmost/prosemirror-markdown';
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Unit tests for the in-app `parseNodeArg` helper. It mirrors the standalone
|
* Unit tests for the shared `parseNodeArg` helper (#414: now the single copy in
|
||||||
* MCP helper (packages/mcp/src/lib/parse-node-arg.ts) and is used by the
|
* `@docmost/prosemirror-markdown`, imported by both the server tool adapters and
|
||||||
* patchNode / insertNode / updatePageJson tool adapters. Behavior must be
|
* `@docmost/mcp`). Used by the patchNode / insertNode / updatePageJson adapters.
|
||||||
* byte-identical: object passthrough, valid-string parse, invalid-string throw.
|
* Behavior: object passthrough, valid-string parse, invalid-string throw.
|
||||||
*/
|
*/
|
||||||
describe('parseNodeArg', () => {
|
describe('parseNodeArg', () => {
|
||||||
it('passes an object through unchanged', () => {
|
it('passes an object through unchanged', () => {
|
||||||
|
|||||||
@@ -1,26 +0,0 @@
|
|||||||
// The model sometimes serializes a ProseMirror node arg as a JSON string
|
|
||||||
// instead of an object. Normalize: parse a string to an object (throwing on
|
|
||||||
// invalid JSON), pass an object through unchanged. Shared by patchNode /
|
|
||||||
// insertNode (and the analogous updatePageJson content parsing).
|
|
||||||
//
|
|
||||||
// This is behaviorally identical to `packages/mcp/src/lib/parse-node-arg.ts`
|
|
||||||
// (the function logic, default/explicit throw messages and branch order match;
|
|
||||||
// only comments and quote style differ). We cannot import that helper here:
|
|
||||||
// `@docmost/mcp` is ESM-only and this server
|
|
||||||
// compiles with module:commonjs, so it is loaded at runtime via the
|
|
||||||
// `new Function('import()')` trick (see docmost-client.loader.ts). Sharing
|
|
||||||
// runtime code across that ESM/CJS boundary by a normal import is impossible,
|
|
||||||
// hence the mirrored copy.
|
|
||||||
export function parseNodeArg(
|
|
||||||
node: unknown,
|
|
||||||
errMsg = 'node was a string but not valid JSON',
|
|
||||||
): unknown {
|
|
||||||
if (typeof node === 'string') {
|
|
||||||
try {
|
|
||||||
return JSON.parse(node);
|
|
||||||
} catch {
|
|
||||||
throw new Error(errMsg);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return node;
|
|
||||||
}
|
|
||||||
@@ -5,16 +5,6 @@ import {
|
|||||||
} from '@nestjs/common';
|
} from '@nestjs/common';
|
||||||
import { CommentService } from './comment.service';
|
import { CommentService } from './comment.service';
|
||||||
import { AuditEvent, AuditResource } from '../../common/events/audit-events';
|
import { AuditEvent, AuditResource } from '../../common/events/audit-events';
|
||||||
import { QueueJob } from '../../integrations/queue/constants';
|
|
||||||
|
|
||||||
// #399: the resolve/unresolve flip and the ephemeral anchor removal are enqueued
|
|
||||||
// as COMMENT_MARK_UPDATE jobs (off the HTTP path), NOT awaited against the collab
|
|
||||||
// gateway. applyCommentSuggestion (the document TEXT edit) is untouched — it
|
|
||||||
// still runs synchronously via the gateway.
|
|
||||||
const markJob = (generalQueue: any, action: string) =>
|
|
||||||
generalQueue.add.mock.calls.find(
|
|
||||||
(c: any[]) => c[0] === QueueJob.COMMENT_MARK_UPDATE && c[1]?.action === action,
|
|
||||||
);
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Focused coverage for CommentService.applySuggestion (comment.service.ts).
|
* Focused coverage for CommentService.applySuggestion (comment.service.ts).
|
||||||
@@ -69,7 +59,6 @@ describe('CommentService — applySuggestion', () => {
|
|||||||
commentRepo,
|
commentRepo,
|
||||||
wsService,
|
wsService,
|
||||||
collaborationGateway,
|
collaborationGateway,
|
||||||
generalQueue,
|
|
||||||
auditService,
|
auditService,
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
@@ -97,15 +86,9 @@ describe('CommentService — applySuggestion', () => {
|
|||||||
|
|
||||||
// --- no replies → ephemeral delete branch -------------------------------
|
// --- no replies → ephemeral delete branch -------------------------------
|
||||||
|
|
||||||
it('applied=true, no replies → replaces text, hard-deletes, enqueues the anchor-mark removal, audits APPLIED, outcome=deleted', async () => {
|
it('applied=true, no replies → replaces text, hard-deletes, strips the anchor mark, audits APPLIED, outcome=deleted', async () => {
|
||||||
const {
|
const { service, commentRepo, wsService, collaborationGateway, auditService } =
|
||||||
service,
|
makeService({ applied: true, currentText: 'new text' });
|
||||||
commentRepo,
|
|
||||||
wsService,
|
|
||||||
collaborationGateway,
|
|
||||||
generalQueue,
|
|
||||||
auditService,
|
|
||||||
} = makeService({ applied: true, currentText: 'new text' });
|
|
||||||
|
|
||||||
const result = await service.applySuggestion(suggestionComment(), user());
|
const result = await service.applySuggestion(suggestionComment(), user());
|
||||||
|
|
||||||
@@ -122,20 +105,12 @@ describe('CommentService — applySuggestion', () => {
|
|||||||
);
|
);
|
||||||
|
|
||||||
// Ephemeral: the redundant comment is hard-deleted (atomic-conditional) and
|
// Ephemeral: the redundant comment is hard-deleted (atomic-conditional) and
|
||||||
// its inline anchor mark removal is ENQUEUED (#399), no longer a sync gateway
|
// its inline anchor mark removed via the deleteCommentMark collab event.
|
||||||
// call. The gateway was only touched for the applyCommentSuggestion text edit.
|
|
||||||
expect(commentRepo.deleteCommentIfChildless).toHaveBeenCalledWith('c-1');
|
expect(commentRepo.deleteCommentIfChildless).toHaveBeenCalledWith('c-1');
|
||||||
const del = markJob(generalQueue, 'delete');
|
expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith(
|
||||||
expect(del).toBeDefined();
|
|
||||||
expect(del[1]).toMatchObject({
|
|
||||||
documentName: 'page.page-1',
|
|
||||||
commentId: 'c-1',
|
|
||||||
userId: 'user-1',
|
|
||||||
});
|
|
||||||
expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalledWith(
|
|
||||||
'deleteCommentMark',
|
'deleteCommentMark',
|
||||||
expect.anything(),
|
'page.page-1',
|
||||||
expect.anything(),
|
expect.objectContaining({ commentId: 'c-1', user: expect.any(Object) }),
|
||||||
);
|
);
|
||||||
// No applied stamps are written for a row about to be deleted.
|
// No applied stamps are written for a row about to be deleted.
|
||||||
expect(appliedPatch(commentRepo)).toBeUndefined();
|
expect(appliedPatch(commentRepo)).toBeUndefined();
|
||||||
@@ -283,7 +258,7 @@ describe('CommentService — applySuggestion', () => {
|
|||||||
// The suggested text is already applied to the document, but between the
|
// The suggested text is already applied to the document, but between the
|
||||||
// hasChildren read and the atomic delete a reply landed. The parent must NOT
|
// hasChildren read and the atomic delete a reply landed. The parent must NOT
|
||||||
// be hard-deleted (cascade would destroy the reply); resolve the thread.
|
// be hard-deleted (cascade would destroy the reply); resolve the thread.
|
||||||
const { service, commentRepo, wsService, generalQueue } =
|
const { service, commentRepo, wsService, collaborationGateway } =
|
||||||
makeService({ applied: true, currentText: 'new text' }, false, 0);
|
makeService({ applied: true, currentText: 'new text' }, false, 0);
|
||||||
|
|
||||||
const result = await service.applySuggestion(suggestionComment(), user());
|
const result = await service.applySuggestion(suggestionComment(), user());
|
||||||
@@ -300,8 +275,11 @@ describe('CommentService — applySuggestion', () => {
|
|||||||
.map((c: any[]) => c[0])
|
.map((c: any[]) => c[0])
|
||||||
.find((p: any) => 'resolvedAt' in p);
|
.find((p: any) => 'resolvedAt' in p);
|
||||||
expect(resolvePatch.resolvedAt).toBeInstanceOf(Date);
|
expect(resolvePatch.resolvedAt).toBeInstanceOf(Date);
|
||||||
// The resolve mark is enqueued (#399), not a sync gateway call.
|
expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith(
|
||||||
expect(markJob(generalQueue, 'resolve')).toBeDefined();
|
'resolveCommentMark',
|
||||||
|
'page.page-1',
|
||||||
|
expect.objectContaining({ commentId: 'c-1', resolved: true }),
|
||||||
|
);
|
||||||
expect(result.outcome).toBe('resolved');
|
expect(result.outcome).toBe('resolved');
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
@@ -313,15 +313,11 @@ describe('CommentService — behavior', () => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
const [patch] = commentRepo.updateComment.mock.calls[0];
|
const [patch] = commentRepo.updateComment.mock.calls[0];
|
||||||
// #399: resolve/unresolve now also stamps updatedAt (the async mark
|
expect(patch).toEqual({
|
||||||
// worker's race-guard reads it to order out-of-order events). The
|
|
||||||
// resolve-state fields are still cleared to null on unresolve.
|
|
||||||
expect(patch).toMatchObject({
|
|
||||||
resolvedAt: null,
|
resolvedAt: null,
|
||||||
resolvedById: null,
|
resolvedById: null,
|
||||||
resolvedSource: null,
|
resolvedSource: null,
|
||||||
});
|
});
|
||||||
expect(patch.updatedAt).toBeInstanceOf(Date);
|
|
||||||
});
|
});
|
||||||
|
|
||||||
it("notifies the author when SOMEONE ELSE resolves their comment", async () => {
|
it("notifies the author when SOMEONE ELSE resolves their comment", async () => {
|
||||||
|
|||||||
@@ -1,15 +1,6 @@
|
|||||||
import { BadRequestException } from '@nestjs/common';
|
import { BadRequestException } from '@nestjs/common';
|
||||||
import { CommentService } from './comment.service';
|
import { CommentService } from './comment.service';
|
||||||
import { AuditEvent, AuditResource } from '../../common/events/audit-events';
|
import { AuditEvent, AuditResource } from '../../common/events/audit-events';
|
||||||
import { QueueJob } from '../../integrations/queue/constants';
|
|
||||||
|
|
||||||
// #399: the inline comment-mark op (resolve flip / ephemeral-suggestion anchor
|
|
||||||
// removal) is now enqueued as a COMMENT_MARK_UPDATE job instead of being awaited
|
|
||||||
// against the collab gateway on the HTTP path. Find that job by action.
|
|
||||||
const markJob = (generalQueue: any, action: string) =>
|
|
||||||
generalQueue.add.mock.calls.find(
|
|
||||||
(c: any[]) => c[0] === QueueJob.COMMENT_MARK_UPDATE && c[1]?.action === action,
|
|
||||||
);
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Coverage for CommentService.dismissSuggestion (#329). Dismiss ("Не применять")
|
* Coverage for CommentService.dismissSuggestion (#329). Dismiss ("Не применять")
|
||||||
@@ -53,14 +44,7 @@ describe('CommentService — dismissSuggestion', () => {
|
|||||||
auditService,
|
auditService,
|
||||||
);
|
);
|
||||||
|
|
||||||
return {
|
return { service, commentRepo, wsService, collaborationGateway, auditService };
|
||||||
service,
|
|
||||||
commentRepo,
|
|
||||||
wsService,
|
|
||||||
collaborationGateway,
|
|
||||||
generalQueue,
|
|
||||||
auditService,
|
|
||||||
};
|
|
||||||
}
|
}
|
||||||
|
|
||||||
const suggestionComment = (over?: Partial<any>): any => ({
|
const suggestionComment = (over?: Partial<any>): any => ({
|
||||||
@@ -78,30 +62,25 @@ describe('CommentService — dismissSuggestion', () => {
|
|||||||
});
|
});
|
||||||
const user = (over?: Partial<any>): any => ({ id: 'user-1', ...over });
|
const user = (over?: Partial<any>): any => ({ id: 'user-1', ...over });
|
||||||
|
|
||||||
it('no replies → hard-deletes, enqueues the anchor-mark removal, does NOT touch page text, audits DISMISSED, outcome=deleted', async () => {
|
it('no replies → hard-deletes, strips the anchor mark, does NOT touch page text, audits DISMISSED, outcome=deleted', async () => {
|
||||||
const {
|
const { service, commentRepo, wsService, collaborationGateway, auditService } =
|
||||||
service,
|
makeService(false);
|
||||||
commentRepo,
|
|
||||||
wsService,
|
|
||||||
collaborationGateway,
|
|
||||||
generalQueue,
|
|
||||||
auditService,
|
|
||||||
} = makeService(false);
|
|
||||||
|
|
||||||
const result = await service.dismissSuggestion(suggestionComment(), user());
|
const result = await service.dismissSuggestion(suggestionComment(), user());
|
||||||
|
|
||||||
// Never applies the suggestion to the document (no sync gateway call at all
|
// Never applies the suggestion to the document.
|
||||||
// now — the mark op is off the HTTP path, #399).
|
expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalledWith(
|
||||||
expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled();
|
'applyCommentSuggestion',
|
||||||
// Hard-delete (atomic-conditional) + enqueue the anchor-mark strip.
|
expect.anything(),
|
||||||
|
expect.anything(),
|
||||||
|
);
|
||||||
|
// Hard-delete (atomic-conditional) + strip mark.
|
||||||
expect(commentRepo.deleteCommentIfChildless).toHaveBeenCalledWith('c-1');
|
expect(commentRepo.deleteCommentIfChildless).toHaveBeenCalledWith('c-1');
|
||||||
const del = markJob(generalQueue, 'delete');
|
expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith(
|
||||||
expect(del).toBeDefined();
|
'deleteCommentMark',
|
||||||
expect(del[1]).toMatchObject({
|
'page.page-1',
|
||||||
documentName: 'page.page-1',
|
expect.objectContaining({ commentId: 'c-1', user: expect.any(Object) }),
|
||||||
commentId: 'c-1',
|
);
|
||||||
userId: 'user-1',
|
|
||||||
});
|
|
||||||
expect(wsService.emitCommentEvent).toHaveBeenCalledWith(
|
expect(wsService.emitCommentEvent).toHaveBeenCalledWith(
|
||||||
'space-1',
|
'space-1',
|
||||||
'page-1',
|
'page-1',
|
||||||
@@ -117,20 +96,20 @@ describe('CommentService — dismissSuggestion', () => {
|
|||||||
expect(result.outcome).toBe('deleted');
|
expect(result.outcome).toBe('deleted');
|
||||||
});
|
});
|
||||||
|
|
||||||
it('no replies → if the anchor-mark ENQUEUE FAILS, the row is NOT deleted and the error propagates (#329/#399: no orphan anchor)', async () => {
|
it('no replies → if the anchor-mark removal FAILS, the row is NOT deleted and the error propagates (#329: no orphan anchor)', async () => {
|
||||||
const { service, commentRepo, wsService, generalQueue } = makeService(false);
|
const { service, commentRepo, wsService, collaborationGateway } =
|
||||||
// #399: the mark removal now runs async in a worker, but the ENQUEUE is
|
makeService(false);
|
||||||
// awaited BEFORE the irreversible row delete — so the anchor-removal job is
|
// Mark removal is FATAL and runs BEFORE the irreversible row delete: a collab
|
||||||
// durably scheduled before the row can vanish. If even the enqueue fails
|
// failure (e.g. COLLAB_DISABLE_REDIS "no live instance") must abort the whole
|
||||||
// (e.g. Redis down), the whole operation aborts, leaving row + mark
|
// operation, leaving row + mark consistent — never a deleted row with an
|
||||||
// consistent — never a deleted row with an orphan anchor reporting success.
|
// orphan anchor left in the document reporting success.
|
||||||
generalQueue.add = jest.fn(async () => {
|
collaborationGateway.handleYjsEvent = jest.fn(async () => {
|
||||||
throw new Error('queue add failed: no redis');
|
throw new Error('requires a live collaboration instance');
|
||||||
});
|
});
|
||||||
|
|
||||||
await expect(
|
await expect(
|
||||||
service.dismissSuggestion(suggestionComment(), user()),
|
service.dismissSuggestion(suggestionComment(), user()),
|
||||||
).rejects.toThrow(/queue add failed/);
|
).rejects.toThrow(/live collaboration/);
|
||||||
|
|
||||||
expect(commentRepo.deleteCommentIfChildless).not.toHaveBeenCalled();
|
expect(commentRepo.deleteCommentIfChildless).not.toHaveBeenCalled();
|
||||||
expect(wsService.emitCommentEvent).not.toHaveBeenCalledWith(
|
expect(wsService.emitCommentEvent).not.toHaveBeenCalledWith(
|
||||||
@@ -141,29 +120,23 @@ describe('CommentService — dismissSuggestion', () => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
it('WITH replies → resolves (not delete), does NOT apply, audits DISMISSED, outcome=resolved', async () => {
|
it('WITH replies → resolves (not delete), does NOT apply, audits DISMISSED, outcome=resolved', async () => {
|
||||||
const {
|
const { service, commentRepo, wsService, collaborationGateway, auditService } =
|
||||||
service,
|
makeService(true);
|
||||||
commentRepo,
|
|
||||||
collaborationGateway,
|
|
||||||
generalQueue,
|
|
||||||
auditService,
|
|
||||||
} = makeService(true);
|
|
||||||
|
|
||||||
const result = await service.dismissSuggestion(suggestionComment(), user());
|
const result = await service.dismissSuggestion(suggestionComment(), user());
|
||||||
|
|
||||||
// Resolved via resolveComment (resolve patch + enqueued resolve mark), NOT
|
// Resolved via resolveComment (resolve patch + resolve mark), NOT deleted.
|
||||||
// deleted.
|
|
||||||
const resolvePatch = commentRepo.updateComment.mock.calls
|
const resolvePatch = commentRepo.updateComment.mock.calls
|
||||||
.map((c: any[]) => c[0])
|
.map((c: any[]) => c[0])
|
||||||
.find((p: any) => 'resolvedAt' in p);
|
.find((p: any) => 'resolvedAt' in p);
|
||||||
expect(resolvePatch.resolvedAt).toBeInstanceOf(Date);
|
expect(resolvePatch.resolvedAt).toBeInstanceOf(Date);
|
||||||
expect(resolvePatch.resolvedById).toBe('user-1');
|
expect(resolvePatch.resolvedById).toBe('user-1');
|
||||||
expect(commentRepo.deleteComment).not.toHaveBeenCalled();
|
expect(commentRepo.deleteComment).not.toHaveBeenCalled();
|
||||||
// No sync gateway call; the resolve mark is enqueued (#399).
|
expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith(
|
||||||
expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled();
|
'resolveCommentMark',
|
||||||
const res = markJob(generalQueue, 'resolve');
|
'page.page-1',
|
||||||
expect(res).toBeDefined();
|
expect.objectContaining({ commentId: 'c-1', resolved: true }),
|
||||||
expect(res[1]).toMatchObject({ documentName: 'page.page-1', commentId: 'c-1' });
|
);
|
||||||
// No applied stamp — dismiss does not apply the edit.
|
// No applied stamp — dismiss does not apply the edit.
|
||||||
const appliedPatch = commentRepo.updateComment.mock.calls
|
const appliedPatch = commentRepo.updateComment.mock.calls
|
||||||
.map((c: any[]) => c[0])
|
.map((c: any[]) => c[0])
|
||||||
@@ -183,7 +156,8 @@ describe('CommentService — dismissSuggestion', () => {
|
|||||||
// but the atomic delete matches 0 rows because a reply landed in the window
|
// but the atomic delete matches 0 rows because a reply landed in the window
|
||||||
// between that read and the delete. The parent must NOT be hard-deleted
|
// between that read and the delete. The parent must NOT be hard-deleted
|
||||||
// (a cascade would destroy the just-added reply); the thread is resolved.
|
// (a cascade would destroy the just-added reply); the thread is resolved.
|
||||||
const { service, commentRepo, wsService, generalQueue } = makeService(false, 0);
|
const { service, commentRepo, wsService, collaborationGateway } =
|
||||||
|
makeService(false, 0);
|
||||||
|
|
||||||
const result = await service.dismissSuggestion(suggestionComment(), user());
|
const result = await service.dismissSuggestion(suggestionComment(), user());
|
||||||
|
|
||||||
@@ -201,9 +175,11 @@ describe('CommentService — dismissSuggestion', () => {
|
|||||||
.find((p: any) => 'resolvedAt' in p);
|
.find((p: any) => 'resolvedAt' in p);
|
||||||
expect(resolvePatch.resolvedAt).toBeInstanceOf(Date);
|
expect(resolvePatch.resolvedAt).toBeInstanceOf(Date);
|
||||||
expect(resolvePatch.resolvedById).toBe('user-1');
|
expect(resolvePatch.resolvedById).toBe('user-1');
|
||||||
// A resolve mark job is enqueued (the anchor was already delete-marked; the
|
expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith(
|
||||||
// resolve mirror is idempotent — #399).
|
'resolveCommentMark',
|
||||||
expect(markJob(generalQueue, 'resolve')).toBeDefined();
|
'page.page-1',
|
||||||
|
expect.objectContaining({ commentId: 'c-1', resolved: true }),
|
||||||
|
);
|
||||||
expect(result.outcome).toBe('resolved');
|
expect(result.outcome).toBe('resolved');
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
@@ -1,179 +0,0 @@
|
|||||||
import { Logger } from '@nestjs/common';
|
|
||||||
import { CommentService } from './comment.service';
|
|
||||||
import { QueueJob } from '../../integrations/queue/constants';
|
|
||||||
|
|
||||||
// Flush pending microtasks so a fire-and-forget `.catch(...)` runs before we assert.
|
|
||||||
const flushMicrotasks = () => new Promise((r) => setImmediate(r));
|
|
||||||
|
|
||||||
/**
|
|
||||||
* #399: the comment inline-mark update is moved OFF the HTTP critical path.
|
|
||||||
* resolveComment / unresolve / the ephemeral-suggestion delete must NO LONGER
|
|
||||||
* await CollaborationGateway.handleYjsEvent (which loaded the whole Y.Doc and
|
|
||||||
* ran the store pipeline synchronously, ~4.5s p95). Instead they enqueue an
|
|
||||||
* idempotent COMMENT_MARK_UPDATE job onto the GENERAL_QUEUE with the payload the
|
|
||||||
* worker replays.
|
|
||||||
*
|
|
||||||
* The service is constructed directly with jest mocks (the @InjectQueue tokens
|
|
||||||
* cannot be resolved by Test.createTestingModule — see comment.service.spec.ts).
|
|
||||||
*/
|
|
||||||
describe('CommentService — async comment mark (#399)', () => {
|
|
||||||
function makeService() {
|
|
||||||
const commentRepo: any = {
|
|
||||||
findById: jest.fn(async (id: string) => ({
|
|
||||||
id,
|
|
||||||
content: {},
|
|
||||||
spaceId: 'space-1',
|
|
||||||
pageId: 'page-1',
|
|
||||||
})),
|
|
||||||
updateComment: jest.fn(async () => undefined),
|
|
||||||
hasChildren: jest.fn(async () => false),
|
|
||||||
deleteCommentIfChildless: jest.fn(async () => 1),
|
|
||||||
};
|
|
||||||
const pageRepo: any = {};
|
|
||||||
const wsService: any = { emitCommentEvent: jest.fn() };
|
|
||||||
// The gateway MUST NOT be touched on the HTTP path anymore.
|
|
||||||
const collaborationGateway: any = {
|
|
||||||
handleYjsEvent: jest.fn(async () => undefined),
|
|
||||||
};
|
|
||||||
const generalQueue: any = { add: jest.fn(() => Promise.resolve()) };
|
|
||||||
const notificationQueue: any = { add: jest.fn(async () => undefined) };
|
|
||||||
const auditService: any = { log: jest.fn() };
|
|
||||||
|
|
||||||
const service = new CommentService(
|
|
||||||
commentRepo,
|
|
||||||
pageRepo,
|
|
||||||
wsService,
|
|
||||||
collaborationGateway,
|
|
||||||
generalQueue,
|
|
||||||
notificationQueue,
|
|
||||||
auditService,
|
|
||||||
);
|
|
||||||
return {
|
|
||||||
service,
|
|
||||||
commentRepo,
|
|
||||||
collaborationGateway,
|
|
||||||
generalQueue,
|
|
||||||
auditService,
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
const comment = (over?: Partial<any>): any => ({
|
|
||||||
id: 'c-1',
|
|
||||||
creatorId: 'user-1',
|
|
||||||
pageId: 'page-1',
|
|
||||||
spaceId: 'space-1',
|
|
||||||
workspaceId: 'ws-1',
|
|
||||||
...over,
|
|
||||||
});
|
|
||||||
const user = (over?: Partial<any>): any => ({ id: 'user-1', ...over });
|
|
||||||
|
|
||||||
const markJob = (generalQueue: any) =>
|
|
||||||
generalQueue.add.mock.calls.find(
|
|
||||||
(c: any[]) => c[0] === QueueJob.COMMENT_MARK_UPDATE,
|
|
||||||
);
|
|
||||||
|
|
||||||
it('resolveComment does NOT call the gateway synchronously, and enqueues a resolve mark job', async () => {
|
|
||||||
const { service, collaborationGateway, generalQueue } = makeService();
|
|
||||||
|
|
||||||
await service.resolveComment(comment(), true, user());
|
|
||||||
|
|
||||||
// The whole point of #399: the Y.Doc mark op is off the HTTP path.
|
|
||||||
expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled();
|
|
||||||
|
|
||||||
const job = markJob(generalQueue);
|
|
||||||
expect(job).toBeDefined();
|
|
||||||
expect(job[0]).toBe(QueueJob.COMMENT_MARK_UPDATE);
|
|
||||||
expect(job[1]).toMatchObject({
|
|
||||||
documentName: 'page.page-1',
|
|
||||||
commentId: 'c-1',
|
|
||||||
action: 'resolve',
|
|
||||||
userId: 'user-1',
|
|
||||||
});
|
|
||||||
expect(typeof job[1].ts).toBe('number');
|
|
||||||
// ts equals the resolvedAt stamp written to the row (shared timestamp).
|
|
||||||
const [patch] = (service as any).commentRepo.updateComment.mock.calls[0];
|
|
||||||
expect(job[1].ts).toBe((patch.resolvedAt as Date).getTime());
|
|
||||||
expect(job[1].ts).toBe((patch.updatedAt as Date).getTime());
|
|
||||||
});
|
|
||||||
|
|
||||||
it('unresolve enqueues an unresolve mark job (action mapped from resolved=false)', async () => {
|
|
||||||
const { service, collaborationGateway, generalQueue } = makeService();
|
|
||||||
|
|
||||||
await service.resolveComment(comment(), false, user());
|
|
||||||
|
|
||||||
expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled();
|
|
||||||
const job = markJob(generalQueue);
|
|
||||||
expect(job[1]).toMatchObject({
|
|
||||||
documentName: 'page.page-1',
|
|
||||||
commentId: 'c-1',
|
|
||||||
action: 'unresolve',
|
|
||||||
userId: 'user-1',
|
|
||||||
});
|
|
||||||
});
|
|
||||||
|
|
||||||
it('dismissing a childless ephemeral suggestion enqueues a delete mark job (not a sync gateway call)', async () => {
|
|
||||||
const { service, collaborationGateway, generalQueue } = makeService();
|
|
||||||
|
|
||||||
await service.dismissSuggestion(
|
|
||||||
comment({ suggestedText: 'new text', selection: 'old', resolvedAt: null }),
|
|
||||||
user(),
|
|
||||||
);
|
|
||||||
|
|
||||||
// The anchor removal is queued, not awaited against the gateway.
|
|
||||||
expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled();
|
|
||||||
const job = markJob(generalQueue);
|
|
||||||
expect(job).toBeDefined();
|
|
||||||
expect(job[1]).toMatchObject({
|
|
||||||
documentName: 'page.page-1',
|
|
||||||
commentId: 'c-1',
|
|
||||||
action: 'delete',
|
|
||||||
userId: 'user-1',
|
|
||||||
});
|
|
||||||
expect(typeof job[1].ts).toBe('number');
|
|
||||||
});
|
|
||||||
|
|
||||||
it('awaits the delete ENQUEUE before the irreversible row hard-delete (ordering preserved)', async () => {
|
|
||||||
const { service, generalQueue, commentRepo } = makeService();
|
|
||||||
const order: string[] = [];
|
|
||||||
generalQueue.add.mockImplementation(async (name: string) => {
|
|
||||||
order.push(`enqueue:${name}`);
|
|
||||||
});
|
|
||||||
commentRepo.deleteCommentIfChildless.mockImplementation(async () => {
|
|
||||||
order.push('delete-row');
|
|
||||||
return 1;
|
|
||||||
});
|
|
||||||
|
|
||||||
await service.dismissSuggestion(
|
|
||||||
comment({ suggestedText: 'new text', selection: 'old', resolvedAt: null }),
|
|
||||||
user(),
|
|
||||||
);
|
|
||||||
|
|
||||||
// The mark-removal job must be durably queued BEFORE the row disappears.
|
|
||||||
expect(order).toEqual([
|
|
||||||
`enqueue:${QueueJob.COMMENT_MARK_UPDATE}`,
|
|
||||||
'delete-row',
|
|
||||||
]);
|
|
||||||
});
|
|
||||||
|
|
||||||
it('resolve is fire-and-forget: a queue-add rejection does NOT fail the HTTP call (best-effort warn)', async () => {
|
|
||||||
const { service, generalQueue } = makeService();
|
|
||||||
// The queue is unavailable — the whole point of #399 is that this must NOT
|
|
||||||
// propagate out of resolveComment onto the HTTP request.
|
|
||||||
const queueErr = new Error('queue is down');
|
|
||||||
generalQueue.add.mockRejectedValue(queueErr);
|
|
||||||
const warnSpy = jest
|
|
||||||
.spyOn(Logger.prototype, 'warn')
|
|
||||||
.mockImplementation(() => undefined);
|
|
||||||
|
|
||||||
// Must resolve, never throw, even though the enqueue rejects.
|
|
||||||
await expect(service.resolveComment(comment(), true, user())).resolves.not.toThrow();
|
|
||||||
|
|
||||||
// The rejection is swallowed on a microtask AFTER the method returns; flush it.
|
|
||||||
await flushMicrotasks();
|
|
||||||
expect(warnSpy).toHaveBeenCalledWith(
|
|
||||||
expect.stringContaining('Failed to enqueue comment mark update for comment c-1'),
|
|
||||||
queueErr,
|
|
||||||
);
|
|
||||||
warnSpy.mockRestore();
|
|
||||||
});
|
|
||||||
});
|
|
||||||
@@ -21,7 +21,6 @@ import { CursorPaginationResult } from '@docmost/db/pagination/cursor-pagination
|
|||||||
import { QueueJob, QueueName } from '../../integrations/queue/constants';
|
import { QueueJob, QueueName } from '../../integrations/queue/constants';
|
||||||
import { extractUserMentionIdsFromJson } from '../../common/helpers/prosemirror/utils';
|
import { extractUserMentionIdsFromJson } from '../../common/helpers/prosemirror/utils';
|
||||||
import {
|
import {
|
||||||
ICommentMarkUpdateJob,
|
|
||||||
ICommentNotificationJob,
|
ICommentNotificationJob,
|
||||||
ICommentResolvedNotificationJob,
|
ICommentResolvedNotificationJob,
|
||||||
} from '../../integrations/queue/constants/queue.interface';
|
} from '../../integrations/queue/constants/queue.interface';
|
||||||
@@ -299,11 +298,7 @@ export class CommentService {
|
|||||||
// source is cleared alongside resolvedAt/resolvedById.
|
// source is cleared alongside resolvedAt/resolvedById.
|
||||||
provenance?: AuthProvenanceData,
|
provenance?: AuthProvenanceData,
|
||||||
): Promise<Comment> {
|
): Promise<Comment> {
|
||||||
// One shared timestamp: it stamps resolvedAt AND updatedAt on the row and is
|
const resolvedAt = resolved ? new Date() : null;
|
||||||
// carried as the mark job's `ts`, so the worker's race-guard can order this
|
|
||||||
// event against the row's authoritative resolve-state mutation time (#399).
|
|
||||||
const now = new Date();
|
|
||||||
const resolvedAt = resolved ? now : null;
|
|
||||||
const resolvedById = resolved ? authUser.id : null;
|
const resolvedById = resolved ? authUser.id : null;
|
||||||
const isAgent = provenance?.actor === 'agent';
|
const isAgent = provenance?.actor === 'agent';
|
||||||
// Set the agent marker only when resolving; on unresolve clear it back to
|
// Set the agent marker only when resolving; on unresolve clear it back to
|
||||||
@@ -312,33 +307,25 @@ export class CommentService {
|
|||||||
const resolvedSource = resolved && isAgent ? 'agent' : null;
|
const resolvedSource = resolved && isAgent ? 'agent' : null;
|
||||||
|
|
||||||
await this.commentRepo.updateComment(
|
await this.commentRepo.updateComment(
|
||||||
// Bump updatedAt (not editedAt — that drives the "edited" badge) so the
|
{ resolvedAt, resolvedById, resolvedSource },
|
||||||
// row records WHEN the resolve state last changed; the async mark worker
|
|
||||||
// compares its job ts against this to skip a superseded out-of-order event.
|
|
||||||
{ resolvedAt, resolvedById, resolvedSource, updatedAt: now },
|
|
||||||
comment.id,
|
comment.id,
|
||||||
);
|
);
|
||||||
|
|
||||||
// #399: mirror the resolved state onto the inline comment mark OFF the HTTP
|
// Reflect the resolved state on the inline comment mark in the
|
||||||
// critical path. The DB row above is the source of truth (updated in ms); the
|
// collaborative document so all connected clients stay in sync.
|
||||||
// mark is an eventual mirror for connected clients, and its failure was
|
|
||||||
// ALREADY swallowed (best-effort warn) — so instead of awaiting the whole
|
|
||||||
// Y.Doc load + immediate store pipeline (~4.5s p95), enqueue an idempotent,
|
|
||||||
// retryable COMMENT_MARK_UPDATE job. (Store-pipeline cost itself is #348's
|
|
||||||
// scope, not duplicated here.)
|
|
||||||
const documentName = `page.${comment.pageId}`;
|
const documentName = `page.${comment.pageId}`;
|
||||||
void this.enqueueCommentMarkUpdate(
|
try {
|
||||||
documentName,
|
await this.collaborationGateway.handleYjsEvent(
|
||||||
comment.id,
|
'resolveCommentMark',
|
||||||
resolved ? 'resolve' : 'unresolve',
|
documentName,
|
||||||
now.getTime(),
|
{ commentId: comment.id, resolved, user: authUser },
|
||||||
authUser.id,
|
);
|
||||||
).catch((error) =>
|
} catch (error) {
|
||||||
this.logger.warn(
|
this.logger.warn(
|
||||||
`Failed to enqueue comment mark update for comment ${comment.id}`,
|
`Failed to update comment mark for comment ${comment.id}`,
|
||||||
error,
|
error,
|
||||||
),
|
);
|
||||||
);
|
}
|
||||||
|
|
||||||
// Notify the comment author when someone else resolves their comment.
|
// Notify the comment author when someone else resolves their comment.
|
||||||
if (resolved && comment.creatorId !== authUser.id) {
|
if (resolved && comment.creatorId !== authUser.id) {
|
||||||
@@ -684,54 +671,23 @@ export class CommentService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Schedule removal of the inline `comment` anchor mark from the collaborative
|
* Remove the inline `comment` mark for a comment from the collaborative
|
||||||
* document (ephemeral suggestion #329), OFF the HTTP critical path (#399).
|
* document. FATAL, NOT best-effort: unlike resolveComment (which keeps the row,
|
||||||
*
|
* so a failed mark update is recoverable), this is used before an irreversible
|
||||||
* ORDERING PRESERVED: we `await` the ENQUEUE (a fast Redis add), not the mark
|
* hard-delete, so the mark removal MUST succeed or throw. Under
|
||||||
* op, and the caller only proceeds to the irreversible row hard-delete after
|
* COLLAB_DISABLE_REDIS the gateway invokes the deleteCommentMark handler
|
||||||
* this resolves. So the anchor-removal job is DURABLY queued before the row
|
* directly (never a silent no-op) and a missing live instance surfaces as a
|
||||||
* vanishes — a queue-add failure throws here and aborts the delete (row + mark
|
* thrown error, which we let propagate so the caller aborts before deleting.
|
||||||
* stay consistent), preserving the invariant the old FATAL sync call gave. The
|
|
||||||
* mark op itself now runs async in the worker: it is idempotent and retried
|
|
||||||
* (3 attempts), so a transient collab failure self-heals; only an exhausted-
|
|
||||||
* retries job leaves a DB↔mark divergence, now VISIBLE via BullMQ failed-job
|
|
||||||
* metrics (was a hard 5xx before). Delete carries no state guard — the row is
|
|
||||||
* being removed, and stripping an absent mark is a no-op.
|
|
||||||
*/
|
*/
|
||||||
private async deleteCommentMark(comment: Comment, user: User): Promise<void> {
|
private async deleteCommentMark(comment: Comment, user: User): Promise<void> {
|
||||||
const documentName = `page.${comment.pageId}`;
|
const documentName = `page.${comment.pageId}`;
|
||||||
await this.enqueueCommentMarkUpdate(
|
await this.collaborationGateway.handleYjsEvent(
|
||||||
|
'deleteCommentMark',
|
||||||
documentName,
|
documentName,
|
||||||
comment.id,
|
{ commentId: comment.id, user },
|
||||||
'delete',
|
|
||||||
Date.now(),
|
|
||||||
user.id,
|
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* Enqueue an idempotent COMMENT_MARK_UPDATE job (#399) — the single path that
|
|
||||||
* mirrors a comment's inline-mark state into the collab Y.Doc off the HTTP
|
|
||||||
* response. The worker (GeneralQueueProcessor) runs the SAME handleYjsEvent
|
|
||||||
* the sync code used, so the mark op is byte-identical.
|
|
||||||
*/
|
|
||||||
private enqueueCommentMarkUpdate(
|
|
||||||
documentName: string,
|
|
||||||
commentId: string,
|
|
||||||
action: 'resolve' | 'unresolve' | 'delete',
|
|
||||||
ts: number,
|
|
||||||
userId: string,
|
|
||||||
): Promise<unknown> {
|
|
||||||
const jobData: ICommentMarkUpdateJob = {
|
|
||||||
documentName,
|
|
||||||
commentId,
|
|
||||||
action,
|
|
||||||
ts,
|
|
||||||
userId,
|
|
||||||
};
|
|
||||||
return this.generalQueue.add(QueueJob.COMMENT_MARK_UPDATE, jobData);
|
|
||||||
}
|
|
||||||
|
|
||||||
private async queueCommentNotification(
|
private async queueCommentNotification(
|
||||||
content: any,
|
content: any,
|
||||||
oldMentionIds: string[],
|
oldMentionIds: string[],
|
||||||
|
|||||||
@@ -61,9 +61,6 @@ export enum QueueJob {
|
|||||||
|
|
||||||
COMMENT_NOTIFICATION = 'comment-notification',
|
COMMENT_NOTIFICATION = 'comment-notification',
|
||||||
COMMENT_RESOLVED_NOTIFICATION = 'comment-resolved-notification',
|
COMMENT_RESOLVED_NOTIFICATION = 'comment-resolved-notification',
|
||||||
// #399: off-critical-path mirror of a comment's inline mark into the collab
|
|
||||||
// Y.Doc (resolve/unresolve flip, or ephemeral-suggestion anchor removal).
|
|
||||||
COMMENT_MARK_UPDATE = 'comment-mark-update',
|
|
||||||
PAGE_MENTION_NOTIFICATION = 'page-mention-notification',
|
PAGE_MENTION_NOTIFICATION = 'page-mention-notification',
|
||||||
PAGE_PERMISSION_GRANTED = 'page-permission-granted',
|
PAGE_PERMISSION_GRANTED = 'page-permission-granted',
|
||||||
PAGE_UPDATE_DIGEST = 'page-update-digest',
|
PAGE_UPDATE_DIGEST = 'page-update-digest',
|
||||||
|
|||||||
@@ -63,33 +63,6 @@ export interface ICommentNotificationJob {
|
|||||||
notifyWatchers: boolean;
|
notifyWatchers: boolean;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* GENERAL_QUEUE payload for the off-critical-path comment inline-mark mirror
|
|
||||||
* (#399). The comment DB row is the source of truth and is already updated
|
|
||||||
* synchronously (ms); this job flips/removes the inline `comment` mark in the
|
|
||||||
* collaborative Y.Doc for connected clients, OFF the HTTP response path, so
|
|
||||||
* `POST /api/comments/resolve` no longer waits the whole Y.Doc load + store
|
|
||||||
* pipeline (was ~4.5s p95). The mark op is idempotent, so BullMQ retries are
|
|
||||||
* safe.
|
|
||||||
*
|
|
||||||
* `action`:
|
|
||||||
* - 'resolve' / 'unresolve' → flip the mark's `resolved` attribute (exactly
|
|
||||||
* what the synchronous resolveCommentMark path did);
|
|
||||||
* - 'delete' → strip the anchor mark entirely (ephemeral suggestion #329).
|
|
||||||
* `ts` is the DB-mutation timestamp (ms). The worker's race-guard uses it (with
|
|
||||||
* the row's authoritative resolved state) to skip a resolve/unresolve event
|
|
||||||
* that a newer, opposite event has already superseded (out-of-order drain).
|
|
||||||
* `userId` supplies the connection-context user the store pipeline attributes
|
|
||||||
* the change to (persistence.extension reads context.user.id).
|
|
||||||
*/
|
|
||||||
export interface ICommentMarkUpdateJob {
|
|
||||||
documentName: string;
|
|
||||||
commentId: string;
|
|
||||||
action: 'resolve' | 'unresolve' | 'delete';
|
|
||||||
ts: number;
|
|
||||||
userId: string;
|
|
||||||
}
|
|
||||||
|
|
||||||
export interface ICommentResolvedNotificationJob {
|
export interface ICommentResolvedNotificationJob {
|
||||||
commentId: string;
|
commentId: string;
|
||||||
commentCreatorId: string;
|
commentCreatorId: string;
|
||||||
|
|||||||
-151
@@ -1,151 +0,0 @@
|
|||||||
import { Job } from 'bullmq';
|
|
||||||
import { GeneralQueueProcessor } from './general-queue.processor';
|
|
||||||
import { QueueJob } from '../constants';
|
|
||||||
import { ICommentMarkUpdateJob } from '../constants/queue.interface';
|
|
||||||
|
|
||||||
/**
|
|
||||||
* #399: the GENERAL_QUEUE worker replays the comment inline-mark op that used to
|
|
||||||
* run synchronously on the HTTP path. It must call the SAME gateway handler with
|
|
||||||
* the SAME semantics (resolve/unresolve → flip the `resolved` attribute; delete
|
|
||||||
* → strip the anchor), and its timestamp race-guard must skip an event a newer,
|
|
||||||
* opposite event already superseded.
|
|
||||||
*/
|
|
||||||
describe('GeneralQueueProcessor — COMMENT_MARK_UPDATE (#399)', () => {
|
|
||||||
function makeProc() {
|
|
||||||
const collaborationGateway: any = {
|
|
||||||
handleYjsEvent: jest.fn(async () => undefined),
|
|
||||||
};
|
|
||||||
const commentRepo: any = { findById: jest.fn() };
|
|
||||||
// #399: the processor resolves CollaborationGateway lazily via ModuleRef
|
|
||||||
// (strict:false) to avoid a DI cycle; the fake returns our gateway spy.
|
|
||||||
const moduleRef: any = { get: jest.fn(() => collaborationGateway) };
|
|
||||||
const proc = new GeneralQueueProcessor(
|
|
||||||
{} as any, // db
|
|
||||||
{} as any, // backlinkRepo
|
|
||||||
{} as any, // watcherRepo
|
|
||||||
commentRepo,
|
|
||||||
moduleRef,
|
|
||||||
);
|
|
||||||
return { proc, collaborationGateway, commentRepo };
|
|
||||||
}
|
|
||||||
|
|
||||||
const job = (data: ICommentMarkUpdateJob): Job =>
|
|
||||||
({ name: QueueJob.COMMENT_MARK_UPDATE, data }) as unknown as Job;
|
|
||||||
|
|
||||||
const base = {
|
|
||||||
documentName: 'page.page-1',
|
|
||||||
commentId: 'c-1',
|
|
||||||
userId: 'user-1',
|
|
||||||
};
|
|
||||||
|
|
||||||
it('resolve → resolveCommentMark with resolved:true and the same-shape args', async () => {
|
|
||||||
const { proc, collaborationGateway, commentRepo } = makeProc();
|
|
||||||
const ts = 1000;
|
|
||||||
// Row reflects the resolve (source of truth), stamped at the same ts.
|
|
||||||
commentRepo.findById.mockResolvedValue({
|
|
||||||
id: 'c-1',
|
|
||||||
resolvedAt: new Date(ts),
|
|
||||||
updatedAt: new Date(ts),
|
|
||||||
});
|
|
||||||
|
|
||||||
await proc.process(job({ ...base, action: 'resolve', ts }));
|
|
||||||
|
|
||||||
expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledTimes(1);
|
|
||||||
expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith(
|
|
||||||
'resolveCommentMark',
|
|
||||||
'page.page-1',
|
|
||||||
{ commentId: 'c-1', resolved: true, user: { id: 'user-1' } },
|
|
||||||
);
|
|
||||||
});
|
|
||||||
|
|
||||||
it('unresolve → resolveCommentMark with resolved:false', async () => {
|
|
||||||
const { proc, collaborationGateway, commentRepo } = makeProc();
|
|
||||||
const ts = 2000;
|
|
||||||
commentRepo.findById.mockResolvedValue({
|
|
||||||
id: 'c-1',
|
|
||||||
resolvedAt: null,
|
|
||||||
updatedAt: new Date(ts),
|
|
||||||
});
|
|
||||||
|
|
||||||
await proc.process(job({ ...base, action: 'unresolve', ts }));
|
|
||||||
|
|
||||||
expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith(
|
|
||||||
'resolveCommentMark',
|
|
||||||
'page.page-1',
|
|
||||||
{ commentId: 'c-1', resolved: false, user: { id: 'user-1' } },
|
|
||||||
);
|
|
||||||
});
|
|
||||||
|
|
||||||
it('delete → deleteCommentMark (strip the anchor), no row lookup / no state guard', async () => {
|
|
||||||
const { proc, collaborationGateway, commentRepo } = makeProc();
|
|
||||||
|
|
||||||
await proc.process(job({ ...base, action: 'delete', ts: 123 }));
|
|
||||||
|
|
||||||
expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith(
|
|
||||||
'deleteCommentMark',
|
|
||||||
'page.page-1',
|
|
||||||
{ commentId: 'c-1', user: { id: 'user-1' } },
|
|
||||||
);
|
|
||||||
// Delete carries no state guard — the row is (being) removed.
|
|
||||||
expect(commentRepo.findById).not.toHaveBeenCalled();
|
|
||||||
});
|
|
||||||
|
|
||||||
it('SKIPS a stale resolve superseded by a newer unresolve (row unresolved, job ts older)', async () => {
|
|
||||||
const { proc, collaborationGateway, commentRepo } = makeProc();
|
|
||||||
// A later unresolve already set the row: resolvedAt null, updatedAt = 5000.
|
|
||||||
commentRepo.findById.mockResolvedValue({
|
|
||||||
id: 'c-1',
|
|
||||||
resolvedAt: null,
|
|
||||||
updatedAt: new Date(5000),
|
|
||||||
});
|
|
||||||
|
|
||||||
// Stale resolve job enqueued at ts=1000 (< 5000), intends resolved=true,
|
|
||||||
// but the row's authoritative state is unresolved → skip.
|
|
||||||
await proc.process(job({ ...base, action: 'resolve', ts: 1000 }));
|
|
||||||
|
|
||||||
expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled();
|
|
||||||
});
|
|
||||||
|
|
||||||
it('SKIPS a stale unresolve superseded by a newer resolve', async () => {
|
|
||||||
const { proc, collaborationGateway, commentRepo } = makeProc();
|
|
||||||
commentRepo.findById.mockResolvedValue({
|
|
||||||
id: 'c-1',
|
|
||||||
resolvedAt: new Date(5000),
|
|
||||||
updatedAt: new Date(5000),
|
|
||||||
});
|
|
||||||
|
|
||||||
await proc.process(job({ ...base, action: 'unresolve', ts: 1000 }));
|
|
||||||
|
|
||||||
expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled();
|
|
||||||
});
|
|
||||||
|
|
||||||
it('applies when the row state agrees even if ts is older (idempotent, not a stale flip)', async () => {
|
|
||||||
const { proc, collaborationGateway, commentRepo } = makeProc();
|
|
||||||
// Row is resolved and its updatedAt is newer than the job ts, but the state
|
|
||||||
// AGREES with the job → this is a harmless idempotent replay, not a stale
|
|
||||||
// opposite event, so it must still apply.
|
|
||||||
commentRepo.findById.mockResolvedValue({
|
|
||||||
id: 'c-1',
|
|
||||||
resolvedAt: new Date(9000),
|
|
||||||
updatedAt: new Date(9000),
|
|
||||||
});
|
|
||||||
|
|
||||||
await proc.process(job({ ...base, action: 'resolve', ts: 1000 }));
|
|
||||||
|
|
||||||
expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith(
|
|
||||||
'resolveCommentMark',
|
|
||||||
'page.page-1',
|
|
||||||
{ commentId: 'c-1', resolved: true, user: { id: 'user-1' } },
|
|
||||||
);
|
|
||||||
});
|
|
||||||
|
|
||||||
it('skips (no throw) when the comment row has vanished', async () => {
|
|
||||||
const { proc, collaborationGateway, commentRepo } = makeProc();
|
|
||||||
commentRepo.findById.mockResolvedValue(undefined);
|
|
||||||
|
|
||||||
await expect(
|
|
||||||
proc.process(job({ ...base, action: 'resolve', ts: 1000 })),
|
|
||||||
).resolves.toBeUndefined();
|
|
||||||
expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled();
|
|
||||||
});
|
|
||||||
});
|
|
||||||
@@ -4,7 +4,6 @@ import { Job } from 'bullmq';
|
|||||||
import { QueueJob, QueueName } from '../constants';
|
import { QueueJob, QueueName } from '../constants';
|
||||||
import {
|
import {
|
||||||
IAddPageWatchersJob,
|
IAddPageWatchersJob,
|
||||||
ICommentMarkUpdateJob,
|
|
||||||
IPageBacklinkJob,
|
IPageBacklinkJob,
|
||||||
} from '../constants/queue.interface';
|
} from '../constants/queue.interface';
|
||||||
import { InjectKysely } from 'nestjs-kysely';
|
import { InjectKysely } from 'nestjs-kysely';
|
||||||
@@ -14,11 +13,8 @@ import {
|
|||||||
WatcherRepo,
|
WatcherRepo,
|
||||||
WatcherType,
|
WatcherType,
|
||||||
} from '@docmost/db/repos/watcher/watcher.repo';
|
} from '@docmost/db/repos/watcher/watcher.repo';
|
||||||
import { InsertableWatcher, User } from '@docmost/db/types/entity.types';
|
import { InsertableWatcher } from '@docmost/db/types/entity.types';
|
||||||
import { processBacklinks } from '../tasks/backlinks.task';
|
import { processBacklinks } from '../tasks/backlinks.task';
|
||||||
import { ModuleRef } from '@nestjs/core';
|
|
||||||
import { CollaborationGateway } from '../../../collaboration/collaboration.gateway';
|
|
||||||
import { CommentRepo } from '@docmost/db/repos/comment/comment.repo';
|
|
||||||
|
|
||||||
@Processor(QueueName.GENERAL_QUEUE)
|
@Processor(QueueName.GENERAL_QUEUE)
|
||||||
export class GeneralQueueProcessor
|
export class GeneralQueueProcessor
|
||||||
@@ -26,32 +22,14 @@ export class GeneralQueueProcessor
|
|||||||
implements OnModuleDestroy
|
implements OnModuleDestroy
|
||||||
{
|
{
|
||||||
private readonly logger = new Logger(GeneralQueueProcessor.name);
|
private readonly logger = new Logger(GeneralQueueProcessor.name);
|
||||||
// #399: CollaborationGateway lives in CollaborationModule. We resolve it lazily
|
|
||||||
// via ModuleRef instead of importing that module into the @Global QueueModule —
|
|
||||||
// CollaborationModule's own HistoryProcessor injects this module's global
|
|
||||||
// GENERAL_QUEUE token, so a static import edge here would form a DI cycle. A
|
|
||||||
// lazy strict:false lookup (cached) sidesteps it; the gateway is a singleton in
|
|
||||||
// both the API-server and collab processes that run this worker.
|
|
||||||
private collaborationGateway?: CollaborationGateway;
|
|
||||||
constructor(
|
constructor(
|
||||||
@InjectKysely() private readonly db: KyselyDB,
|
@InjectKysely() private readonly db: KyselyDB,
|
||||||
private readonly backlinkRepo: BacklinkRepo,
|
private readonly backlinkRepo: BacklinkRepo,
|
||||||
private readonly watcherRepo: WatcherRepo,
|
private readonly watcherRepo: WatcherRepo,
|
||||||
private readonly commentRepo: CommentRepo,
|
|
||||||
private readonly moduleRef: ModuleRef,
|
|
||||||
) {
|
) {
|
||||||
super();
|
super();
|
||||||
}
|
}
|
||||||
|
|
||||||
private getCollaborationGateway(): CollaborationGateway {
|
|
||||||
if (!this.collaborationGateway) {
|
|
||||||
this.collaborationGateway = this.moduleRef.get(CollaborationGateway, {
|
|
||||||
strict: false,
|
|
||||||
});
|
|
||||||
}
|
|
||||||
return this.collaborationGateway;
|
|
||||||
}
|
|
||||||
|
|
||||||
async process(job: Job): Promise<void> {
|
async process(job: Job): Promise<void> {
|
||||||
try {
|
try {
|
||||||
switch (job.name) {
|
switch (job.name) {
|
||||||
@@ -78,87 +56,12 @@ export class GeneralQueueProcessor
|
|||||||
);
|
);
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|
||||||
case QueueJob.COMMENT_MARK_UPDATE: {
|
|
||||||
await this.processCommentMarkUpdate(
|
|
||||||
job.data as ICommentMarkUpdateJob,
|
|
||||||
);
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
throw err;
|
throw err;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* #399: apply a comment's inline-mark mirror in the collab Y.Doc, off the HTTP
|
|
||||||
* critical path. Runs the SAME gateway path the synchronous comment.service
|
|
||||||
* code used (byte-identical mark op):
|
|
||||||
* - resolve / unresolve → resolveCommentMark (flip the `resolved` attribute);
|
|
||||||
* - delete → deleteCommentMark (strip the ephemeral-suggestion anchor #329).
|
|
||||||
* The op is idempotent, so a BullMQ retry is safe. Throwing propagates to
|
|
||||||
* WorkerHost → the job is retried and, on exhaustion, surfaces in failed-job
|
|
||||||
* metrics (the divergence is now visible rather than a silently-swallowed warn).
|
|
||||||
*/
|
|
||||||
private async processCommentMarkUpdate(
|
|
||||||
data: ICommentMarkUpdateJob,
|
|
||||||
): Promise<void> {
|
|
||||||
const { documentName, commentId, action, ts, userId } = data;
|
|
||||||
// Minimal connection-context user: the store pipeline reads context.user.id
|
|
||||||
// to attribute the change (persistence.extension). The mark mutation itself
|
|
||||||
// does not depend on the user, so the op stays byte-identical. Deliberate
|
|
||||||
// trade-off: the store pipeline's transient `page.updated` broadcast carries
|
|
||||||
// only { id } here, so its live "who edited" badge loses name/avatarUrl for
|
|
||||||
// this async mark replay. lastUpdatedById is still set correctly; the diff is
|
|
||||||
// cosmetic and self-heals on the next real edit — worth it to stay off the
|
|
||||||
// HTTP path and avoid re-loading the users row.
|
|
||||||
const user = { id: userId } as User;
|
|
||||||
|
|
||||||
if (action === 'delete') {
|
|
||||||
await this.getCollaborationGateway().handleYjsEvent(
|
|
||||||
'deleteCommentMark',
|
|
||||||
documentName,
|
|
||||||
{ commentId, user },
|
|
||||||
);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
// resolve / unresolve. The comment row is written SYNCHRONOUSLY before this
|
|
||||||
// job is enqueued, so it is the source of truth for the final resolved state
|
|
||||||
// and its updatedAt records when that state last changed. Race-guard: if a
|
|
||||||
// newer, OPPOSITE event has already superseded this one (its ts is older than
|
|
||||||
// the row's last resolve-state mutation AND the row's current resolved state
|
|
||||||
// disagrees with what this job intends — e.g. an unresolve that drained ahead
|
|
||||||
// of this resolve), skip it rather than flip the mark to a stale state.
|
|
||||||
const comment = await this.commentRepo.findById(commentId);
|
|
||||||
if (!comment) {
|
|
||||||
// The comment vanished (e.g. hard-deleted) → nothing left to mirror.
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
const wantResolved = action === 'resolve';
|
|
||||||
const rowResolved = comment.resolvedAt != null;
|
|
||||||
const rowMutatedAt = new Date(comment.updatedAt).getTime();
|
|
||||||
// `<=`, not `<`: on a sub-millisecond tie (two opposite toggles stamped in
|
|
||||||
// the same ms) skip the disagreeing job rather than let queue order decide.
|
|
||||||
// The consistent job (whose intent matches the row) short-circuits on the
|
|
||||||
// first condition, so a real update is never dropped; only a mark that both
|
|
||||||
// disagrees with the row AND is no newer than it is discarded.
|
|
||||||
if (rowResolved !== wantResolved && ts <= rowMutatedAt) {
|
|
||||||
this.logger.debug(
|
|
||||||
`Skipping stale comment mark '${action}' for ${commentId} ` +
|
|
||||||
`(job ts ${ts} < row ${rowMutatedAt}, row resolved=${rowResolved})`,
|
|
||||||
);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
await this.getCollaborationGateway().handleYjsEvent(
|
|
||||||
'resolveCommentMark',
|
|
||||||
documentName,
|
|
||||||
{ commentId, resolved: wantResolved, user },
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
@OnWorkerEvent('active')
|
@OnWorkerEvent('active')
|
||||||
onActive(job: Job) {
|
onActive(job: Job) {
|
||||||
this.logger.debug(`Processing ${job.name} job`);
|
this.logger.debug(`Processing ${job.name} job`);
|
||||||
|
|||||||
+39
-77
@@ -13,14 +13,12 @@ Read the **Gotchas** section before you trust any error count.
|
|||||||
- Agent chats live in Postgres, DB `docmost`, tables `ai_chat_*`.
|
- Agent chats live in Postgres, DB `docmost`, tables `ai_chat_*`.
|
||||||
- Each tool invocation is stored as **two** array elements (a `tool-call` part and
|
- Each tool invocation is stored as **two** array elements (a `tool-call` part and
|
||||||
a `tool-result` part), so naive counting double-counts.
|
a `tool-result` part), so naive counting double-counts.
|
||||||
- **A tool that *throws* writes no result part.** Since the #407 fix its error is
|
- **A tool that *throws* writes no result part at all.** Its error text is nowhere
|
||||||
persisted as a dedicated `{toolName, error}` element in `tool_calls` (queryable +
|
in the DB — not in `tool_calls`, `content`, or `metadata`. It is shown live in
|
||||||
replayed to the model). **Rows written before #407 still drop it** — the error is
|
the UI only. So `isError` / `success=false` scans under-report by design.
|
||||||
nowhere in the DB and shows only in the live UI. So `isError` / `success=false`
|
- To find where agents fail you need **three** sources: (1) soft-failure markers in
|
||||||
scans under-report by design, and pre-#407 thrown errors are invisible.
|
`tool_calls`, (2) the orphan-gap proxy for thrown errors, (3) server logs / the
|
||||||
- To find where agents fail: (1) soft-failure markers in `tool_calls`, (2) the new
|
live UI for the actual error text.
|
||||||
`error` field for thrown errors (new rows) / the orphan-gap proxy (old rows),
|
|
||||||
(3) server logs / the live UI for full stack traces beyond the truncated message.
|
|
||||||
|
|
||||||
## Where the data lives
|
## Where the data lives
|
||||||
|
|
||||||
@@ -63,17 +61,13 @@ index 0: { "toolName": "getPage", "input": { "pageId": "…" } } ← tool-ca
|
|||||||
index 1: { "toolName": "getPage", "output": { … } } ← tool-result (has output, NO input)
|
index 1: { "toolName": "getPage", "output": { … } } ← tool-result (has output, NO input)
|
||||||
```
|
```
|
||||||
|
|
||||||
The keys that appear on an element are `toolName`, `input`, `output`, and — for a
|
The **only** keys that ever appear on an element are `toolName`, `input`, `output`.
|
||||||
**thrown** failure on rows written after the #407 fix — `error` (the tool's error
|
There is no `state`, no `errorText`, no `type`. Consequences:
|
||||||
message; see the "Hard failures" section below). There is no `state`, no `errorText`,
|
|
||||||
no `type`. On pre-#407 rows a thrown failure has NO paired result element at all
|
|
||||||
(silent orphan). Consequences:
|
|
||||||
|
|
||||||
1. **Real invocation count = elements that have `output` or `error`.** Counting every
|
1. **Real invocation count = elements that have `output`.** Counting every element
|
||||||
element double-counts (you get ~2× and a spurious "~50% of every tool has no output").
|
double-counts (you get ~2× and a spurious "~50% of every tool has no output").
|
||||||
2. **Pairing:** a call = a `tool-call` part followed by its result part. A success
|
2. **Pairing:** a successful call = a `tool-call` part followed by its `tool-result`
|
||||||
carries `output`; a thrown failure (post-#407) carries `error` instead. Both carry
|
part. Both carry `toolName`, so you can group by tool on either.
|
||||||
`toolName`, so you can group by tool on either.
|
|
||||||
|
|
||||||
## The two classes of failure (and which the DB can see)
|
## The two classes of failure (and which the DB can see)
|
||||||
|
|
||||||
@@ -91,37 +85,25 @@ These are visible in the `tool-result` `output`. The marker differs per tool:
|
|||||||
Note `editPageText` returns `failed: []` on success — filtering on the *presence*
|
Note `editPageText` returns `failed: []` on success — filtering on the *presence*
|
||||||
of the key gives false positives; filter on **non-empty**.
|
of the key gives false positives; filter on **non-empty**.
|
||||||
|
|
||||||
### 2. Hard failures — tool THREW → NOW PERSISTED ✅ (since the #407 fix)
|
### 2. Hard failures — tool THREW → NOT PERSISTED ❌ (the trap)
|
||||||
|
|
||||||
When a tool throws (the classic one is `patchNode` / `insertNode` / `tableUpdateCell`
|
When a tool throws (the classic one is `patchNode` / `insertNode` / `tableUpdateCell`
|
||||||
→ `Failed to encode document to Yjs (fromJSON): Unknown node type: undefined`), the
|
→ `Failed to encode document to Yjs (fromJSON): Unknown node type: undefined`), the
|
||||||
runtime still writes **no `tool-result` part** — the failure is an ai@6 `tool-error`
|
runtime writes **no `tool-result` part**. The orphaned `tool-call` part stays, but
|
||||||
content part instead. **Since the #407 fix, that error is persisted**: `serializeSteps`
|
the error text is **nowhere in the DB**. It is streamed to the UI live and (until
|
||||||
appends a dedicated element `{toolName, error: "<message>"}` right after the failed
|
rotation) to server logs — that is it.
|
||||||
call, mirroring how a successful `{toolName, output}` element is appended. So a thrown
|
|
||||||
error now leaves a queryable `error` field carrying its (truncated) reason, and the
|
|
||||||
same real text is replayed to the model on the next turn (an `output-error` part with
|
|
||||||
the real `errorText`, no longer the `'Tool call did not complete.'` placeholder).
|
|
||||||
|
|
||||||
**Cutover caveat — old rows keep the old blind shape.** Rows written **before** this
|
So any query like `count(*) FILTER (WHERE output.success = false)` will happily
|
||||||
change have the two-part shape (`call` + `output` only) and simply **drop** thrown
|
return **0** for `patchNode` even when the chat is visibly full of red failures.
|
||||||
errors, leaving a silent **orphan** (a `call` with no `output` *and* no `error`). Rows
|
That is survivorship bias, not reliability.
|
||||||
written **after** the fix additionally carry the `error` element. So:
|
|
||||||
|
|
||||||
- **New rows:** query the `error` field directly (see the hard-error query below) — no
|
The only DB-side proxy for a thrown error is an **orphan**: a `tool-call` part with
|
||||||
orphan heuristic needed for thrown failures.
|
no matching `tool-result`. Caveat: orphans also appear when a run is **aborted**
|
||||||
- **Old rows (pre-#407):** the only DB-side proxy is still an **orphan**: a `tool-call`
|
mid-flight (server restart), so a high-volume tool (`createComment`, `searchInPage`,
|
||||||
part with no matching `tool-result` *and* no `error`. Orphans also appear when a run
|
`Search_web_search`) shows orphans from aborts, not from real errors. Treat the
|
||||||
is **aborted** mid-flight (server restart), so a high-volume tool (`createComment`,
|
orphan gap as an *upper bound* on hard errors, and cross-check the tool: a gap on a
|
||||||
`searchInPage`, `Search_web_search`) shows orphans from aborts, not real errors on
|
structural editor (`patchNode`, `insertNode`, `updatePageJson`, `transformPage`) is
|
||||||
old rows. Treat the orphan gap as an *upper bound*, and cross-check the tool: a gap on
|
almost certainly a thrown Yjs-encode error; a gap on `createComment` is mostly aborts.
|
||||||
a structural editor (`patchNode`, `insertNode`, `updatePageJson`, `transformPage`) is
|
|
||||||
almost certainly a thrown Yjs-encode error; a gap on `createComment` is mostly aborts.
|
|
||||||
|
|
||||||
A note on the aborted-call fallback: a call with **neither** a result **nor** a
|
|
||||||
`tool-error` (genuinely interrupted mid-step) still replays with the
|
|
||||||
`'Tool call did not complete.'` placeholder and persists as an orphan — that path is
|
|
||||||
unchanged, and is distinct from a real thrown error, which now carries `error`.
|
|
||||||
|
|
||||||
### 3. Run-level failures → `ai_chat_runs`
|
### 3. Run-level failures → `ai_chat_runs`
|
||||||
|
|
||||||
@@ -182,28 +164,14 @@ WHERE jsonb_typeof(o->'failed') = 'array'
|
|||||||
GROUP BY 1 ORDER BY 2 DESC;
|
GROUP BY 1 ORDER BY 2 DESC;
|
||||||
```
|
```
|
||||||
|
|
||||||
**Hard errors — persisted `error` field per tool (NEW rows, since #407)** — thrown
|
**Hard-error proxy — orphan gap per tool, WITH a spread column** (call parts minus
|
||||||
tool failures now carry their real reason, so query them directly:
|
result parts, plus how many distinct chats the gap is spread across):
|
||||||
|
|
||||||
```sql
|
|
||||||
SELECT elem->>'toolName' AS tool, count(*) AS thrown_errors,
|
|
||||||
min(elem->>'error') AS sample_error
|
|
||||||
FROM ai_chat_messages m, jsonb_array_elements(m.tool_calls) elem
|
|
||||||
WHERE jsonb_typeof(m.tool_calls) = 'array' AND elem ? 'error'
|
|
||||||
GROUP BY 1 ORDER BY 2 DESC;
|
|
||||||
```
|
|
||||||
|
|
||||||
**Hard-error proxy for OLD rows (pre-#407) — orphan gap per tool, WITH a spread column**
|
|
||||||
(call parts minus result parts, plus how many distinct chats the gap is spread across).
|
|
||||||
This covers rows written before thrown errors were persisted; on new rows a thrown
|
|
||||||
failure now has its own `error` element (use the query above) and an orphan means only
|
|
||||||
a genuinely aborted mid-step call:
|
|
||||||
|
|
||||||
```sql
|
```sql
|
||||||
WITH parts AS (
|
WITH parts AS (
|
||||||
SELECT m.chat_id, elem->>'toolName' AS tool,
|
SELECT m.chat_id, elem->>'toolName' AS tool,
|
||||||
(elem ? 'input' AND NOT (elem ? 'output')) AS is_call,
|
(elem ? 'input' AND NOT (elem ? 'output')) AS is_call,
|
||||||
(elem ? 'output' OR elem ? 'error') AS is_result
|
(elem ? 'output') AS is_result
|
||||||
FROM ai_chat_messages m, jsonb_array_elements(m.tool_calls) elem
|
FROM ai_chat_messages m, jsonb_array_elements(m.tool_calls) elem
|
||||||
WHERE jsonb_typeof(m.tool_calls) = 'array' AND m.role = 'assistant'
|
WHERE jsonb_typeof(m.tool_calls) = 'array' AND m.role = 'assistant'
|
||||||
),
|
),
|
||||||
@@ -220,12 +188,8 @@ HAVING sum(gap) FILTER (WHERE gap > 0) > 0
|
|||||||
ORDER BY missing_results DESC;
|
ORDER BY missing_results DESC;
|
||||||
```
|
```
|
||||||
|
|
||||||
The `is_result` predicate counts an `error` element as a paired result too, so on new
|
**`missing_results` mixes thrown errors AND aborted/interrupted runs — you cannot
|
||||||
rows a persisted thrown error no longer inflates the orphan gap; a remaining gap is an
|
split them from `output` alone** (a positional "what follows the orphan" heuristic
|
||||||
aborted/interrupted call.
|
|
||||||
|
|
||||||
**On OLD rows, `missing_results` mixes thrown errors AND aborted/interrupted runs — you
|
|
||||||
cannot split them from `output` alone** (a positional "what follows the orphan" heuristic
|
|
||||||
breaks on parallel tool batches, which persist as `call,call,…,result,result`). Use
|
breaks on parallel tool batches, which persist as `call,call,…,result,result`). Use
|
||||||
`chats_spread` to disambiguate:
|
`chats_spread` to disambiguate:
|
||||||
|
|
||||||
@@ -280,20 +244,18 @@ docker compose -p gitmost logs -f --tail=100 # whole stack
|
|||||||
```
|
```
|
||||||
|
|
||||||
Logging is `json-file`, `max-size=10m max-file=5` → ~50 MB retained, then rotated,
|
Logging is `json-file`, `max-size=10m max-file=5` → ~50 MB retained, then rotated,
|
||||||
and **wiped on container recreate**. Since the #407 fix, thrown-tool error text is
|
and **wiped on container recreate**. So thrown-tool error text is only reliably
|
||||||
**persisted in the `error` field** of `tool_calls` (see the hard-error query above), so
|
caught **in real time** (or in the live chat UI, which renders the failed part with
|
||||||
you no longer depend on live logs for it. Logs/live UI remain useful for **pre-#407
|
its message). There is no durable, queryable store of hard tool errors today — if you
|
||||||
rows** (whose thrown errors were dropped) and for full stack traces beyond the
|
need one, that is a feature to add (persist `output-error` parts, or emit a
|
||||||
truncated stored message. A per-tool `tool_calls_total{tool,status}` metric to
|
`tool_calls_total{tool,status}` metric to VictoriaMetrics).
|
||||||
VictoriaMetrics is still a possible future add for aggregate dashboards.
|
|
||||||
|
|
||||||
## Gotchas checklist
|
## Gotchas checklist
|
||||||
|
|
||||||
- [ ] Counting every `tool_calls` element → **overcount**. Count `output` elements; add `error` elements for thrown failures (new rows), but don't count both as invocations.
|
- [ ] Counting every `tool_calls` element → **2× overcount**. Count elements with `output`.
|
||||||
- [ ] `isError` / `success=false` ≈ 0 does **not** mean "no errors" — thrown errors are a separate `error` element (new rows) or dropped entirely (pre-#407 rows).
|
- [ ] `isError` / `success=false` ≈ 0 does **not** mean "no errors" — thrown errors aren't persisted.
|
||||||
- [ ] Thrown errors persist only on rows written **after the #407 fix** — pre-#407 rows still drop them (orphan only). Mind the cutover when trending over time.
|
|
||||||
- [ ] `editPageText.failed` is `[]` on success — test for **non-empty**, not presence.
|
- [ ] `editPageText.failed` is `[]` on success — test for **non-empty**, not presence.
|
||||||
- [ ] Orphan gap on OLD rows mixes thrown errors **and** aborted runs — split by tool. On NEW rows a thrown error is its own `error` element, so a gap ≈ aborted call.
|
- [ ] Orphan gap mixes thrown errors **and** aborted runs — split by tool before concluding.
|
||||||
- [ ] `aborted` runs = server restarts, `failed` runs = provider overload — not agent mistakes.
|
- [ ] `aborted` runs = server restarts, `failed` runs = provider overload — not agent mistakes.
|
||||||
- [ ] Never dump a raw `tool_calls` cell — it can be hundreds of KB.
|
- [ ] Never dump a raw `tool_calls` cell — it can be hundreds of KB.
|
||||||
- [ ] Logs are ephemeral (≤50 MB, wiped on recreate) — grab hard-error text live.
|
- [ ] Logs are ephemeral (≤50 MB, wiped on recreate) — grab hard-error text live.
|
||||||
|
|||||||
+1
-2
@@ -97,8 +97,7 @@
|
|||||||
"patchedDependencies": {
|
"patchedDependencies": {
|
||||||
"scimmy@1.3.5": "patches/scimmy@1.3.5.patch",
|
"scimmy@1.3.5": "patches/scimmy@1.3.5.patch",
|
||||||
"yjs@13.6.30": "patches/yjs@13.6.30.patch",
|
"yjs@13.6.30": "patches/yjs@13.6.30.patch",
|
||||||
"ai@6.0.134": "patches/ai@6.0.134.patch",
|
"ai@6.0.134": "patches/ai@6.0.134.patch"
|
||||||
"@hocuspocus/server@3.4.4": "patches/@hocuspocus__server@3.4.4.patch"
|
|
||||||
},
|
},
|
||||||
"overrides": {
|
"overrides": {
|
||||||
"prosemirror-changeset": "2.4.0",
|
"prosemirror-changeset": "2.4.0",
|
||||||
|
|||||||
@@ -52,7 +52,6 @@
|
|||||||
"form-data": "^4.0.0",
|
"form-data": "^4.0.0",
|
||||||
"jsdom": "^27.4.0",
|
"jsdom": "^27.4.0",
|
||||||
"marked": "^17.0.1",
|
"marked": "^17.0.1",
|
||||||
"pako": "^2.0.3",
|
|
||||||
"re2": "^1.21.0",
|
"re2": "^1.21.0",
|
||||||
"ws": "^8.19.0",
|
"ws": "^8.19.0",
|
||||||
"y-prosemirror": "1.3.7",
|
"y-prosemirror": "1.3.7",
|
||||||
|
|||||||
+189
-597
@@ -8,6 +8,10 @@ import {
|
|||||||
filterComment,
|
filterComment,
|
||||||
filterSearchResult,
|
filterSearchResult,
|
||||||
} from "./lib/filters.js";
|
} from "./lib/filters.js";
|
||||||
|
import { HocuspocusProvider } from "@hocuspocus/provider";
|
||||||
|
import { TiptapTransformer } from "@hocuspocus/transformer";
|
||||||
|
import * as Y from "yjs";
|
||||||
|
import WebSocket from "ws";
|
||||||
import { convertProseMirrorToMarkdown } from "./lib/markdown-converter.js";
|
import { convertProseMirrorToMarkdown } from "./lib/markdown-converter.js";
|
||||||
import {
|
import {
|
||||||
collectInternalFileNodes,
|
collectInternalFileNodes,
|
||||||
@@ -20,10 +24,11 @@ import {
|
|||||||
markdownToProseMirror,
|
markdownToProseMirror,
|
||||||
markdownToProseMirrorCanonical,
|
markdownToProseMirrorCanonical,
|
||||||
mutatePageContent,
|
mutatePageContent,
|
||||||
|
buildCollabWsUrl,
|
||||||
assertYjsEncodable,
|
assertYjsEncodable,
|
||||||
|
applyDocToFragment,
|
||||||
MutationResult,
|
MutationResult,
|
||||||
} from "./lib/collaboration.js";
|
} from "./lib/collaboration.js";
|
||||||
import { acquireCollabSession } from "./lib/collab-session.js";
|
|
||||||
import { footnoteWarningsField } from "./lib/footnote-analyze.js";
|
import { footnoteWarningsField } from "./lib/footnote-analyze.js";
|
||||||
import { buildPageTree } from "./lib/tree.js";
|
import { buildPageTree } from "./lib/tree.js";
|
||||||
import {
|
import {
|
||||||
@@ -35,25 +40,15 @@ import {
|
|||||||
deleteNodeById,
|
deleteNodeById,
|
||||||
assertUnambiguousMatch,
|
assertUnambiguousMatch,
|
||||||
insertNodeRelative,
|
insertNodeRelative,
|
||||||
blockPlainText,
|
|
||||||
buildOutline,
|
buildOutline,
|
||||||
getNodeByRef,
|
getNodeByRef,
|
||||||
readTable,
|
readTable,
|
||||||
insertTableRow,
|
insertTableRow,
|
||||||
deleteTableRow,
|
deleteTableRow,
|
||||||
updateTableCell,
|
updateTableCell,
|
||||||
} from "./lib/node-ops.js";
|
} from "@docmost/prosemirror-markdown";
|
||||||
import { searchInDoc, SearchOptions } from "./lib/page-search.js";
|
import { searchInDoc, SearchOptions } from "./lib/page-search.js";
|
||||||
import { withPageLock } from "./lib/page-lock.js";
|
import { withPageLock } from "./lib/page-lock.js";
|
||||||
import {
|
|
||||||
prepareModel,
|
|
||||||
decodeDrawioSvg,
|
|
||||||
buildDrawioSvg,
|
|
||||||
mxHash,
|
|
||||||
normalizeXml,
|
|
||||||
countUserCells,
|
|
||||||
} from "./lib/drawio-xml.js";
|
|
||||||
import { renderDiagramShapes } from "./lib/drawio-preview.js";
|
|
||||||
import {
|
import {
|
||||||
applyTextEdits,
|
applyTextEdits,
|
||||||
TextEdit,
|
TextEdit,
|
||||||
@@ -64,12 +59,10 @@ import { getCollabToken, performLogin } from "./lib/auth-utils.js";
|
|||||||
import { diffDocs, summarizeChange } from "./lib/diff.js";
|
import { diffDocs, summarizeChange } from "./lib/diff.js";
|
||||||
import {
|
import {
|
||||||
applyAnchorInDoc,
|
applyAnchorInDoc,
|
||||||
|
canAnchorInDoc,
|
||||||
countAnchorMatches,
|
countAnchorMatches,
|
||||||
getAnchoredText,
|
getAnchoredText,
|
||||||
resolveAnchorSelection,
|
|
||||||
normalizeForMatch,
|
|
||||||
} from "./lib/comment-anchor.js";
|
} from "./lib/comment-anchor.js";
|
||||||
import { closestBlockHint } from "./lib/text-normalize.js";
|
|
||||||
import {
|
import {
|
||||||
blockText,
|
blockText,
|
||||||
walk,
|
walk,
|
||||||
@@ -424,32 +417,177 @@ export class DocmostClient {
|
|||||||
* change report. The report is computed AFTER the atomic read->write and
|
* change report. The report is computed AFTER the atomic read->write and
|
||||||
* never throws.
|
* never throws.
|
||||||
*/
|
*/
|
||||||
private async mutateLiveContentUnlocked(
|
private mutateLiveContentUnlocked(
|
||||||
pageId: string,
|
pageId: string,
|
||||||
collabToken: string,
|
collabToken: string,
|
||||||
transform: (liveDoc: any) => any | null,
|
transform: (liveDoc: any) => any | null,
|
||||||
): Promise<MutationResult> {
|
): Promise<MutationResult> {
|
||||||
// Reuse a live CollabSession for the page (issue #400) instead of opening a
|
const CONNECT_TIMEOUT_MS = 25000;
|
||||||
// fresh provider per op. acquireCollabSession does NOT take the per-page
|
const PERSIST_TIMEOUT_MS = 20000;
|
||||||
// lock — the caller (replaceImage) already holds ONE withPageLock across its
|
const ydoc = new Y.Doc();
|
||||||
// scan -> upload -> write sequence, and the mutex is not reentrant, so
|
const wsUrl = buildCollabWsUrl(this.apiUrl);
|
||||||
// taking it here would deadlock. The synchronous read->write section and the
|
|
||||||
// unsyncedChanges/connectionLost ack logic live in CollabSession.mutate,
|
return new Promise<MutationResult>((resolve, reject) => {
|
||||||
// preserved verbatim from the old inline machine (incl. the #152 structural
|
let provider: HocuspocusProvider | undefined;
|
||||||
// diff that keeps a live editor's cursor anchored).
|
let applied = false; // onSynced may fire again on reconnect — apply once.
|
||||||
const session = await acquireCollabSession(pageId, collabToken, this.apiUrl, {
|
let settled = false;
|
||||||
// Only the actual 25s collab connect timeout emits this — the connect-vs-
|
let connectionLost = false;
|
||||||
// unload signal; the other failure paths must NOT emit it.
|
let connectTimer: ReturnType<typeof setTimeout> | undefined;
|
||||||
onConnectTimeout: () =>
|
let persistTimer: ReturnType<typeof setTimeout> | undefined;
|
||||||
this.onMetricFn?.("collab_connect_timeouts_total", 1),
|
let unsyncedHandler: ((data: { number: number }) => void) | undefined;
|
||||||
|
// The verifiable result resolved on every success/abort path. Set on abort
|
||||||
|
// (no-op report) and after a real write (computed change report).
|
||||||
|
let mutationResult: MutationResult;
|
||||||
|
|
||||||
|
const cleanup = () => {
|
||||||
|
if (connectTimer) clearTimeout(connectTimer);
|
||||||
|
if (persistTimer) clearTimeout(persistTimer);
|
||||||
|
if (provider) {
|
||||||
|
if (unsyncedHandler) {
|
||||||
|
try {
|
||||||
|
provider.off("unsyncedChanges", unsyncedHandler);
|
||||||
|
} catch (err) {}
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
provider.destroy();
|
||||||
|
} catch (err) {}
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
const finish = (err: Error | null, value?: MutationResult) => {
|
||||||
|
if (settled) return;
|
||||||
|
settled = true;
|
||||||
|
cleanup();
|
||||||
|
if (err) reject(err);
|
||||||
|
else resolve(value as MutationResult);
|
||||||
|
};
|
||||||
|
|
||||||
|
connectTimer = setTimeout(() => {
|
||||||
|
// Only the actual 25s collab connect timeout fires here — the agent's
|
||||||
|
// collab connection to the server never became ready. This is the
|
||||||
|
// connect-vs-unload signal; the other finish() paths must NOT emit it.
|
||||||
|
this.onMetricFn?.("collab_connect_timeouts_total", 1);
|
||||||
|
finish(new Error("Connection timeout to collaboration server"));
|
||||||
|
}, CONNECT_TIMEOUT_MS);
|
||||||
|
|
||||||
|
const waitForPersistence = () => {
|
||||||
|
if (settled) return;
|
||||||
|
if (!provider) {
|
||||||
|
finish(new Error("collab provider gone before persistence"));
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
if (provider.unsyncedChanges === 0) {
|
||||||
|
finish(null, mutationResult);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
persistTimer = setTimeout(() => {
|
||||||
|
finish(
|
||||||
|
new Error(
|
||||||
|
"Timeout waiting for collaboration server to persist the update",
|
||||||
|
),
|
||||||
|
);
|
||||||
|
}, PERSIST_TIMEOUT_MS);
|
||||||
|
unsyncedHandler = (data: { number: number }) => {
|
||||||
|
if (data.number === 0 && !connectionLost) {
|
||||||
|
finish(null, mutationResult);
|
||||||
|
}
|
||||||
|
};
|
||||||
|
provider.on("unsyncedChanges", unsyncedHandler);
|
||||||
|
};
|
||||||
|
|
||||||
|
provider = new HocuspocusProvider({
|
||||||
|
url: wsUrl,
|
||||||
|
name: `page.${pageId}`,
|
||||||
|
document: ydoc,
|
||||||
|
token: collabToken,
|
||||||
|
// @ts-ignore - Required for Node.js environment
|
||||||
|
WebSocketPolyfill: WebSocket,
|
||||||
|
onDisconnect: () => {
|
||||||
|
connectionLost = true;
|
||||||
|
finish(
|
||||||
|
new Error(
|
||||||
|
"Collaboration connection closed before the update was persisted/synced",
|
||||||
|
),
|
||||||
|
);
|
||||||
|
},
|
||||||
|
onClose: () => {
|
||||||
|
connectionLost = true;
|
||||||
|
finish(
|
||||||
|
new Error(
|
||||||
|
"Collaboration connection closed before the update was persisted/synced",
|
||||||
|
),
|
||||||
|
);
|
||||||
|
},
|
||||||
|
onSynced: () => {
|
||||||
|
if (applied || settled) return;
|
||||||
|
applied = true;
|
||||||
|
|
||||||
|
// CRITICAL: keep everything between reading and writing the live doc
|
||||||
|
// synchronous (no await) so no remote update can interleave.
|
||||||
|
let newDoc: any;
|
||||||
|
let beforeDoc: any;
|
||||||
|
try {
|
||||||
|
let liveDoc = TiptapTransformer.fromYdoc(ydoc, "default");
|
||||||
|
if (
|
||||||
|
!liveDoc ||
|
||||||
|
typeof liveDoc !== "object" ||
|
||||||
|
!Array.isArray(liveDoc.content)
|
||||||
|
) {
|
||||||
|
liveDoc = { type: "doc", content: [] };
|
||||||
|
}
|
||||||
|
|
||||||
|
// Snapshot the before-doc for the change report (safe deep clone).
|
||||||
|
beforeDoc = JSON.parse(JSON.stringify(liveDoc));
|
||||||
|
|
||||||
|
newDoc = transform(liveDoc);
|
||||||
|
|
||||||
|
if (newDoc == null) {
|
||||||
|
// Transform aborted — write nothing, return the live doc with a
|
||||||
|
// no-op change report.
|
||||||
|
mutationResult = {
|
||||||
|
doc: liveDoc,
|
||||||
|
verify: {
|
||||||
|
changed: false,
|
||||||
|
textInserted: 0,
|
||||||
|
textDeleted: 0,
|
||||||
|
blocksChanged: 0,
|
||||||
|
marks: {},
|
||||||
|
summary: "no changes (transform aborted)",
|
||||||
|
},
|
||||||
|
};
|
||||||
|
finish(null, mutationResult);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Structural diff into the live fragment (issue #152), mirroring
|
||||||
|
// the main write path: preserves the Yjs ids of unchanged nodes so
|
||||||
|
// an open editor's cursor is not yanked to the end of the document.
|
||||||
|
// The previous destructive rewrite (delete-all + applyUpdate of a
|
||||||
|
// fresh Y.Doc) discarded every node id, so replaceImage — the only
|
||||||
|
// caller of this method — still reproduced the #152 cursor jump
|
||||||
|
// (#164). applyDocToFragment runs its own atomic `transact`.
|
||||||
|
applyDocToFragment(ydoc, newDoc);
|
||||||
|
} catch (e) {
|
||||||
|
finish(e instanceof Error ? e : new Error(String(e)));
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Compute the verifiable change report AFTER the transact write: it
|
||||||
|
// only needs the JSON before/after, so it cannot affect the atomic
|
||||||
|
// read->write window, and summarizeChange never throws.
|
||||||
|
mutationResult = {
|
||||||
|
doc: newDoc,
|
||||||
|
verify: summarizeChange(beforeDoc, newDoc),
|
||||||
|
};
|
||||||
|
waitForPersistence();
|
||||||
|
},
|
||||||
|
onAuthenticationFailed: () => {
|
||||||
|
finish(
|
||||||
|
new Error("Authentication failed for collaboration connection"),
|
||||||
|
);
|
||||||
|
},
|
||||||
|
});
|
||||||
});
|
});
|
||||||
try {
|
|
||||||
return await session.mutate(transform);
|
|
||||||
} catch (e) {
|
|
||||||
// Drop the session on any failure so the next call reconnects fresh.
|
|
||||||
session.destroy("mutate failed");
|
|
||||||
throw e;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -2336,64 +2474,6 @@ export class DocmostClient {
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Plain text of each TOP-LEVEL block of `doc`, for anchor-failure hints. */
|
|
||||||
private topLevelBlockTexts(doc: any): string[] {
|
|
||||||
const content = doc && Array.isArray(doc.content) ? doc.content : [];
|
|
||||||
return content
|
|
||||||
.map((b: any) => blockPlainText(b))
|
|
||||||
.filter((t: string) => t.length > 0);
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* True when per-block anchoring failed but the (normalized) selection DOES
|
|
||||||
* appear in the blocks' joined plain text — i.e. it straddles a block
|
|
||||||
* boundary. Blocks are joined with a newline (collapsed to one space by
|
|
||||||
* normalizeForMatch) so a selection whose parts are separated by a paragraph
|
|
||||||
* break still matches. Callers only reach here after single-block anchoring
|
|
||||||
* (incl. the markdown-strip fallback) has already failed.
|
|
||||||
*/
|
|
||||||
private selectionSpansMultipleBlocks(
|
|
||||||
blockTexts: string[],
|
|
||||||
selection: string,
|
|
||||||
): boolean {
|
|
||||||
const normSel = normalizeForMatch(selection).norm.trim();
|
|
||||||
if (normSel.length === 0) return false;
|
|
||||||
const joined = normalizeForMatch(blockTexts.join("\n")).norm;
|
|
||||||
return joined.indexOf(normSel) !== -1;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Build the actionable error for a create_comment anchor MISS, porting
|
|
||||||
* edit_page_text's self-correction affordances: an explicit "spans multiple
|
|
||||||
* blocks" message when the selection straddles a block boundary, otherwise a
|
|
||||||
* "closest block text" hint quoting the block that holds the selection's
|
|
||||||
* longest token. `live` switches the wording between the pre-check (reading the
|
|
||||||
* persisted page) and the post-create live-anchor failure (which rolls back).
|
|
||||||
*/
|
|
||||||
private anchorNotFoundError(
|
|
||||||
doc: any,
|
|
||||||
selection: string,
|
|
||||||
live: boolean,
|
|
||||||
): Error {
|
|
||||||
const blockTexts = this.topLevelBlockTexts(doc);
|
|
||||||
const rolled = live ? " The comment was rolled back." : "";
|
|
||||||
if (this.selectionSpansMultipleBlocks(blockTexts, selection)) {
|
|
||||||
return new Error(
|
|
||||||
"create_comment: the selection spans multiple blocks; anchor on a " +
|
|
||||||
"contiguous fragment within a SINGLE paragraph/block (<=250 chars)." +
|
|
||||||
rolled,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
const where = live ? "in the live document" : "in the page";
|
|
||||||
return new Error(
|
|
||||||
`create_comment: could not find the selection text ${where} to anchor ` +
|
|
||||||
"the comment. Provide the EXACT contiguous text from a single " +
|
|
||||||
"paragraph/block (<=250 chars)." +
|
|
||||||
closestBlockHint(blockTexts, selection) +
|
|
||||||
rolled,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Create an inline comment anchored to its `selection` text, or a reply.
|
* Create an inline comment anchored to its `selection` text, or a reply.
|
||||||
*
|
*
|
||||||
@@ -2455,10 +2535,6 @@ export class DocmostClient {
|
|||||||
// Captured in the pre-check below (which already reads the page) and used as
|
// Captured in the pre-check below (which already reads the page) and used as
|
||||||
// payload.selection. Ordinary comments keep sending the raw agent selection.
|
// payload.selection. Ordinary comments keep sending the raw agent selection.
|
||||||
let anchoredSelection: string | null = null;
|
let anchoredSelection: string | null = null;
|
||||||
// Set when the anchor matched only after stripping markdown from the
|
|
||||||
// selection (the strip fallback); surfaced as a soft warning like
|
|
||||||
// edit_page_text does, so a stale-markdown selection is flagged.
|
|
||||||
let anchorNormalized = false;
|
|
||||||
|
|
||||||
// For a top-level comment, fail BEFORE creating anything when the selection
|
// For a top-level comment, fail BEFORE creating anything when the selection
|
||||||
// is not present in the persisted document — this avoids leaving an orphan
|
// is not present in the persisted document — this avoids leaving an orphan
|
||||||
@@ -2474,7 +2550,10 @@ export class DocmostClient {
|
|||||||
// rejected BEFORE creating the comment.
|
// rejected BEFORE creating the comment.
|
||||||
const matches = countAnchorMatches(page.content, selection);
|
const matches = countAnchorMatches(page.content, selection);
|
||||||
if (matches === 0) {
|
if (matches === 0) {
|
||||||
throw this.anchorNotFoundError(page.content, selection, false);
|
throw new Error(
|
||||||
|
"create_comment: could not find the selection text in the page to anchor the comment. " +
|
||||||
|
"Provide the EXACT contiguous text from a single paragraph/block (<=250 chars).",
|
||||||
|
);
|
||||||
}
|
}
|
||||||
if (matches >= 2) {
|
if (matches >= 2) {
|
||||||
throw new Error(
|
throw new Error(
|
||||||
@@ -2488,27 +2567,18 @@ export class DocmostClient {
|
|||||||
// null despite countAnchorMatches===1 (shouldn't happen), fall back to
|
// null despite countAnchorMatches===1 (shouldn't happen), fall back to
|
||||||
// the raw agent selection below rather than crash.
|
// the raw agent selection below rather than crash.
|
||||||
anchoredSelection = getAnchoredText(page.content, selection);
|
anchoredSelection = getAnchoredText(page.content, selection);
|
||||||
anchorNormalized = resolveAnchorSelection(
|
} else if (!canAnchorInDoc(page.content, selection)) {
|
||||||
page.content,
|
throw new Error(
|
||||||
selection,
|
"create_comment: could not find the selection text in the page to anchor the comment. " +
|
||||||
).normalized;
|
"Provide the EXACT contiguous text from a single paragraph/block (<=250 chars).",
|
||||||
} else {
|
);
|
||||||
const resolved = resolveAnchorSelection(page.content, selection);
|
|
||||||
if (!resolved.found) {
|
|
||||||
throw this.anchorNotFoundError(page.content, selection, false);
|
|
||||||
}
|
|
||||||
anchorNormalized = resolved.normalized;
|
|
||||||
}
|
}
|
||||||
} catch (e) {
|
} catch (e) {
|
||||||
// Rethrow our own "not found"/"ambiguous"/"spans multiple blocks" errors;
|
// Rethrow our own "not found"/"ambiguous" errors; swallow read/network
|
||||||
// swallow read/network errors so the live anchor step can still try (and
|
// errors so the live anchor step can still try (and enforce) anchoring.
|
||||||
// enforce) anchoring.
|
|
||||||
if (
|
if (
|
||||||
e instanceof Error &&
|
e instanceof Error &&
|
||||||
(e.message.startsWith("create_comment: could not find the selection") ||
|
(e.message.startsWith("create_comment: could not find the selection") ||
|
||||||
e.message.startsWith(
|
|
||||||
"create_comment: the selection spans multiple blocks",
|
|
||||||
) ||
|
|
||||||
e.message.startsWith(
|
e.message.startsWith(
|
||||||
"create_comment: the suggestion's selection is ambiguous",
|
"create_comment: the suggestion's selection is ambiguous",
|
||||||
))
|
))
|
||||||
@@ -2580,10 +2650,6 @@ export class DocmostClient {
|
|||||||
// Set inside the transform when a suggestion's live anchor is ambiguous
|
// Set inside the transform when a suggestion's live anchor is ambiguous
|
||||||
// (>=2 occurrences), so the rollback path can surface the right error.
|
// (>=2 occurrences), so the rollback path can surface the right error.
|
||||||
let ambiguousInLiveDoc = false;
|
let ambiguousInLiveDoc = false;
|
||||||
// Captured inside the transform on a not-found abort, so the rollback path
|
|
||||||
// can surface the closest-block / spans-multiple-blocks hint built from the
|
|
||||||
// LIVE document (the pre-check page is not in scope there).
|
|
||||||
let liveNotFoundError: Error | null = null;
|
|
||||||
try {
|
try {
|
||||||
const collabToken = await this.getCollabTokenWithReauth();
|
const collabToken = await this.getCollabTokenWithReauth();
|
||||||
// Open the collab doc by the canonical UUID, never the slugId (#260). The
|
// Open the collab doc by the canonical UUID, never the slugId (#260). The
|
||||||
@@ -2611,13 +2677,6 @@ export class DocmostClient {
|
|||||||
const liveCount = countAnchorMatches(doc, selection as string);
|
const liveCount = countAnchorMatches(doc, selection as string);
|
||||||
if (liveCount !== 1) {
|
if (liveCount !== 1) {
|
||||||
ambiguousInLiveDoc = liveCount >= 2;
|
ambiguousInLiveDoc = liveCount >= 2;
|
||||||
if (liveCount === 0) {
|
|
||||||
liveNotFoundError = this.anchorNotFoundError(
|
|
||||||
doc,
|
|
||||||
selection as string,
|
|
||||||
true,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -2627,11 +2686,6 @@ export class DocmostClient {
|
|||||||
}
|
}
|
||||||
// Selection text not found in the LIVE document: abort the write. The
|
// Selection text not found in the LIVE document: abort the write. The
|
||||||
// rollback + throw below turns this into a hard error.
|
// rollback + throw below turns this into a hard error.
|
||||||
liveNotFoundError = this.anchorNotFoundError(
|
|
||||||
doc,
|
|
||||||
selection as string,
|
|
||||||
true,
|
|
||||||
);
|
|
||||||
return null;
|
return null;
|
||||||
},
|
},
|
||||||
);
|
);
|
||||||
@@ -2648,28 +2702,13 @@ export class DocmostClient {
|
|||||||
// suggestion, was ambiguous) in the live document. Roll back the comment
|
// suggestion, was ambiguous) in the live document. Roll back the comment
|
||||||
// and surface a hard error.
|
// and surface a hard error.
|
||||||
await this.safeDeleteComment(newCommentId);
|
await this.safeDeleteComment(newCommentId);
|
||||||
if (ambiguousInLiveDoc) {
|
throw new Error(
|
||||||
throw new Error(
|
ambiguousInLiveDoc
|
||||||
"create_comment: the suggestion's selection is ambiguous in the live document (multiple occurrences); the comment was rolled back. Expand the selection with surrounding context so it is unique.",
|
? "create_comment: the suggestion's selection is ambiguous in the live document (multiple occurrences); the comment was rolled back. Expand the selection with surrounding context so it is unique."
|
||||||
);
|
: "create_comment: failed to anchor the comment (selection not found in the live document); the comment was rolled back",
|
||||||
}
|
|
||||||
throw (
|
|
||||||
liveNotFoundError ??
|
|
||||||
new Error(
|
|
||||||
"create_comment: failed to anchor the comment (selection not found in the live document); the comment was rolled back",
|
|
||||||
)
|
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
// Soft warning (like edit_page_text): the selection only matched after
|
|
||||||
// stripping markdown, so the caller likely quoted a styled fragment.
|
|
||||||
if (anchorNormalized) {
|
|
||||||
result.warning =
|
|
||||||
"The selection matched only after stripping markdown syntax; the comment " +
|
|
||||||
"was anchored on the document's plain text. Copy the selection verbatim " +
|
|
||||||
"from get_page / search_in_page output to avoid this.";
|
|
||||||
}
|
|
||||||
|
|
||||||
result.anchored = true;
|
result.anchored = true;
|
||||||
return result;
|
return result;
|
||||||
}
|
}
|
||||||
@@ -3393,453 +3432,6 @@ export class DocmostClient {
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
// --- draw.io diagrams (issue #423) ---
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Upload a ready-made byte buffer as a page attachment via the same
|
|
||||||
* multipart /files/upload endpoint uploadImage uses. Split out as its own
|
|
||||||
* (overridable) seam so drawio_create/update can upload the generated
|
|
||||||
* `.drawio.svg` without going through the URL-fetch path, and so tests can
|
|
||||||
* stub the network. Mirrors uploadImage's fresh-FormData + one-shot 401/403
|
|
||||||
* re-auth handling (a FormData body is single-use, so it must be rebuilt per
|
|
||||||
* attempt).
|
|
||||||
*/
|
|
||||||
protected async uploadAttachmentBuffer(
|
|
||||||
pageId: string,
|
|
||||||
buffer: Buffer,
|
|
||||||
fileName: string,
|
|
||||||
mime: string,
|
|
||||||
): Promise<{ id: string; fileName: string; fileSize: number }> {
|
|
||||||
await this.ensureAuthenticated();
|
|
||||||
const buildForm = () => {
|
|
||||||
const form = new FormData();
|
|
||||||
form.append("pageId", pageId);
|
|
||||||
form.append("file", buffer, { filename: fileName, contentType: mime });
|
|
||||||
return form;
|
|
||||||
};
|
|
||||||
const uploadUrl = `${this.apiUrl}/files/upload`;
|
|
||||||
let response;
|
|
||||||
try {
|
|
||||||
const form = buildForm();
|
|
||||||
response = await axios.post(uploadUrl, form, {
|
|
||||||
headers: {
|
|
||||||
...form.getHeaders(),
|
|
||||||
Authorization: this.client.defaults.headers.common["Authorization"],
|
|
||||||
},
|
|
||||||
timeout: 60000,
|
|
||||||
});
|
|
||||||
} catch (error) {
|
|
||||||
if (
|
|
||||||
axios.isAxiosError(error) &&
|
|
||||||
(error.response?.status === 401 || error.response?.status === 403)
|
|
||||||
) {
|
|
||||||
await this.login();
|
|
||||||
const form2 = buildForm();
|
|
||||||
response = await axios.post(uploadUrl, form2, {
|
|
||||||
headers: {
|
|
||||||
...form2.getHeaders(),
|
|
||||||
Authorization: this.client.defaults.headers.common["Authorization"],
|
|
||||||
},
|
|
||||||
timeout: 60000,
|
|
||||||
});
|
|
||||||
} else if (axios.isAxiosError(error)) {
|
|
||||||
if (process.env.DEBUG) {
|
|
||||||
console.error(
|
|
||||||
"Attachment upload failed; response body:",
|
|
||||||
JSON.stringify(error.response?.data),
|
|
||||||
);
|
|
||||||
}
|
|
||||||
throw new Error(
|
|
||||||
`Attachment upload failed: ${error.response?.status} ${error.response?.statusText}`,
|
|
||||||
);
|
|
||||||
} else {
|
|
||||||
throw error;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
const att = response.data?.data ?? response.data;
|
|
||||||
if (!att?.id || !att?.fileName) {
|
|
||||||
throw new Error(
|
|
||||||
"Unexpected /files/upload response: " + JSON.stringify(response.data),
|
|
||||||
);
|
|
||||||
}
|
|
||||||
return {
|
|
||||||
id: att.id,
|
|
||||||
fileName: att.fileName,
|
|
||||||
fileSize: att.fileSize ?? buffer.length,
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Fetch a stored `.drawio.svg` attachment as text. Overridable seam over
|
|
||||||
* fetchInternalFile (the authed loopback fetch, which also rejects any
|
|
||||||
* traversal/SSRF src) so drawio_get/update can read the current diagram and
|
|
||||||
* tests can stub the bytes.
|
|
||||||
*/
|
|
||||||
protected async fetchAttachmentText(src: string): Promise<string> {
|
|
||||||
const { buffer } = await this.fetchInternalFile(src);
|
|
||||||
return buffer.toString("utf-8");
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Resolve a drawio node on a page by `attrs.id` or `#<index>` and return the
|
|
||||||
* node plus its ref. Throws a clear error if the ref does not resolve to a
|
|
||||||
* drawio node.
|
|
||||||
*/
|
|
||||||
private async resolveDrawioNode(
|
|
||||||
pageId: string,
|
|
||||||
node: string,
|
|
||||||
): Promise<{ node: any; ref: string }> {
|
|
||||||
const data = await this.getPageRaw(pageId);
|
|
||||||
const hit = getNodeByRef(
|
|
||||||
data.content ?? { type: "doc", content: [] },
|
|
||||||
node,
|
|
||||||
);
|
|
||||||
if (!hit) {
|
|
||||||
throw new Error(
|
|
||||||
`drawio: no node found for "${node}" on page ${pageId} (use the drawio node's attrs.id or "#<index>" from get_outline)`,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
if (hit.type !== "drawio") {
|
|
||||||
throw new Error(
|
|
||||||
`drawio: node "${node}" on page ${pageId} is a ${hit.type}, not a drawio diagram`,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
return { node: hit.node, ref: node };
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Read a drawio diagram as mxGraph XML (default) or as the raw `.drawio.svg`.
|
|
||||||
* Runs the decode chain (base64/entity content= → drawio file → nested XML or
|
|
||||||
* pako-inflated compressed <diagram>). The returned `hash` is the
|
|
||||||
* optimistic-lock key for drawio_update.
|
|
||||||
*/
|
|
||||||
async drawioGet(
|
|
||||||
pageId: string,
|
|
||||||
node: string,
|
|
||||||
format: "xml" | "svg" = "xml",
|
|
||||||
): Promise<{
|
|
||||||
pageId: string;
|
|
||||||
nodeId: string;
|
|
||||||
format: "xml" | "svg";
|
|
||||||
content: string;
|
|
||||||
meta: {
|
|
||||||
attachmentId: string | null;
|
|
||||||
title: string | null;
|
|
||||||
width: number | null;
|
|
||||||
height: number | null;
|
|
||||||
cellCount: number;
|
|
||||||
hash: string;
|
|
||||||
};
|
|
||||||
}> {
|
|
||||||
await this.ensureAuthenticated();
|
|
||||||
const { node: drawio } = await this.resolveDrawioNode(pageId, node);
|
|
||||||
const attrs = drawio.attrs || {};
|
|
||||||
const src = attrs.src;
|
|
||||||
if (!src) {
|
|
||||||
throw new Error(
|
|
||||||
`drawio: node "${node}" on page ${pageId} has no src to read`,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
const svg = await this.fetchAttachmentText(src);
|
|
||||||
const modelXml = decodeDrawioSvg(svg);
|
|
||||||
const meta = {
|
|
||||||
attachmentId: attrs.attachmentId ?? null,
|
|
||||||
title: attrs.title ?? null,
|
|
||||||
width: attrs.width != null ? Number(attrs.width) : null,
|
|
||||||
height: attrs.height != null ? Number(attrs.height) : null,
|
|
||||||
cellCount: countUserCells(modelXml),
|
|
||||||
hash: mxHash(modelXml),
|
|
||||||
};
|
|
||||||
return {
|
|
||||||
pageId,
|
|
||||||
nodeId: attrs.id ?? node,
|
|
||||||
format,
|
|
||||||
content: format === "svg" ? svg : normalizeXml(modelXml),
|
|
||||||
meta,
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Create a drawio diagram from mxGraph XML: lint → schematic SVG preview
|
|
||||||
* (pure TS) → build the `.drawio.svg` (createDrawioSvg contract) → create the
|
|
||||||
* attachment → insert a `drawio` node before/after an anchor or appended.
|
|
||||||
* `xml` is a bare `<mxGraphModel>` or a list of `<mxCell>` (the server wraps
|
|
||||||
* it and adds the id=0/id=1 sentinels).
|
|
||||||
*/
|
|
||||||
async drawioCreate(
|
|
||||||
pageId: string,
|
|
||||||
where: {
|
|
||||||
position: "before" | "after" | "append";
|
|
||||||
anchorNodeId?: string;
|
|
||||||
anchorText?: string;
|
|
||||||
},
|
|
||||||
xml: string,
|
|
||||||
title?: string,
|
|
||||||
): Promise<{
|
|
||||||
success: boolean;
|
|
||||||
nodeId: string;
|
|
||||||
attachmentId: string;
|
|
||||||
warnings: string[];
|
|
||||||
verify?: any;
|
|
||||||
}> {
|
|
||||||
await this.ensureAuthenticated();
|
|
||||||
if (
|
|
||||||
!where ||
|
|
||||||
(where.position !== "before" &&
|
|
||||||
where.position !== "after" &&
|
|
||||||
where.position !== "append")
|
|
||||||
) {
|
|
||||||
throw new Error(
|
|
||||||
'drawio_create: `where.position` must be one of "before", "after", "append"',
|
|
||||||
);
|
|
||||||
}
|
|
||||||
if (where.position === "before" || where.position === "after") {
|
|
||||||
const hasId =
|
|
||||||
typeof where.anchorNodeId === "string" && where.anchorNodeId.length > 0;
|
|
||||||
const hasText =
|
|
||||||
typeof where.anchorText === "string" && where.anchorText.length > 0;
|
|
||||||
if (hasId === hasText) {
|
|
||||||
throw new Error(
|
|
||||||
`drawio_create: position "${where.position}" requires exactly one of anchorNodeId or anchorText`,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Pre-write pipeline (throws a structured DrawioLintError on any violation).
|
|
||||||
const prepared = prepareModel(xml);
|
|
||||||
const inner = renderDiagramShapes(prepared.cells, prepared.bbox);
|
|
||||||
const diagramTitle = title || "Page-1";
|
|
||||||
const svg = buildDrawioSvg(prepared.modelXml, inner, prepared.bbox, diagramTitle);
|
|
||||||
|
|
||||||
const att = await this.uploadAttachmentBuffer(
|
|
||||||
pageId,
|
|
||||||
Buffer.from(svg, "utf-8"),
|
|
||||||
"diagram.drawio.svg",
|
|
||||||
"image/svg+xml",
|
|
||||||
);
|
|
||||||
|
|
||||||
// NOTE: no `id` attribute is set here. The vendored `drawio` node schema
|
|
||||||
// (diagramAttributes) declares no `id`, so any block id would be silently
|
|
||||||
// dropped by PMNode.fromJSON on save and the returned handle would fail to
|
|
||||||
// resolve. The addressable handle is the node's "#<index>" (like image/table
|
|
||||||
// nodes), computed after the insert below.
|
|
||||||
const drawioNode: any = {
|
|
||||||
type: "drawio",
|
|
||||||
attrs: {
|
|
||||||
src: `/api/files/${att.id}/${att.fileName}`,
|
|
||||||
attachmentId: att.id,
|
|
||||||
width: prepared.bbox.width,
|
|
||||||
height: prepared.bbox.height,
|
|
||||||
align: "center",
|
|
||||||
},
|
|
||||||
};
|
|
||||||
if (title) drawioNode.attrs.title = title;
|
|
||||||
// Reuse the existing URL trust boundary (rejects unsafe src schemes).
|
|
||||||
this.validateDocUrls(drawioNode);
|
|
||||||
|
|
||||||
const collabToken = await this.getCollabTokenWithReauth();
|
|
||||||
const pageUuid = await this.resolvePageId(pageId);
|
|
||||||
|
|
||||||
let inserted = false;
|
|
||||||
let insertedIndex = -1;
|
|
||||||
const mutation = await this.mutatePage(
|
|
||||||
pageUuid,
|
|
||||||
collabToken,
|
|
||||||
this.apiUrl,
|
|
||||||
(liveDoc) => {
|
|
||||||
inserted = false;
|
|
||||||
insertedIndex = -1;
|
|
||||||
const { doc: nd, inserted: ins } = insertNodeRelative(
|
|
||||||
liveDoc,
|
|
||||||
drawioNode,
|
|
||||||
where,
|
|
||||||
);
|
|
||||||
inserted = ins;
|
|
||||||
if (!inserted) return null; // anchor not found -> skip the write
|
|
||||||
// Locate the freshly-inserted node to derive its "#<index>" handle. The
|
|
||||||
// just-uploaded attachmentId is unique, so it identifies our node.
|
|
||||||
if (Array.isArray(nd.content)) {
|
|
||||||
insertedIndex = nd.content.findIndex(
|
|
||||||
(b: any) =>
|
|
||||||
b &&
|
|
||||||
b.type === "drawio" &&
|
|
||||||
b.attrs &&
|
|
||||||
b.attrs.attachmentId === att.id,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
return nd;
|
|
||||||
},
|
|
||||||
);
|
|
||||||
|
|
||||||
if (!inserted) {
|
|
||||||
const anchorDesc = where.anchorNodeId
|
|
||||||
? `anchorNodeId "${where.anchorNodeId}"`
|
|
||||||
: `anchorText "${where.anchorText}"`;
|
|
||||||
throw new Error(
|
|
||||||
`drawio_create: anchor not found (${anchorDesc}) on page ${pageId}. The diagram attachment ${att.id} is now an unreferenced orphan.`,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
if (insertedIndex < 0) {
|
|
||||||
// The node was inserted nested (e.g. inside a callout/table cell via an
|
|
||||||
// anchor), where "#<index>" — which addresses only top-level blocks —
|
|
||||||
// cannot reference it. drawio nodes carry no persisted id, so there is no
|
|
||||||
// stable handle for a nested diagram.
|
|
||||||
throw new Error(
|
|
||||||
`drawio_create: the diagram was inserted on page ${pageId} but not as a ` +
|
|
||||||
`top-level block, so it has no addressable "#<index>" handle. Anchor ` +
|
|
||||||
`on a top-level block (or append) so the diagram can be re-read.`,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
// The returned handle is POSITIONAL ("#<index>"): valid for the immediate
|
|
||||||
// create -> get/update flow, but re-resolve via get_outline if the document
|
|
||||||
// structure changes (blocks added/removed before it shift the index).
|
|
||||||
const nodeId = `#${insertedIndex}`;
|
|
||||||
|
|
||||||
return {
|
|
||||||
success: true,
|
|
||||||
nodeId,
|
|
||||||
attachmentId: att.id,
|
|
||||||
warnings: prepared.warnings,
|
|
||||||
verify: mutation.verify,
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Full-replacement update of a drawio diagram. `baseHash` is MANDATORY: it is
|
|
||||||
* compared against the hash of the diagram's CURRENT XML (from drawio_get);
|
|
||||||
* any mismatch means a human or another agent edited the diagram after the
|
|
||||||
* read, so the write is refused with a conflict error. On success the new
|
|
||||||
* `.drawio.svg` is uploaded as a FRESH attachment (in-place byte overwrite is
|
|
||||||
* avoided — some Docmost versions corrupt an attachment on overwrite, exactly
|
|
||||||
* as replaceImage documents) and the node is repointed with new dimensions.
|
|
||||||
*/
|
|
||||||
async drawioUpdate(
|
|
||||||
pageId: string,
|
|
||||||
node: string,
|
|
||||||
xml: string,
|
|
||||||
baseHash: string,
|
|
||||||
): Promise<{
|
|
||||||
success: boolean;
|
|
||||||
nodeId: string;
|
|
||||||
attachmentId: string;
|
|
||||||
warnings: string[];
|
|
||||||
verify?: any;
|
|
||||||
}> {
|
|
||||||
await this.ensureAuthenticated();
|
|
||||||
if (typeof baseHash !== "string" || baseHash.length === 0) {
|
|
||||||
throw new Error(
|
|
||||||
"drawio_update: baseHash is mandatory — read the diagram with drawio_get first and pass back its meta.hash",
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
// Resolve the node and read the CURRENT diagram to enforce the optimistic
|
|
||||||
// lock before doing any write or upload.
|
|
||||||
const { node: drawio, ref } = await this.resolveDrawioNode(pageId, node);
|
|
||||||
const oldAttrs = drawio.attrs || {};
|
|
||||||
const oldSrc = oldAttrs.src;
|
|
||||||
// The returned handle is the caller-supplied reference. drawio nodes carry
|
|
||||||
// no persisted id, so `ref` (an "#<index>" or a rare legacy attrs.id) is the
|
|
||||||
// honest identifier to hand back.
|
|
||||||
const nodeId = oldAttrs.id ?? ref;
|
|
||||||
if (!oldSrc) {
|
|
||||||
throw new Error(
|
|
||||||
`drawio_update: node "${node}" on page ${pageId} has no src to compare against`,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
const currentSvg = await this.fetchAttachmentText(oldSrc);
|
|
||||||
const currentHash = mxHash(decodeDrawioSvg(currentSvg));
|
|
||||||
if (currentHash !== baseHash) {
|
|
||||||
throw new Error(
|
|
||||||
`drawio_update: conflict — the diagram changed since it was read ` +
|
|
||||||
`(baseHash ${baseHash} != current ${currentHash}). Re-read it with drawio_get and retry.`,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
// Pipeline for the new content (throws a structured DrawioLintError).
|
|
||||||
const prepared = prepareModel(xml);
|
|
||||||
const inner = renderDiagramShapes(prepared.cells, prepared.bbox);
|
|
||||||
const diagramTitle = oldAttrs.title || "Page-1";
|
|
||||||
const svg = buildDrawioSvg(prepared.modelXml, inner, prepared.bbox, diagramTitle);
|
|
||||||
|
|
||||||
const att = await this.uploadAttachmentBuffer(
|
|
||||||
pageId,
|
|
||||||
Buffer.from(svg, "utf-8"),
|
|
||||||
"diagram.drawio.svg",
|
|
||||||
"image/svg+xml",
|
|
||||||
);
|
|
||||||
const newSrc = `/api/files/${att.id}/${att.fileName}`;
|
|
||||||
|
|
||||||
const collabToken = await this.getCollabTokenWithReauth();
|
|
||||||
const pageUuid = await this.resolvePageId(pageId);
|
|
||||||
|
|
||||||
let repointed = 0;
|
|
||||||
const repoint = (n: any) => {
|
|
||||||
n.attrs = {
|
|
||||||
...n.attrs,
|
|
||||||
src: newSrc,
|
|
||||||
attachmentId: att.id,
|
|
||||||
width: prepared.bbox.width,
|
|
||||||
height: prepared.bbox.height,
|
|
||||||
};
|
|
||||||
repointed++;
|
|
||||||
};
|
|
||||||
|
|
||||||
const mutation = await this.mutatePage(
|
|
||||||
pageUuid,
|
|
||||||
collabToken,
|
|
||||||
this.apiUrl,
|
|
||||||
(liveDoc) => {
|
|
||||||
repointed = 0;
|
|
||||||
const doc =
|
|
||||||
liveDoc && liveDoc.type === "doc"
|
|
||||||
? liveDoc
|
|
||||||
: { type: "doc", content: [] };
|
|
||||||
if (!Array.isArray(doc.content)) doc.content = [];
|
|
||||||
// Repoint ONLY the resolved node — never every node that happens to
|
|
||||||
// share this attachmentId (a copied diagram is two nodes with one
|
|
||||||
// attachmentId; keying on it would clobber both). Re-resolve the same
|
|
||||||
// handle against the live doc and walk to its exact position.
|
|
||||||
const hit = getNodeByRef(doc, ref);
|
|
||||||
if (!hit || hit.type !== "drawio") return null; // vanished/changed -> skip
|
|
||||||
let target: any = doc;
|
|
||||||
for (const idx of hit.path) {
|
|
||||||
if (!target || !Array.isArray(target.content)) {
|
|
||||||
target = null;
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
target = target.content[idx];
|
|
||||||
}
|
|
||||||
if (!target || target.type !== "drawio") return null;
|
|
||||||
repoint(target);
|
|
||||||
if (repointed === 0) return null; // node vanished concurrently -> skip
|
|
||||||
return doc;
|
|
||||||
},
|
|
||||||
);
|
|
||||||
|
|
||||||
if (repointed === 0) {
|
|
||||||
return {
|
|
||||||
success: true,
|
|
||||||
nodeId,
|
|
||||||
attachmentId: att.id,
|
|
||||||
warnings: [
|
|
||||||
...prepared.warnings,
|
|
||||||
"target drawio node was removed concurrently; uploaded attachment is unreferenced",
|
|
||||||
],
|
|
||||||
verify: mutation.verify,
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
return {
|
|
||||||
success: true,
|
|
||||||
nodeId,
|
|
||||||
attachmentId: att.id,
|
|
||||||
warnings: prepared.warnings,
|
|
||||||
verify: mutation.verify,
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
// --- Page history / diff / transform ---
|
// --- Page history / diff / transform ---
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ import { readFileSync } from "fs";
|
|||||||
import { fileURLToPath } from "url";
|
import { fileURLToPath } from "url";
|
||||||
import { dirname, join } from "path";
|
import { dirname, join } from "path";
|
||||||
import { DocmostClient, DocmostMcpConfig } from "./client.js";
|
import { DocmostClient, DocmostMcpConfig } from "./client.js";
|
||||||
import { parseNodeArg } from "./lib/parse-node-arg.js";
|
import { parseNodeArg } from "@docmost/prosemirror-markdown";
|
||||||
import { SHARED_TOOL_SPECS, SharedToolSpec } from "./tool-specs.js";
|
import { SHARED_TOOL_SPECS, SharedToolSpec } from "./tool-specs.js";
|
||||||
|
|
||||||
// Re-export the client and its config type so embedding hosts (e.g. the gitmost
|
// Re-export the client and its config type so embedding hosts (e.g. the gitmost
|
||||||
@@ -13,11 +13,6 @@ import { SHARED_TOOL_SPECS, SharedToolSpec } from "./tool-specs.js";
|
|||||||
export { DocmostClient } from "./client.js";
|
export { DocmostClient } from "./client.js";
|
||||||
export type { DocmostMcpConfig } from "./client.js";
|
export type { DocmostMcpConfig } from "./client.js";
|
||||||
|
|
||||||
// Teardown for the live per-page CollabSession cache (issue #400). An embedding
|
|
||||||
// HTTP host (the gitmost NestJS server) should call this from its own shutdown
|
|
||||||
// hook so no cached collab provider outlives the process.
|
|
||||||
export { destroyAllSessions } from "./lib/collab-session.js";
|
|
||||||
|
|
||||||
// Re-export the zod-agnostic shared tool-spec registry so the in-app AI-SDK
|
// Re-export the zod-agnostic shared tool-spec registry so the in-app AI-SDK
|
||||||
// service can read it off the loaded module (it cannot import the ESM package's
|
// service can read it off the loaded module (it cannot import the ESM package's
|
||||||
// internals directly; it goes through loadDocmostMcp()).
|
// internals directly; it goes through loadDocmostMcp()).
|
||||||
@@ -52,7 +47,7 @@ const VERSION = packageJson.version;
|
|||||||
export const SERVER_INSTRUCTIONS =
|
export const SERVER_INSTRUCTIONS =
|
||||||
"Docmost editing guide — choose the tool by intent.\n" +
|
"Docmost editing guide — choose the tool by intent.\n" +
|
||||||
"READ: find a page -> search (workspace-wide full-text); list -> list_pages / list_spaces. Locate blocks and their ids CHEAPLY -> get_outline (compact top-level map; start here, not get_page_json). One block's subtree -> get_node (by attrs.id, or \"#<index>\" for tables, which carry no id). Find every occurrence of a string/regex ON a page (and where each is) -> search_in_page, NOT block-by-block get_node — it returns each hit's node ref + block index + context for a targeted comment. Whole page -> get_page (Markdown, lossy; inline <span data-comment-id> tags are comment anchors — markup, not text) or get_page_json (lossless ProseMirror with block ids). Hand a huge page (with images) to an external consumer without pulling it through the model context -> stash_page (returns a short-lived anonymous URL).\n" +
|
"READ: find a page -> search (workspace-wide full-text); list -> list_pages / list_spaces. Locate blocks and their ids CHEAPLY -> get_outline (compact top-level map; start here, not get_page_json). One block's subtree -> get_node (by attrs.id, or \"#<index>\" for tables, which carry no id). Find every occurrence of a string/regex ON a page (and where each is) -> search_in_page, NOT block-by-block get_node — it returns each hit's node ref + block index + context for a targeted comment. Whole page -> get_page (Markdown, lossy; inline <span data-comment-id> tags are comment anchors — markup, not text) or get_page_json (lossless ProseMirror with block ids). Hand a huge page (with images) to an external consumer without pulling it through the model context -> stash_page (returns a short-lived anonymous URL).\n" +
|
||||||
"EDIT: fix wording/typos/numbers -> edit_page_text (find/replace inside blocks, no node id needed). Change ONE block (paragraph/heading/callout/etc.) structurally -> patch_node (by attrs.id from get_outline). Add a block -> insert_node (before/after a block by attrs.id or by anchor text, or append). Remove a block -> delete_node (by attrs.id). Tables -> table_get / table_update_cell / table_insert_row / table_delete_row (address by \"#<index>\" from get_outline; table nodes have no attrs.id). Images -> insert_image (add from a web URL) / replace_image (swap an existing image). Draw.io diagrams -> drawio_create (create from mxGraph XML and insert), drawio_get (read a diagram as mxGraph XML + a hash), drawio_update (replace a diagram; pass the hash from drawio_get as baseHash for optimistic locking). Footnotes -> insert_footnote. Bulk/structural rewrite -> update_page_json (full ProseMirror replace; prefer the granular tools above to avoid resending the whole ~100KB+ document). Complex/scripted rewrite (multiple coordinated edits, renumbering) -> docmost_transform: write a JS `(doc, ctx) => doc` transform, preview the diff with dryRun (default), then apply with dryRun:false; ctx.helpers includes commentsToFootnotes for turning inline comments into numbered footnotes.\n" +
|
"EDIT: fix wording/typos/numbers -> edit_page_text (find/replace inside blocks, no node id needed). Change ONE block (paragraph/heading/callout/etc.) structurally -> patch_node (by attrs.id from get_outline). Add a block -> insert_node (before/after a block by attrs.id or by anchor text, or append). Remove a block -> delete_node (by attrs.id). Tables -> table_get / table_update_cell / table_insert_row / table_delete_row (address by \"#<index>\" from get_outline; table nodes have no attrs.id). Images -> insert_image (add from a web URL) / replace_image (swap an existing image). Footnotes -> insert_footnote. Bulk/structural rewrite -> update_page_json (full ProseMirror replace; prefer the granular tools above to avoid resending the whole ~100KB+ document). Complex/scripted rewrite (multiple coordinated edits, renumbering) -> docmost_transform: write a JS `(doc, ctx) => doc` transform, preview the diff with dryRun (default), then apply with dryRun:false; ctx.helpers includes commentsToFootnotes for turning inline comments into numbered footnotes.\n" +
|
||||||
"PAGES: new -> create_page (Markdown). Rename (title only) -> rename_page. Move -> move_page. Delete -> delete_page (SOFT delete — the page goes to trash and is restorable; nothing is permanent). Copy/replace a page's whole content from another page (server-side, no document through the model) -> copy_page_content. Sharing -> share_page / unshare_page / list_shares; share_page makes the page PUBLICLY accessible — do it only when explicitly asked.\n" +
|
"PAGES: new -> create_page (Markdown). Rename (title only) -> rename_page. Move -> move_page. Delete -> delete_page (SOFT delete — the page goes to trash and is restorable; nothing is permanent). Copy/replace a page's whole content from another page (server-side, no document through the model) -> copy_page_content. Sharing -> share_page / unshare_page / list_shares; share_page makes the page PUBLICLY accessible — do it only when explicitly asked.\n" +
|
||||||
"COMMENTS: create_comment is always inline and requires an EXACT selection — contiguous text from a single block, <=250 chars (fails rather than leaving an unanchored comment); reply to a thread via parentCommentId. Propose a concrete text fix for one-click human approval -> create_comment with suggestedText (the exact plain-text replacement for the selection; the selection must then be UNIQUE in the page — extend it with context if needed); prefer this over editing directly when the change is subjective or needs the author's sign-off. Manage -> list_comments, update_comment, resolve_comment (resolve/reopen, reversible — prefer over delete to close), delete_comment, check_new_comments.\n" +
|
"COMMENTS: create_comment is always inline and requires an EXACT selection — contiguous text from a single block, <=250 chars (fails rather than leaving an unanchored comment); reply to a thread via parentCommentId. Propose a concrete text fix for one-click human approval -> create_comment with suggestedText (the exact plain-text replacement for the selection; the selection must then be UNIQUE in the page — extend it with context if needed); prefer this over editing directly when the change is subjective or needs the author's sign-off. Manage -> list_comments, update_comment, resolve_comment (resolve/reopen, reversible — prefer over delete to close), delete_comment, check_new_comments.\n" +
|
||||||
"HISTORY: review what changed -> diff_page_versions (a historyId vs current, or two versions). List saved versions -> list_page_history. Undo a bad edit -> restore_page_version (writes a past version back as current; itself revertible). Lossless markdown round-trip (download, edit, re-upload, incl. comment anchors) -> export_page_markdown / import_page_markdown.";
|
"HISTORY: review what changed -> diff_page_versions (a historyId vs current, or two versions). List saved versions -> list_page_history. Undo a bad edit -> restore_page_version (writes a past version back as current; itself revertible). Lossless markdown round-trip (download, edit, re-upload, incl. comment anchors) -> export_page_markdown / import_page_markdown.";
|
||||||
@@ -465,38 +460,6 @@ registerShared(
|
|||||||
},
|
},
|
||||||
);
|
);
|
||||||
|
|
||||||
// Tool: drawio_get — read a draw.io diagram as mxGraph XML (or the raw SVG).
|
|
||||||
registerShared(
|
|
||||||
SHARED_TOOL_SPECS.drawioGet,
|
|
||||||
async ({ pageId, node, format }) => {
|
|
||||||
const result = await docmostClient.drawioGet(pageId, node, format ?? "xml");
|
|
||||||
return jsonContent(result);
|
|
||||||
},
|
|
||||||
);
|
|
||||||
|
|
||||||
// Tool: drawio_create — lint mxGraph XML, build the .drawio.svg, insert a node.
|
|
||||||
registerShared(
|
|
||||||
SHARED_TOOL_SPECS.drawioCreate,
|
|
||||||
async ({ pageId, xml, position, anchorNodeId, anchorText, title }) => {
|
|
||||||
const result = await docmostClient.drawioCreate(
|
|
||||||
pageId,
|
|
||||||
{ position, anchorNodeId, anchorText },
|
|
||||||
xml,
|
|
||||||
title,
|
|
||||||
);
|
|
||||||
return jsonContent(result);
|
|
||||||
},
|
|
||||||
);
|
|
||||||
|
|
||||||
// Tool: drawio_update — optimistic-locked full replacement of a diagram.
|
|
||||||
registerShared(
|
|
||||||
SHARED_TOOL_SPECS.drawioUpdate,
|
|
||||||
async ({ pageId, node, xml, baseHash }) => {
|
|
||||||
const result = await docmostClient.drawioUpdate(pageId, node, xml, baseHash);
|
|
||||||
return jsonContent(result);
|
|
||||||
},
|
|
||||||
);
|
|
||||||
|
|
||||||
// Tool: share_page
|
// Tool: share_page
|
||||||
// Schema + description now live in the shared registry (#294). The execute body
|
// Schema + description now live in the shared registry (#294). The execute body
|
||||||
// keeps this transport's own `searchIndexing ?? true` default.
|
// keeps this transport's own `searchIndexing ?? true` default.
|
||||||
|
|||||||
@@ -1,668 +0,0 @@
|
|||||||
import { HocuspocusProvider } from "@hocuspocus/provider";
|
|
||||||
import { TiptapTransformer } from "@hocuspocus/transformer";
|
|
||||||
import * as Y from "yjs";
|
|
||||||
import WebSocket from "ws";
|
|
||||||
import {
|
|
||||||
buildCollabWsUrl,
|
|
||||||
applyDocToFragment,
|
|
||||||
MutationResult,
|
|
||||||
} from "./collaboration.js";
|
|
||||||
import { summarizeChange } from "./diff.js";
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Live per-page collaboration session cache (issue #400).
|
|
||||||
*
|
|
||||||
* The one-shot write path (collaboration.mutatePageContent /
|
|
||||||
* client.mutateLiveContentUnlocked) used to open a NEW HocuspocusProvider, run
|
|
||||||
* the full connect -> auth -> onLoadDocument -> initial-sync handshake, apply a
|
|
||||||
* single edit, wait for persistence, and then `provider.destroy()` — for EVERY
|
|
||||||
* content mutation. Disconnecting after every edit means that once the pause
|
|
||||||
* between calls exceeds the server's write debounce, the server does a full
|
|
||||||
* store -> unload -> reload per cell, causing 25s connect timeouts and
|
|
||||||
* event-loop lag under a burst of edits on one page.
|
|
||||||
*
|
|
||||||
* This module keeps ONE live provider + ydoc per (wsUrl, pageId, token) alive
|
|
||||||
* across a SERIES of edits. While the provider stays connected the server never
|
|
||||||
* enters store -> unload -> reload, its debounce coalesces N writes into 1-2
|
|
||||||
* stores, and the repeated auth/load/initial-sync disappears.
|
|
||||||
*
|
|
||||||
* The synchronous read -> transform -> write section and the per-edit
|
|
||||||
* persistence-ack logic are preserved VERBATIM from the one-shot machine — the
|
|
||||||
* only change is that they run on a persistent provider instead of a throwaway
|
|
||||||
* one. See CollabSession.mutate.
|
|
||||||
*/
|
|
||||||
|
|
||||||
/** Time we wait for the initial handshake/sync before giving up. */
|
|
||||||
const CONNECT_TIMEOUT_MS = 25000;
|
|
||||||
/** Time we wait for the server to acknowledge our write before giving up. */
|
|
||||||
const PERSIST_TIMEOUT_MS = 20000;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Tunables, read fresh from the environment on every acquire so tests (and a
|
|
||||||
* live rollback) can change them without reloading the module. Mirrors how
|
|
||||||
* http.ts parses MCP_SESSION_IDLE_MS.
|
|
||||||
* - MCP_COLLAB_SESSION_IDLE_MS: idle TTL, reset after every op. Default 60s.
|
|
||||||
* 0 (or negative) DISABLES the cache — every op opens its own provider and
|
|
||||||
* destroys it after the op, i.e. the exact legacy per-op-provider behavior
|
|
||||||
* (the rollback path).
|
|
||||||
* - MCP_COLLAB_SESSION_MAX_AGE_MS: hard lifetime checked at acquire; bounds
|
|
||||||
* the permission-staleness window. Default 10 min.
|
|
||||||
* - MCP_COLLAB_SESSION_MAX_ENTRIES: registry cap; the least-recently-used
|
|
||||||
* session is destroy-evicted when the cap is reached. Default 32.
|
|
||||||
*/
|
|
||||||
interface SessionConfig {
|
|
||||||
idleMs: number;
|
|
||||||
maxAgeMs: number;
|
|
||||||
maxEntries: number;
|
|
||||||
}
|
|
||||||
|
|
||||||
function parseEnvInt(value: string | undefined, fallback: number): number {
|
|
||||||
const parsed = parseInt(value ?? "", 10);
|
|
||||||
return Number.isFinite(parsed) ? parsed : fallback;
|
|
||||||
}
|
|
||||||
|
|
||||||
function readConfig(): SessionConfig {
|
|
||||||
// idleMs: allow 0 (disable). A malformed value falls back to the default.
|
|
||||||
const idleRaw = parseInt(process.env.MCP_COLLAB_SESSION_IDLE_MS ?? "", 10);
|
|
||||||
const idleMs = Number.isFinite(idleRaw) ? Math.max(0, idleRaw) : 60 * 1000;
|
|
||||||
const maxAgeMs = Math.max(
|
|
||||||
0,
|
|
||||||
parseEnvInt(process.env.MCP_COLLAB_SESSION_MAX_AGE_MS, 10 * 60 * 1000),
|
|
||||||
);
|
|
||||||
const maxEntriesRaw = parseEnvInt(
|
|
||||||
process.env.MCP_COLLAB_SESSION_MAX_ENTRIES,
|
|
||||||
32,
|
|
||||||
);
|
|
||||||
const maxEntries = maxEntriesRaw > 0 ? maxEntriesRaw : 32;
|
|
||||||
return { idleMs, maxAgeMs, maxEntries };
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* The subset of HocuspocusProvider this module depends on, so the provider can
|
|
||||||
* be replaced with a fake in unit tests (there is no server in the test env).
|
|
||||||
*/
|
|
||||||
export interface CollabProviderLike {
|
|
||||||
synced: boolean;
|
|
||||||
unsyncedChanges: number;
|
|
||||||
destroy(): void;
|
|
||||||
on(event: "unsyncedChanges", handler: (data: { number: number }) => void): void;
|
|
||||||
off(event: "unsyncedChanges", handler: (data: { number: number }) => void): void;
|
|
||||||
}
|
|
||||||
|
|
||||||
/** The configuration object passed to the provider factory. */
|
|
||||||
export interface CollabProviderConfig {
|
|
||||||
url: string;
|
|
||||||
name: string;
|
|
||||||
document: Y.Doc;
|
|
||||||
token: string;
|
|
||||||
WebSocketPolyfill: unknown;
|
|
||||||
onConnect: () => void;
|
|
||||||
onSynced: () => void;
|
|
||||||
onDisconnect: () => void;
|
|
||||||
onClose: () => void;
|
|
||||||
onAuthenticationFailed: () => void;
|
|
||||||
}
|
|
||||||
|
|
||||||
export type CollabProviderFactory = (
|
|
||||||
config: CollabProviderConfig,
|
|
||||||
) => CollabProviderLike;
|
|
||||||
|
|
||||||
const defaultProviderFactory: CollabProviderFactory = (config) =>
|
|
||||||
// @ts-ignore - WebSocketPolyfill is required for the Node.js environment.
|
|
||||||
new HocuspocusProvider(config) as unknown as CollabProviderLike;
|
|
||||||
|
|
||||||
let providerFactory: CollabProviderFactory = defaultProviderFactory;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* TEST SEAM: swap the provider factory (pass null to restore the real one).
|
|
||||||
* Not part of the public API — used only by the unit tests, which cannot reach
|
|
||||||
* a real collaboration server.
|
|
||||||
*/
|
|
||||||
export function __setCollabProviderFactory(
|
|
||||||
factory: CollabProviderFactory | null,
|
|
||||||
): void {
|
|
||||||
providerFactory = factory ?? defaultProviderFactory;
|
|
||||||
}
|
|
||||||
|
|
||||||
/** Optional per-acquire hooks (metrics), passed through from the call site. */
|
|
||||||
export interface AcquireOptions {
|
|
||||||
/** Invoked when the initial connect handshake times out (CONNECT_TIMEOUT_MS). */
|
|
||||||
onConnectTimeout?: () => void;
|
|
||||||
}
|
|
||||||
|
|
||||||
type SessionState = "connecting" | "ready" | "dead";
|
|
||||||
|
|
||||||
/**
|
|
||||||
* One live provider + ydoc for a single (wsUrl, pageId, token) triple.
|
|
||||||
*
|
|
||||||
* Lifecycle: connecting -> ready -> dead. A session becomes `dead` on the first
|
|
||||||
* disconnect/close/auth-failure at ANY time, on an idle/eviction/max-age
|
|
||||||
* teardown, or on an explicit destroy(); death is terminal and removes the
|
|
||||||
* session from the registry so the next acquire opens a fresh one. We never use
|
|
||||||
* the provider's auto-reconnect — destroying on the first disconnect closes the
|
|
||||||
* "reconnect drove unsyncedChanges to 0 without retransmitting our write" class
|
|
||||||
* of false success.
|
|
||||||
*/
|
|
||||||
export class CollabSession {
|
|
||||||
readonly key: string;
|
|
||||||
readonly pageId: string;
|
|
||||||
readonly wsUrl: string;
|
|
||||||
readonly token: string;
|
|
||||||
readonly createdAt: number;
|
|
||||||
state: SessionState = "connecting";
|
|
||||||
/**
|
|
||||||
* Set true on disconnect/close/auth-failure so a reconnect-driven
|
|
||||||
* unsyncedChanges->0 cannot be mistaken for a successful persist of our
|
|
||||||
* write (preserved verbatim from the one-shot machine).
|
|
||||||
*/
|
|
||||||
connectionLost = false;
|
|
||||||
|
|
||||||
provider: CollabProviderLike | undefined;
|
|
||||||
private readonly ydoc: Y.Doc;
|
|
||||||
private readonly cfg: SessionConfig;
|
|
||||||
/**
|
|
||||||
* Ephemeral sessions (cache disabled, MCP_COLLAB_SESSION_IDLE_MS<=0) are never
|
|
||||||
* registered and self-destroy after their single op — the legacy
|
|
||||||
* provider-per-op behavior.
|
|
||||||
*/
|
|
||||||
private readonly ephemeral: boolean;
|
|
||||||
private readonly opts: AcquireOptions | undefined;
|
|
||||||
|
|
||||||
private dead = false;
|
|
||||||
private connectTimer: ReturnType<typeof setTimeout> | undefined;
|
|
||||||
private idleTimer: ReturnType<typeof setTimeout> | undefined;
|
|
||||||
private openPromise: Promise<void> | undefined;
|
|
||||||
private openResolve: (() => void) | undefined;
|
|
||||||
private openReject: ((err: Error) => void) | undefined;
|
|
||||||
private openSettled = false;
|
|
||||||
/**
|
|
||||||
* The rejector of the CURRENT in-flight mutate, if any. A disconnect/close/
|
|
||||||
* auth-failure or timeout at ANY time rejects the in-flight op through this
|
|
||||||
* with the SAME error text the one-shot machine emitted.
|
|
||||||
*/
|
|
||||||
private inflightReject: ((err: Error) => void) | undefined;
|
|
||||||
|
|
||||||
constructor(
|
|
||||||
key: string,
|
|
||||||
pageId: string,
|
|
||||||
wsUrl: string,
|
|
||||||
token: string,
|
|
||||||
cfg: SessionConfig,
|
|
||||||
ephemeral: boolean,
|
|
||||||
opts: AcquireOptions | undefined,
|
|
||||||
) {
|
|
||||||
this.key = key;
|
|
||||||
this.pageId = pageId;
|
|
||||||
this.wsUrl = wsUrl;
|
|
||||||
this.token = token;
|
|
||||||
this.cfg = cfg;
|
|
||||||
this.ephemeral = ephemeral;
|
|
||||||
this.opts = opts;
|
|
||||||
this.createdAt = Date.now();
|
|
||||||
this.ydoc = new Y.Doc();
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* A cached session may be reused only when it is fully ready, still synced,
|
|
||||||
* has not lost its connection, and has not exceeded its max age (invariant 5
|
|
||||||
* "validate on reuse" + the max-age acquire check).
|
|
||||||
*/
|
|
||||||
isReusable(): boolean {
|
|
||||||
return (
|
|
||||||
!this.dead &&
|
|
||||||
this.state === "ready" &&
|
|
||||||
!this.connectionLost &&
|
|
||||||
!!this.provider &&
|
|
||||||
this.provider.synced === true &&
|
|
||||||
Date.now() - this.createdAt < this.cfg.maxAgeMs
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Connect and wait for the initial sync (onSynced) within CONNECT_TIMEOUT_MS.
|
|
||||||
* Idempotent: repeated calls return the same in-flight/settled promise.
|
|
||||||
*/
|
|
||||||
open(): Promise<void> {
|
|
||||||
if (this.openPromise) return this.openPromise;
|
|
||||||
this.openPromise = new Promise<void>((resolve, reject) => {
|
|
||||||
this.openResolve = resolve;
|
|
||||||
this.openReject = reject;
|
|
||||||
|
|
||||||
this.connectTimer = setTimeout(() => {
|
|
||||||
// The 25s connect timeout: the collab connection never became ready.
|
|
||||||
this.opts?.onConnectTimeout?.();
|
|
||||||
this.teardown(
|
|
||||||
new Error("Connection timeout to collaboration server"),
|
|
||||||
false,
|
|
||||||
);
|
|
||||||
}, CONNECT_TIMEOUT_MS);
|
|
||||||
|
|
||||||
if (process.env.DEBUG)
|
|
||||||
console.error(`Connecting to WebSocket: ${this.wsUrl}`);
|
|
||||||
|
|
||||||
this.provider = providerFactory({
|
|
||||||
url: this.wsUrl,
|
|
||||||
name: `page.${this.pageId}`,
|
|
||||||
document: this.ydoc,
|
|
||||||
token: this.token,
|
|
||||||
WebSocketPolyfill: WebSocket,
|
|
||||||
onConnect: () => {
|
|
||||||
if (process.env.DEBUG) console.error("WS Connect");
|
|
||||||
},
|
|
||||||
// An unexpected disconnect/close at ANY time (during the connect-wait,
|
|
||||||
// between edits, or during a persistence wait) makes the session dead:
|
|
||||||
// surface it now instead of hanging, reject any in-flight op with the
|
|
||||||
// same error text as the one-shot machine, and remove ourselves from
|
|
||||||
// the registry so the next acquire opens fresh. `teardown` is idempotent
|
|
||||||
// so the onClose our own destroy() triggers is a harmless no-op.
|
|
||||||
onDisconnect: () => {
|
|
||||||
if (process.env.DEBUG) console.error("WS Disconnect");
|
|
||||||
this.teardown(
|
|
||||||
new Error(
|
|
||||||
"Collaboration connection closed before the update was persisted/synced",
|
|
||||||
),
|
|
||||||
true,
|
|
||||||
);
|
|
||||||
},
|
|
||||||
onClose: () => {
|
|
||||||
if (process.env.DEBUG) console.error("WS Close");
|
|
||||||
this.teardown(
|
|
||||||
new Error(
|
|
||||||
"Collaboration connection closed before the update was persisted/synced",
|
|
||||||
),
|
|
||||||
true,
|
|
||||||
);
|
|
||||||
},
|
|
||||||
onSynced: () => {
|
|
||||||
if (this.dead || this.openSettled) return;
|
|
||||||
if (process.env.DEBUG) console.error("Connected and synced!");
|
|
||||||
if (this.connectTimer) {
|
|
||||||
clearTimeout(this.connectTimer);
|
|
||||||
this.connectTimer = undefined;
|
|
||||||
}
|
|
||||||
this.state = "ready";
|
|
||||||
this.openSettled = true;
|
|
||||||
this.openResolve?.();
|
|
||||||
},
|
|
||||||
onAuthenticationFailed: () => {
|
|
||||||
this.teardown(
|
|
||||||
new Error("Authentication failed for collaboration connection"),
|
|
||||||
true,
|
|
||||||
);
|
|
||||||
},
|
|
||||||
});
|
|
||||||
});
|
|
||||||
return this.openPromise;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Run one atomic read -> transform -> write against the LIVE doc and wait for
|
|
||||||
* the server to acknowledge the write.
|
|
||||||
*
|
|
||||||
* INVARIANT 1 (read->write atomicity): between `TiptapTransformer.fromYdoc`
|
|
||||||
* and `applyDocToFragment` there is NO `await`. Yjs applies remote updates
|
|
||||||
* only when the event loop yields, so this synchronous block sees a consistent
|
|
||||||
* live doc and no concurrent human edit can interleave and be clobbered —
|
|
||||||
* exactly as in the one-shot onSynced code, just on a persistent provider.
|
|
||||||
*
|
|
||||||
* INVARIANT 2 (per-edit ack): after the write, resolve immediately if
|
|
||||||
* unsyncedChanges is already 0, else wait for the unsyncedChanges->0 event
|
|
||||||
* (PERSIST_TIMEOUT_MS), guarded by connectionLost so a reconnect handshake
|
|
||||||
* cannot report a false success.
|
|
||||||
*
|
|
||||||
* CONCURRENCY: not safe to invoke concurrently on ONE session — the caller
|
|
||||||
* MUST serialize (hold the per-page lock), mirroring acquireCollabSession.
|
|
||||||
* The in-flight op is tracked in a single `inflightReject` field, so an
|
|
||||||
* overlapping second call would clobber the first's rejector and leave it
|
|
||||||
* hanging on disconnect. A fail-fast guard below rejects the overlap instead.
|
|
||||||
* Sequential (awaited) mutates are fine: localFinish clears inflightReject
|
|
||||||
* before the promise settles, so the guard is clear by the time the next runs.
|
|
||||||
*/
|
|
||||||
mutate(
|
|
||||||
transform: (liveDoc: any) => any | null,
|
|
||||||
): Promise<MutationResult> {
|
|
||||||
// Belt-and-suspenders (acquire already validated): refuse to write on a
|
|
||||||
// session that is not in a live, synced, ready state.
|
|
||||||
if (
|
|
||||||
this.dead ||
|
|
||||||
this.state !== "ready" ||
|
|
||||||
this.connectionLost ||
|
|
||||||
!this.provider ||
|
|
||||||
this.provider.synced !== true
|
|
||||||
) {
|
|
||||||
return Promise.reject(
|
|
||||||
new Error("Collaboration session is not in a ready state"),
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
// Fail-fast on concurrent use: a second overlapping mutate would overwrite
|
|
||||||
// the first's inflightReject, so a disconnect would only reject the second
|
|
||||||
// and hang the first until PERSIST_TIMEOUT_MS. Reject the overlap WITHOUT
|
|
||||||
// touching the in-flight op's state (no localFinish/teardown here).
|
|
||||||
if (this.inflightReject) {
|
|
||||||
return Promise.reject(
|
|
||||||
new Error(
|
|
||||||
"mutate already in-flight; caller must serialize (hold the page lock)",
|
|
||||||
),
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
return new Promise<MutationResult>((resolve, reject) => {
|
|
||||||
let settled = false;
|
|
||||||
let persistTimer: ReturnType<typeof setTimeout> | undefined;
|
|
||||||
let unsyncedHandler:
|
|
||||||
| ((data: { number: number }) => void)
|
|
||||||
| undefined;
|
|
||||||
// The verifiable result resolved on every success/abort path. Set on
|
|
||||||
// abort (no-op report) and after a real write (computed change report).
|
|
||||||
let mutationResult: MutationResult;
|
|
||||||
|
|
||||||
const localFinish = (err: Error | null, value?: MutationResult) => {
|
|
||||||
if (settled) return;
|
|
||||||
settled = true;
|
|
||||||
if (persistTimer) clearTimeout(persistTimer);
|
|
||||||
if (unsyncedHandler && this.provider) {
|
|
||||||
try {
|
|
||||||
this.provider.off("unsyncedChanges", unsyncedHandler);
|
|
||||||
} catch (e) {}
|
|
||||||
}
|
|
||||||
this.inflightReject = undefined;
|
|
||||||
if (err) reject(err);
|
|
||||||
else resolve(value as MutationResult);
|
|
||||||
// Post-settle lifecycle: an ephemeral (cache-disabled) session dies with
|
|
||||||
// its single op; a cached session that is still alive re-arms its idle
|
|
||||||
// TTL so the clock starts from the LAST op.
|
|
||||||
if (this.ephemeral) {
|
|
||||||
this.destroy("ephemeral op complete");
|
|
||||||
} else if (!this.dead) {
|
|
||||||
this.armIdle();
|
|
||||||
}
|
|
||||||
};
|
|
||||||
|
|
||||||
// Register so a disconnect/close/auth-failure/teardown rejects THIS op
|
|
||||||
// with the connection-loss error text. localFinish's `settled` guard makes
|
|
||||||
// a racing teardown + normal resolve safe (first one wins).
|
|
||||||
this.inflightReject = (e: Error) => localFinish(e);
|
|
||||||
|
|
||||||
// Resolve once the server acknowledges our update: the provider increments
|
|
||||||
// unsyncedChanges when the local update is sent and decrements it on the
|
|
||||||
// server's SyncStatus(applied=true); reaching 0 means the authoritative
|
|
||||||
// in-memory ydoc on the server now contains our write.
|
|
||||||
const waitForPersistence = () => {
|
|
||||||
if (settled) return;
|
|
||||||
// A missing provider is a failure, not a success: without it the write
|
|
||||||
// can never have been acknowledged.
|
|
||||||
if (!this.provider) {
|
|
||||||
localFinish(new Error("collab provider gone before persistence"));
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
if (this.provider.unsyncedChanges === 0) {
|
|
||||||
localFinish(null, mutationResult);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
persistTimer = setTimeout(() => {
|
|
||||||
localFinish(
|
|
||||||
new Error(
|
|
||||||
"Timeout waiting for collaboration server to persist the update",
|
|
||||||
),
|
|
||||||
);
|
|
||||||
}, PERSIST_TIMEOUT_MS);
|
|
||||||
unsyncedHandler = (data: { number: number }) => {
|
|
||||||
// Only treat unsyncedChanges->0 as success when the connection is
|
|
||||||
// still up. A transient disconnect + reconnect handshake can drive the
|
|
||||||
// counter back to 0 without our write being re-transmitted; in that
|
|
||||||
// case let the disconnect/close error win instead.
|
|
||||||
if (data.number === 0 && !this.connectionLost) {
|
|
||||||
localFinish(null, mutationResult);
|
|
||||||
}
|
|
||||||
};
|
|
||||||
this.provider.on("unsyncedChanges", unsyncedHandler);
|
|
||||||
};
|
|
||||||
|
|
||||||
// CRITICAL: everything between reading the live doc and writing it back
|
|
||||||
// must stay synchronous (no await). While the JS event loop is not
|
|
||||||
// yielded, no incoming remote update can interleave, so any already-synced
|
|
||||||
// concurrent edits are preserved in liveDoc.
|
|
||||||
let newDoc: any;
|
|
||||||
let beforeDoc: any;
|
|
||||||
try {
|
|
||||||
let liveDoc = TiptapTransformer.fromYdoc(this.ydoc, "default");
|
|
||||||
if (
|
|
||||||
!liveDoc ||
|
|
||||||
typeof liveDoc !== "object" ||
|
|
||||||
!Array.isArray(liveDoc.content)
|
|
||||||
) {
|
|
||||||
liveDoc = { type: "doc", content: [] };
|
|
||||||
}
|
|
||||||
|
|
||||||
// Snapshot the before-doc for the change report. Docs are
|
|
||||||
// JSON-serializable, so this is a safe deep clone.
|
|
||||||
beforeDoc = JSON.parse(JSON.stringify(liveDoc));
|
|
||||||
|
|
||||||
newDoc = transform(liveDoc);
|
|
||||||
|
|
||||||
if (newDoc == null) {
|
|
||||||
// Transform aborted — write nothing, return the live doc with a no-op
|
|
||||||
// change report.
|
|
||||||
mutationResult = {
|
|
||||||
doc: liveDoc,
|
|
||||||
verify: {
|
|
||||||
changed: false,
|
|
||||||
textInserted: 0,
|
|
||||||
textDeleted: 0,
|
|
||||||
blocksChanged: 0,
|
|
||||||
marks: {},
|
|
||||||
summary: "no changes (transform aborted)",
|
|
||||||
},
|
|
||||||
};
|
|
||||||
localFinish(null, mutationResult);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
// Structural diff into the live fragment (issue #152): preserves the Yjs
|
|
||||||
// ids of unchanged nodes, so an open editor's cursor is not yanked to the
|
|
||||||
// end of the document on every agent write.
|
|
||||||
applyDocToFragment(this.ydoc, newDoc);
|
|
||||||
} catch (e) {
|
|
||||||
// Includes errors thrown by transform (e.g. "afterText not found",
|
|
||||||
// "text not found"): propagate them verbatim to the caller.
|
|
||||||
localFinish(e instanceof Error ? e : new Error(String(e)));
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
// Compute the verifiable change report AFTER the transact write: it only
|
|
||||||
// needs the JSON before/after, so it cannot affect the atomic read->write
|
|
||||||
// window, and summarizeChange never throws.
|
|
||||||
mutationResult = {
|
|
||||||
doc: newDoc,
|
|
||||||
verify: summarizeChange(beforeDoc, newDoc),
|
|
||||||
};
|
|
||||||
if (process.env.DEBUG)
|
|
||||||
console.error("Content written, waiting for server to persist...");
|
|
||||||
waitForPersistence();
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
/** (Re)arm the idle TTL so the clock starts from the most recent activity. */
|
|
||||||
armIdle(): void {
|
|
||||||
if (this.dead || this.ephemeral) return;
|
|
||||||
if (this.idleTimer) clearTimeout(this.idleTimer);
|
|
||||||
if (this.cfg.idleMs > 0) {
|
|
||||||
this.idleTimer = setTimeout(() => {
|
|
||||||
this.destroy("idle timeout");
|
|
||||||
}, this.cfg.idleMs);
|
|
||||||
// Never let the idle timer keep the process alive.
|
|
||||||
(this.idleTimer as any).unref?.();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Idempotent teardown: mark dead, clear timers, remove from the registry, fail
|
|
||||||
* any pending open/in-flight op, and destroy the provider. `inflightError` is
|
|
||||||
* the error a pending open or in-flight op is rejected with; `connectionLoss`
|
|
||||||
* marks the session as connection-lost so the ack guard cannot report a false
|
|
||||||
* success on a racing unsyncedChanges->0.
|
|
||||||
*/
|
|
||||||
private teardown(inflightError: Error | null, connectionLoss: boolean): void {
|
|
||||||
if (this.dead) return;
|
|
||||||
this.dead = true;
|
|
||||||
this.state = "dead";
|
|
||||||
if (connectionLoss) this.connectionLost = true;
|
|
||||||
|
|
||||||
if (this.connectTimer) {
|
|
||||||
clearTimeout(this.connectTimer);
|
|
||||||
this.connectTimer = undefined;
|
|
||||||
}
|
|
||||||
if (this.idleTimer) {
|
|
||||||
clearTimeout(this.idleTimer);
|
|
||||||
this.idleTimer = undefined;
|
|
||||||
}
|
|
||||||
|
|
||||||
// Remove ourselves from the registry (only if we are still the live entry —
|
|
||||||
// a re-open under the same key must not be evicted by our teardown).
|
|
||||||
if (sessions.get(this.key) === this) {
|
|
||||||
sessions.delete(this.key);
|
|
||||||
}
|
|
||||||
|
|
||||||
// Fail a pending open() and any in-flight mutate with the terminal error.
|
|
||||||
if (!this.openSettled) {
|
|
||||||
this.openSettled = true;
|
|
||||||
this.openReject?.(
|
|
||||||
inflightError ?? new Error("Collaboration session destroyed"),
|
|
||||||
);
|
|
||||||
}
|
|
||||||
if (this.inflightReject) {
|
|
||||||
const rej = this.inflightReject;
|
|
||||||
this.inflightReject = undefined;
|
|
||||||
rej(inflightError ?? new Error("Collaboration session destroyed"));
|
|
||||||
}
|
|
||||||
|
|
||||||
if (this.provider) {
|
|
||||||
try {
|
|
||||||
this.provider.destroy();
|
|
||||||
} catch (e) {}
|
|
||||||
this.provider = undefined;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Public idempotent teardown used by the acquire/eviction paths and by a
|
|
||||||
* caller that wants the session dropped after a failed op ("next call
|
|
||||||
* reconnects fresh").
|
|
||||||
*/
|
|
||||||
destroy(reason: string): void {
|
|
||||||
if (this.dead) return;
|
|
||||||
if (process.env.DEBUG)
|
|
||||||
console.error(`Destroying collab session ${this.pageId}: ${reason}`);
|
|
||||||
this.teardown(new Error(`Collaboration session destroyed: ${reason}`), false);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/** key = wsUrl + pageId + collabToken (identity isolation: invariant 4). */
|
|
||||||
const sessions = new Map<string, CollabSession>();
|
|
||||||
|
|
||||||
function sessionKey(wsUrl: string, pageId: string, token: string): string {
|
|
||||||
// The token is part of the key so sessions are NEVER shared between different
|
|
||||||
// users' MCP sessions (HTTP mode), and a token rotation makes a new entry
|
|
||||||
// while the old one idles out.
|
|
||||||
return `${wsUrl} | |||||||