Merge pull request 'feat(#370 Stage A): MCP save_page_version — агентские версии страниц' (#565) from feat/370-agent-save-version into develop
Reviewed-on: #565
This commit was merged in pull request #565.
This commit is contained in:
@@ -346,6 +346,9 @@ roles:
|
||||
□ Working sections ("Log", "Open Questions", "Revision") are moved to an
|
||||
appendix at the end of the document or clearly separated from the report
|
||||
body.
|
||||
□ Once the report is complete, call save_page_version to pin the finished
|
||||
document as a restorable named version (a checkpoint the reader can
|
||||
return to). A repeat call after no further edits is a harmless no-op.
|
||||
Be honest about gaps. If you couldn't find something, say so — don't disguise
|
||||
a guess as a fact.
|
||||
autoStart: false
|
||||
@@ -442,6 +445,7 @@ roles:
|
||||
- **The language of the notes = the main language of the call.** Technical terms — in their canonical spelling (usually Latin).
|
||||
- **Don't evaluate the participants** and don't comment on the quality of the discussion.
|
||||
- The output is the notes only, with no preambles or meta-comments, apart from targeted uncertainty marks.
|
||||
- **Pin the result.** Once the notes are complete on the page, call save_page_version to save them as a restorable named version (a checkpoint). Calling it again after no further edits is a harmless no-op.
|
||||
|
||||
## Style example (excerpt)
|
||||
|
||||
|
||||
@@ -345,6 +345,9 @@ roles:
|
||||
□ Working sections («Журнал», «Открытые вопросы», «Ревизия») are moved to
|
||||
an appendix at the end of the document or clearly separated from the
|
||||
report body.
|
||||
□ Once the report is complete, call save_page_version to pin the finished
|
||||
document as a restorable named version (a checkpoint the reader can
|
||||
return to). A repeat call after no further edits is a harmless no-op.
|
||||
Be honest about gaps. If you couldn't find something, say so — don't disguise
|
||||
a guess as a fact.
|
||||
autoStart: false
|
||||
@@ -441,6 +444,7 @@ roles:
|
||||
- **Язык конспекта = основной язык созвона.** Технические термины — в каноническом написании (обычно латиницей).
|
||||
- **Не оценивай участников** и не комментируй качество обсуждения.
|
||||
- На выходе — только конспект, без преамбул и мета-комментариев, кроме точечных пометок неуверенности.
|
||||
- **Зафиксируй результат.** Когда конспект готов на странице, вызови save_page_version, чтобы сохранить его как именованную версию (точку восстановления). Повторный вызов без новых правок — безвредный no-op.
|
||||
|
||||
## Пример стиля (фрагмент)
|
||||
|
||||
|
||||
@@ -33,6 +33,6 @@ bundles:
|
||||
- en
|
||||
roles:
|
||||
- slug: researcher
|
||||
version: 9
|
||||
version: 10
|
||||
- slug: call-summarizer
|
||||
version: 1
|
||||
version: 2
|
||||
|
||||
@@ -736,6 +736,53 @@ describe('PersistenceExtension.onStoreDocument — Approach-A boundary snapshot'
|
||||
expect(historyQueue.remove).toHaveBeenCalledWith(PAGE_ID);
|
||||
expect(historyQueue.remove).not.toHaveBeenCalledWith(SLUG);
|
||||
});
|
||||
|
||||
// #370 F2 — an effectively-empty page is a REACHABLE no-op (agent calls
|
||||
// save_page_version on a blank page): the version tx short-circuits with
|
||||
// nothing to pin. The handler MUST still broadcast a TERMINAL reply
|
||||
// (version.skipped, reason:'empty') so the client resolves at once instead of
|
||||
// waiting out its 20s ack timeout and misreporting a healthy server as
|
||||
// unreachable. MUTATION: drop the `else if (skipped)` broadcast → no terminal
|
||||
// reply is sent → this reddens.
|
||||
it('empty page → no version written, broadcasts a terminal version.skipped(empty)', async () => {
|
||||
const emptyDoc = { type: 'doc', content: [{ type: 'paragraph' }] };
|
||||
const document = ydocFor(emptyDoc);
|
||||
pageRepo.findById.mockResolvedValue({
|
||||
...persistedHumanPage('IGNORED'),
|
||||
content: emptyDoc,
|
||||
});
|
||||
|
||||
await emitSave(document, 'agent');
|
||||
|
||||
// Nothing pinned.
|
||||
expect(pageHistoryRepo.saveHistory).not.toHaveBeenCalled();
|
||||
expect(pageHistoryRepo.updateHistoryKind).not.toHaveBeenCalled();
|
||||
// But a terminal reply WAS sent so the client never times out. The flush
|
||||
// (onStoreDocument) emits its own `page.updated`; the version.skipped is the
|
||||
// LAST broadcast (dropping the skip branch leaves page.updated last → reds).
|
||||
const calls = (document as any).broadcastStateless.mock.calls;
|
||||
const msg = JSON.parse(calls[calls.length - 1][0]);
|
||||
expect(msg).toEqual({ type: 'version.skipped', reason: 'empty' });
|
||||
});
|
||||
|
||||
// #370 F2 — the page row is gone (deleted / never persisted). Same rule: a
|
||||
// terminal reply MUST be sent (version.skipped, reason:'page-not-found') so the
|
||||
// client surfaces a truthful "not found" immediately rather than a health
|
||||
// timeout. onStoreDocument's own `!page` guard returns early without throwing,
|
||||
// so the handler reaches the version tx and its `!page` skip branch.
|
||||
it('page not found → broadcasts a terminal version.skipped(page-not-found)', async () => {
|
||||
const document = ydocFor(doc('GONE'));
|
||||
pageRepo.findById.mockResolvedValue(null);
|
||||
|
||||
await emitSave(document, 'agent');
|
||||
|
||||
expect(pageHistoryRepo.saveHistory).not.toHaveBeenCalled();
|
||||
expect((document as any).broadcastStateless).toHaveBeenCalledTimes(1);
|
||||
const msg = JSON.parse(
|
||||
(document as any).broadcastStateless.mock.calls[0][0],
|
||||
);
|
||||
expect(msg).toEqual({ type: 'version.skipped', reason: 'page-not-found' });
|
||||
});
|
||||
});
|
||||
|
||||
// #370 — the in-memory idle-burst marker must be dropped on doc unload (like
|
||||
|
||||
@@ -68,6 +68,19 @@ export const INTENTIONAL_CLEAR_MESSAGE_TYPE = 'intentional-clear';
|
||||
*/
|
||||
export const SAVE_VERSION_MESSAGE_TYPE = 'save-version';
|
||||
|
||||
/**
|
||||
* #370 F2 — wire format of the server→client REPLY to a save-version signal, sent
|
||||
* over the same stateless channel. `version.saved` means a version was created or
|
||||
* promoted; `version.skipped` is a TERMINAL "nothing was pinned" reply for the two
|
||||
* reachable no-op cases (an effectively-empty page, or the page row is gone) so the
|
||||
* client resolves at once instead of waiting out its ack timeout and misreporting a
|
||||
* healthy server as unreachable. EXACTLY ONE of these is broadcast per handled
|
||||
* save. The MCP client duplicates these literals (it cannot import server code) —
|
||||
* keep the two in sync (see packages/mcp/src/lib/collaboration.ts).
|
||||
*/
|
||||
export const VERSION_SAVED_MESSAGE_TYPE = 'version.saved';
|
||||
export const VERSION_SKIPPED_MESSAGE_TYPE = 'version.skipped';
|
||||
|
||||
/**
|
||||
* #251 — how long an intentional-clear signal stays "pending" before it is
|
||||
* ignored. The signal is set on the clearing keystroke but consumed by the
|
||||
@@ -643,6 +656,10 @@ export class PersistenceExtension implements Extension {
|
||||
let result:
|
||||
| { historyId: string; kind: PageHistoryKind; alreadySaved: boolean }
|
||||
| undefined;
|
||||
// #370 F2 — set when there is nothing to version (empty page / page gone), so
|
||||
// the tail broadcasts a terminal `version.skipped` instead of staying silent
|
||||
// and forcing the client to time out. Mutually exclusive with `result`.
|
||||
let skipped: 'empty' | 'page-not-found' | undefined;
|
||||
|
||||
// #370 F8-twin — the contributor set popped from Redis (destructive SPOP)
|
||||
// must be restored if the version row does not durably land. The inner
|
||||
@@ -668,11 +685,20 @@ export class PersistenceExtension implements Extension {
|
||||
includeContent: true,
|
||||
trx,
|
||||
});
|
||||
if (!page) return;
|
||||
if (!page) {
|
||||
// The page row is gone (deleted/never persisted). Nothing to version —
|
||||
// record it so the tail sends a terminal skip reply (#370 F2).
|
||||
skipped = 'page-not-found';
|
||||
return;
|
||||
}
|
||||
versionedPageId = page.id;
|
||||
// Never version an effectively-empty page (mirrors the processor's
|
||||
// first-history guard); there is nothing intentional to pin.
|
||||
if (isEmptyParagraphDoc(page.content as any)) return;
|
||||
// first-history guard); there is nothing intentional to pin. Record the
|
||||
// skip so the client gets a terminal reply rather than a timeout (#370 F2).
|
||||
if (isEmptyParagraphDoc(page.content as any)) {
|
||||
skipped = 'empty';
|
||||
return;
|
||||
}
|
||||
|
||||
const lastHistory = await this.pageHistoryRepo.findPageLastHistory(
|
||||
page.id,
|
||||
@@ -747,15 +773,26 @@ export class PersistenceExtension implements Extension {
|
||||
}
|
||||
this.idleBurstStart.delete(documentName);
|
||||
|
||||
// EXACTLY ONE terminal reply per handled save (#370 F2): a real save/promote
|
||||
// broadcasts `version.saved`; the two no-op early returns (empty page / page
|
||||
// gone) broadcast `version.skipped` so the client never waits out its ack
|
||||
// timeout. A genuine failure threw above and rejected before reaching here.
|
||||
if (result) {
|
||||
document.broadcastStateless(
|
||||
JSON.stringify({
|
||||
type: 'version.saved',
|
||||
type: VERSION_SAVED_MESSAGE_TYPE,
|
||||
historyId: result.historyId,
|
||||
kind: result.kind,
|
||||
alreadySaved: result.alreadySaved,
|
||||
}),
|
||||
);
|
||||
} else if (skipped) {
|
||||
document.broadcastStateless(
|
||||
JSON.stringify({
|
||||
type: VERSION_SKIPPED_MESSAGE_TYPE,
|
||||
reason: skipped,
|
||||
}),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -106,6 +106,7 @@ function __assertClientCallContract(client: DocmostClientLike): void {
|
||||
void client.sharePage(s, true);
|
||||
void client.unsharePage(s);
|
||||
void client.restorePageVersion(s);
|
||||
void client.savePageVersion(s);
|
||||
void client.transformPage(s, s, { dryRun: true });
|
||||
void client.stashPage(s);
|
||||
// --- write (image / footnote), in-app since #410 ---
|
||||
|
||||
@@ -58,6 +58,7 @@ type DocmostClientMethod =
|
||||
| 'sharePage'
|
||||
| 'unsharePage'
|
||||
| 'restorePageVersion'
|
||||
| 'savePageVersion'
|
||||
| 'transformPage'
|
||||
| 'stashPage'
|
||||
// --- write (image / footnote), in-app since #410 ---
|
||||
|
||||
@@ -11,6 +11,7 @@ import {
|
||||
markdownToProseMirror,
|
||||
markdownToProseMirrorCanonical,
|
||||
mutatePageContent,
|
||||
savePageVersionRealtime,
|
||||
assertYjsEncodable,
|
||||
MutationResult,
|
||||
} from "../lib/collaboration.js";
|
||||
@@ -55,6 +56,7 @@ export interface IPagesMixin {
|
||||
listPageHistory(pageId: string, cursor?: string): any;
|
||||
getPageHistory(historyId: string): any;
|
||||
restorePageVersion(historyId: string): any;
|
||||
savePageVersion(pageId: string): any;
|
||||
diffPageVersions(pageId: string, from?: string, to?: string): any;
|
||||
}
|
||||
|
||||
@@ -587,6 +589,24 @@ export function PagesMixin<TBase extends GConstructor<DocmostClientContext>>(Bas
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Save an intentional NAMED version of a page's CURRENT live content (#370).
|
||||
* The write goes over the same agent-authenticated collab session content edits
|
||||
* use, so the server derives kind='agent' from the signed actor. Resolves a
|
||||
* SaveVersionResult: `{ saved:true, historyId, kind, alreadySaved }` on success,
|
||||
* or `{ saved:false, skipped:true, reason }` when the server had nothing to pin
|
||||
* (e.g. an empty page). A stale/missing pageId throws (not a benign skip).
|
||||
*/
|
||||
async savePageVersion(pageId: string) {
|
||||
const collabToken = await this.getCollabTokenWithReauth();
|
||||
const pageUuid = await this.resolvePageId(pageId);
|
||||
// Self-heal a rejected WS handshake once (#486): re-mint the collab token and
|
||||
// retry, symmetric to the content-write path.
|
||||
return this.writeWithCollabAuthRetry(collabToken, (token) =>
|
||||
savePageVersionRealtime(pageUuid, token, this.apiUrl),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Diff two versions of a page and return a Docmost-equivalent change set.
|
||||
* `from`/`to` each resolve to a ProseMirror doc:
|
||||
|
||||
@@ -139,7 +139,13 @@ export interface CollabProviderLike {
|
||||
unsyncedChanges: number;
|
||||
destroy(): void;
|
||||
on(event: "unsyncedChanges", handler: (data: { number: number }) => void): void;
|
||||
on(event: "stateless", handler: (data: { payload: string }) => void): void;
|
||||
off(event: "unsyncedChanges", handler: (data: { number: number }) => void): void;
|
||||
off(event: "stateless", handler: (data: { payload: string }) => void): void;
|
||||
// Send a stateless (out-of-band, non-doc) message to the server. Used by the
|
||||
// #370 explicit save-version path (payload `{type:'save-version'}`); the server
|
||||
// replies with a broadcast `{type:'version.saved', …}` on the same channel.
|
||||
sendStateless(payload: string): void;
|
||||
}
|
||||
|
||||
/** The configuration object passed to the provider factory. */
|
||||
@@ -564,6 +570,125 @@ export class CollabSession {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a stateless message over the live collaboration connection and resolve
|
||||
* with the FIRST reply whose parsed payload satisfies `predicate`, rejecting on
|
||||
* a bounded `timeoutMs`.
|
||||
*
|
||||
* Used by the #370 explicit save-version path: the client sends
|
||||
* `{type:'save-version'}` and awaits the server's broadcast
|
||||
* `{type:'version.saved', historyId, kind, alreadySaved}`. The stateless channel
|
||||
* (not a REST read) is deliberate — the server versions the LIVE in-memory ydoc,
|
||||
* which the up-to-10s-stale page row would not yet reflect.
|
||||
*
|
||||
* `predicate` receives each incoming stateless message (already JSON-parsed) and
|
||||
* returns the resolved value for a match or `undefined` to keep waiting; a
|
||||
* malformed / unrelated payload is ignored. The listener is registered BEFORE
|
||||
* the send so a fast reply is never missed.
|
||||
*
|
||||
* Lifecycle mirrors mutate(): it registers through `inflightReject` so a
|
||||
* disconnect/close/auth-failure/teardown rejects THIS wait with the same
|
||||
* connection-loss error, refuses to run on a non-ready session, fails fast on a
|
||||
* concurrent in-flight op (the caller MUST hold the per-page lock), and re-arms
|
||||
* the idle TTL (or self-destroys an ephemeral session) on completion.
|
||||
*
|
||||
* NOTE: the server broadcast carries no request-correlation id, so under a
|
||||
* genuinely concurrent save on the SAME page (e.g. a human Cmd+S racing the
|
||||
* agent) this resolves on whichever `version.saved` arrives first. The per-page
|
||||
* lock serializes THIS process's saves; a cross-client race is inherent to the
|
||||
* broadcast design and is benign (the result still describes a real save of this
|
||||
* page's current content, and the server's save is promote-not-duplicate).
|
||||
*/
|
||||
sendStatelessAndAwait<T>(
|
||||
payload: string,
|
||||
predicate: (message: any) => T | undefined,
|
||||
timeoutMs: number,
|
||||
): Promise<T> {
|
||||
// Belt-and-suspenders (acquire already validated): refuse to send 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 op would overwrite the
|
||||
// first's inflightReject and hang it on disconnect (same guard as mutate).
|
||||
if (this.inflightReject) {
|
||||
return Promise.reject(
|
||||
new Error(
|
||||
"stateless op already in-flight; caller must serialize (hold the page lock)",
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
return new Promise<T>((resolve, reject) => {
|
||||
let settled = false;
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
let statelessHandler: ((data: { payload: string }) => void) | undefined;
|
||||
|
||||
const localFinish = (err: Error | null, value?: T) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
if (timer) clearTimeout(timer);
|
||||
if (statelessHandler && this.provider) {
|
||||
try {
|
||||
this.provider.off("stateless", statelessHandler);
|
||||
} catch (e) {}
|
||||
}
|
||||
this.inflightReject = undefined;
|
||||
if (err) reject(err);
|
||||
else resolve(value as T);
|
||||
// Post-settle lifecycle, identical to mutate's localFinish.
|
||||
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 (the `settled` guard makes a racing
|
||||
// teardown + normal resolve safe — first one wins).
|
||||
this.inflightReject = (e: Error) => localFinish(e);
|
||||
|
||||
// Register the reply listener BEFORE sending so a fast server broadcast is
|
||||
// never missed.
|
||||
statelessHandler = (data: { payload: string }) => {
|
||||
if (settled) return;
|
||||
let message: any;
|
||||
try {
|
||||
message = JSON.parse(data.payload);
|
||||
} catch {
|
||||
return; // unrelated / malformed stateless message — keep waiting
|
||||
}
|
||||
const matched = predicate(message);
|
||||
if (matched !== undefined) localFinish(null, matched);
|
||||
};
|
||||
this.provider!.on("stateless", statelessHandler);
|
||||
|
||||
timer = setTimeout(() => {
|
||||
localFinish(
|
||||
new Error(
|
||||
`Timeout waiting for a stateless reply from the collaboration server ${this.hint()}`,
|
||||
),
|
||||
);
|
||||
}, timeoutMs);
|
||||
|
||||
try {
|
||||
this.provider!.sendStateless(payload);
|
||||
} catch (e) {
|
||||
localFinish(e instanceof Error ? e : new Error(String(e)));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/** (Re)arm the idle TTL so the clock starts from the most recent activity. */
|
||||
armIdle(): void {
|
||||
if (this.dead || this.ephemeral) return;
|
||||
|
||||
@@ -324,6 +324,146 @@ export async function mutatePageContent(
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Stateless-channel message types for the #370 explicit save-version handshake.
|
||||
* These MUST match the server constants in
|
||||
* apps/server/src/collaboration/extensions/persistence.extension.ts
|
||||
* (SAVE_VERSION_MESSAGE_TYPE and the VERSION_SAVED / VERSION_SKIPPED replies) —
|
||||
* the mcp package cannot import server code, so the literals are duplicated here.
|
||||
* KEEP THE TWO SIDES IN SYNC: the server broadcasts exactly ONE terminal reply per
|
||||
* handled save (`version.saved` for a real save/promote, `version.skipped` for a
|
||||
* reachable no-op — an empty page or a missing page row); the client must match
|
||||
* BOTH or it waits out the ack timeout on the skip case and misreports a healthy
|
||||
* server as unreachable (#370 F2).
|
||||
*/
|
||||
const SAVE_VERSION_MESSAGE_TYPE = "save-version";
|
||||
const VERSION_SAVED_MESSAGE_TYPE = "version.saved";
|
||||
const VERSION_SKIPPED_MESSAGE_TYPE = "version.skipped";
|
||||
|
||||
/**
|
||||
* Bounded wait for the server's terminal reply after we ask it to save a version.
|
||||
* The server flushes the live ydoc through its store path and writes a history row
|
||||
* inside a DB transaction before broadcasting, so allow the same headroom as a
|
||||
* persistence ack; reject (do NOT hang) past it. A timeout here means NO reply at
|
||||
* all arrived — the genuine "collab server unreachable/overloaded" case — as
|
||||
* distinct from a `version.skipped` reply, which resolves immediately.
|
||||
*/
|
||||
const SAVE_VERSION_ACK_TIMEOUT_MS = 20000;
|
||||
|
||||
/**
|
||||
* The resolved outcome of an explicit save-version, surfaced to the tool caller.
|
||||
* `saved:true` → a version was created or promoted (historyId/kind/alreadySaved
|
||||
* are present). `saved:false` → the server had nothing to pin (a reachable no-op,
|
||||
* e.g. an empty page); `skipped` + `reason` explain why. A page-not-found reply is
|
||||
* NOT represented here — it is surfaced as a thrown error (a bad/stale pageId).
|
||||
*/
|
||||
export interface SaveVersionResult {
|
||||
saved: boolean;
|
||||
historyId?: string;
|
||||
kind?: string;
|
||||
alreadySaved?: boolean;
|
||||
skipped?: boolean;
|
||||
reason?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Parsed terminal reply the predicate hands back to savePageVersionRealtime. Kept
|
||||
* internal: it carries the page-not-found case (mapped to a thrown error, never a
|
||||
* returned result) that SaveVersionResult deliberately does not model.
|
||||
*/
|
||||
type SaveVersionReply =
|
||||
| { outcome: "saved"; historyId: string; kind: string; alreadySaved: boolean }
|
||||
| { outcome: "skipped"; reason: string };
|
||||
|
||||
/** Match the server's terminal reply (version.saved / version.skipped), ignoring
|
||||
* any unrelated / cross-page stateless message so the wait is not resolved by
|
||||
* noise. Returns `undefined` to keep waiting (#370 F1 predicate coverage). */
|
||||
function matchSaveVersionReply(message: any): SaveVersionReply | undefined {
|
||||
if (!message || typeof message !== "object") return undefined;
|
||||
if (message.type === VERSION_SAVED_MESSAGE_TYPE) {
|
||||
return {
|
||||
outcome: "saved",
|
||||
historyId: String(message.historyId),
|
||||
kind: String(message.kind),
|
||||
alreadySaved: !!message.alreadySaved,
|
||||
};
|
||||
}
|
||||
if (message.type === VERSION_SKIPPED_MESSAGE_TYPE) {
|
||||
return { outcome: "skipped", reason: String(message.reason ?? "unknown") };
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Save an intentional version of a page's CURRENT live collaboration content
|
||||
* (#370). Runs under the per-page lock, acquires the SAME cached CollabSession the
|
||||
* content writes use (#400) — so it authenticates with the caller's agent collab
|
||||
* token, and the server derives kind='agent' from that signed actor — then sends a
|
||||
* `{type:'save-version'}` stateless message and awaits the server's terminal reply
|
||||
* (`version.saved` or `version.skipped`) on the same channel.
|
||||
*
|
||||
* Deliberately does NOT read `pages.content` over REST: the versioned content is
|
||||
* the live in-memory ydoc, which the debounced (up-to-10s-stale) page row would
|
||||
* not yet reflect. The stateless round-trip is what makes the save exact.
|
||||
*
|
||||
* Outcomes:
|
||||
* - a real save/promote → `{ saved:true, historyId, kind, alreadySaved }`;
|
||||
* - the server had nothing to pin (empty page) → `{ saved:false, skipped:true,
|
||||
* reason:'empty' }` — a clean, immediate no-op, NOT a stall;
|
||||
* - the page row is gone (a stale/bad pageId) → a thrown error, immediate and
|
||||
* truthful (not the health-timeout path);
|
||||
* - no reply within the timeout → a thrown timeout error (the genuine "server
|
||||
* unreachable/overloaded" case).
|
||||
*
|
||||
* On a TRANSPORT failure (timeout / disconnect) the session is destroyed so the
|
||||
* next call reconnects fresh; a page-not-found is a healthy-connection terminal
|
||||
* reply, so the session is left cached. A save is safe to retry — the server
|
||||
* promotes-not-duplicates an identical latest version — so the caller (and its
|
||||
* agent) may re-issue it without risking a duplicate heavy history row.
|
||||
*/
|
||||
export async function savePageVersionRealtime(
|
||||
pageId: PageId,
|
||||
collabToken: string,
|
||||
baseUrl: string,
|
||||
): Promise<SaveVersionResult> {
|
||||
return withPageLock(pageId, async () => {
|
||||
const session = await acquireCollabSession(pageId, collabToken, baseUrl);
|
||||
let reply: SaveVersionReply;
|
||||
try {
|
||||
reply = await session.sendStatelessAndAwait<SaveVersionReply>(
|
||||
JSON.stringify({ type: SAVE_VERSION_MESSAGE_TYPE }),
|
||||
matchSaveVersionReply,
|
||||
SAVE_VERSION_ACK_TIMEOUT_MS,
|
||||
);
|
||||
} catch (e) {
|
||||
// TRANSPORT failure (no reply within the timeout, or a disconnect): drop the
|
||||
// session so the next call reconnects fresh.
|
||||
session.destroy("save-version failed");
|
||||
throw e;
|
||||
}
|
||||
// A terminal reply arrived over a healthy connection — do NOT destroy the
|
||||
// session; interpret the outcome.
|
||||
if (reply.outcome === "skipped") {
|
||||
if (reply.reason === "page-not-found") {
|
||||
// A resolved pageId that the collab server no longer holds (deleted, or a
|
||||
// stale id). Surface it immediately and truthfully, not as a health error.
|
||||
throw new Error(
|
||||
`savePageVersion: page ${pageId} was not found on the collaboration ` +
|
||||
`server (it may have been deleted) — nothing was saved.`,
|
||||
);
|
||||
}
|
||||
// Reachable benign no-op (e.g. an empty page): a clean result, not a throw.
|
||||
return { saved: false, skipped: true, reason: reply.reason };
|
||||
}
|
||||
return {
|
||||
saved: true,
|
||||
historyId: reply.historyId,
|
||||
kind: reply.kind,
|
||||
alreadySaved: reply.alreadySaved,
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Replace the live content of a page over the collaboration websocket.
|
||||
* Accepts a ready ProseMirror JSON document; the caller controls whether
|
||||
|
||||
@@ -44,7 +44,7 @@ export const ROUTING_PROSE =
|
||||
"EDIT: fix wording/typos/numbers -> editPageText (find/replace inside blocks, no node id needed). Edit a block -> getNode(markdown) -> edit the markdown -> patchNode(markdown) (by attrs.id from getOutline; the markdown fragment may be several blocks — a 1->N section rewrite in one call, the first block keeps the id). Reach for patchNode's `node`-JSON only for fine attr/mark work; a table cell with spans/colors/fixed width -> the table tools (patchNode markdown refuses it). Add a block -> insertNode (markdown, before/after a block by attrs.id or by anchor text, or append; `node` for raw JSON or bare table structure). Remove a block -> deleteNode (by attrs.id). Tables -> tableGet / tableUpdateCell / tableInsertRow / tableDeleteRow (address by \"#<index>\" from getOutline; table nodes have no attrs.id). Images -> insertImage (add from a web URL) / replaceImage (swap an existing image). Draw.io diagrams -> PREFER the high-level semantic tools that hide coordinates/styles: drawioFromGraph (architecture/cloud/network diagrams — describe nodes/groups/edges by kind+icon, the server picks layout, colors and verified icons; hints layer/sameLayerAs/pinned and layout:full|incremental|none) and drawioFromMermaid (standard flowcharts — write Mermaid, get an editable diagram). For targeted tweaks of an existing diagram use drawioEditCells (id-based add/update/delete with cascade delete + baseHash lock). Raw mxGraph XML via drawioCreate/drawioUpdate is the escape-hatch for exotic/wireframe diagrams; drawioGet reads a diagram as mxGraph XML + a hash (pass it as baseHash to drawioUpdate/drawioEditCells for optimistic locking). Before authoring raw XML, drawioShapes (look up verified stencil style-strings so a shape name never renders as an empty box) and drawioGuide (on-demand authoring reference: skeleton/layout/containers/icons-aws/icons-azure), and pass layout:\"elk\" to drawioCreate/drawioUpdate to auto-place nodes. Footnotes -> insertFootnote. Bulk/structural rewrite -> updatePageJson (full ProseMirror replace) or updatePageMarkdown (full plain-Markdown body replace, re-imported — block ids regenerate); prefer the granular tools above to avoid resending the whole ~100KB+ document. Complex/scripted rewrite (multiple coordinated edits, renumbering) -> docmostTransform: 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 -> createPage (Markdown). Rename (title only) -> renamePage. Move -> movePage. Delete -> deletePage (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) -> copyPageContent. Sharing -> sharePage / unsharePage / listShares; sharePage makes the page PUBLICLY accessible — do it only when explicitly asked.\n" +
|
||||
"COMMENTS: createComment 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 -> createComment 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 -> listComments, updateComment, resolveComment (resolve/reopen, reversible — prefer over delete to close), deleteComment, checkNewComments.\n" +
|
||||
"HISTORY: review what changed -> diffPageVersions (a historyId vs current, or two versions). List saved versions -> listPageHistory. Undo a bad edit -> restorePageVersion (writes a past version back as current; itself revertible). Export a page to self-contained Docmost Markdown (with comment anchors) -> exportPageMarkdown.";
|
||||
"HISTORY: review what changed -> diffPageVersions (a historyId vs current, or two versions). List saved versions -> listPageHistory. Undo a bad edit -> restorePageVersion (writes a past version back as current; itself revertible). Pin the page's CURRENT content as a restorable named checkpoint when you finish a coherent editing pass -> savePageVersion (kind derived server-side as an agent version; an identical-to-last save is promoted/no-op'd, so a redundant call is harmless). Export a page to self-contained Docmost Markdown (with comment anchors) -> exportPageMarkdown.";
|
||||
|
||||
/**
|
||||
* Non-tool camelCase identifiers that legitimately appear in ROUTING_PROSE:
|
||||
@@ -218,6 +218,7 @@ const TOOL_FAMILY: Record<string, Family> = {
|
||||
diffPageVersions: "HISTORY",
|
||||
listPageHistory: "HISTORY",
|
||||
restorePageVersion: "HISTORY",
|
||||
savePageVersion: "HISTORY",
|
||||
exportPageMarkdown: "HISTORY",
|
||||
// importPageMarkdown is now inAppOnly (#411) — it is not registered on the
|
||||
// external MCP host, so it no longer appears in the generated inventory.
|
||||
|
||||
@@ -93,6 +93,7 @@ export type DocmostClientLike = Pick<
|
||||
| 'sharePage'
|
||||
| 'unsharePage'
|
||||
| 'restorePageVersion'
|
||||
| 'savePageVersion'
|
||||
| 'stashPage'
|
||||
| 'insertFootnote'
|
||||
| 'insertImage'
|
||||
@@ -743,6 +744,30 @@ export const SHARED_TOOL_SPECS = {
|
||||
client.restorePageVersion(historyId as string),
|
||||
},
|
||||
|
||||
savePageVersion: {
|
||||
mcpName: 'savePageVersion',
|
||||
inAppKey: 'savePageVersion',
|
||||
writeClass: 'write',
|
||||
description:
|
||||
'Save an intentional, NAMED version (kind=agent) of the page\'s CURRENT ' +
|
||||
'live content — a restorable checkpoint pinned into its history. Call it ' +
|
||||
'when you have FINISHED a coherent editing pass (not after every small ' +
|
||||
'edit), so the reader can see and roll back to the state you left. The ' +
|
||||
'version type is derived SERVER-SIDE from your signed agent identity (you ' +
|
||||
'cannot mislabel it); a save whose content is IDENTICAL to the last saved ' +
|
||||
'version is promoted/no-op\'d server-side, so a redundant call is harmless ' +
|
||||
'and never duplicates a version. Returns { saved:true, historyId, kind, ' +
|
||||
'alreadySaved } on success, or { saved:false, skipped:true, reason:\'empty\' } ' +
|
||||
'when the page had nothing to pin (e.g. it was empty).',
|
||||
tier: 'deferred',
|
||||
catalogLine:
|
||||
'savePageVersion — pin the page\'s current content as a named agent version (restorable checkpoint).',
|
||||
buildShape: (z) => ({
|
||||
pageId: z.string().min(1),
|
||||
}),
|
||||
execute: (client, { pageId }) => client.savePageVersion(pageId as string),
|
||||
},
|
||||
|
||||
// --- markdown round-trip ---
|
||||
|
||||
importPageMarkdown: {
|
||||
|
||||
@@ -0,0 +1,300 @@
|
||||
import { test, beforeEach, afterEach, mock } from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import { EventEmitter } from "node:events";
|
||||
|
||||
import {
|
||||
acquireCollabSession,
|
||||
destroyAllSessions,
|
||||
__setCollabProviderFactory,
|
||||
} from "../../build/lib/collab-session.js";
|
||||
import { savePageVersionRealtime } from "../../build/lib/collaboration.js";
|
||||
|
||||
// #370 — unit coverage for the MCP client transport of the explicit save-version.
|
||||
// There is no collaboration server in the test env, so a fake HocuspocusProvider
|
||||
// drives the stateless round-trip: it auto-completes the connect/sync handshake on
|
||||
// a microtask (like the real provider), records sends, and either auto-broadcasts a
|
||||
// configured reply (`opts.reply`) or lets the test emit stateless messages /
|
||||
// disconnects by hand (`_emitStateless` / `_disconnect`) to exercise the predicate
|
||||
// filter, the teardown-mid-wait path, and the timeout.
|
||||
class FakeProvider extends EventEmitter {
|
||||
static instances = [];
|
||||
|
||||
static reset() {
|
||||
FakeProvider.instances = [];
|
||||
}
|
||||
|
||||
static last() {
|
||||
return FakeProvider.instances[FakeProvider.instances.length - 1];
|
||||
}
|
||||
|
||||
constructor(config, opts = {}) {
|
||||
super();
|
||||
this.config = config;
|
||||
this.synced = false;
|
||||
this.unsyncedChanges = 0;
|
||||
this.destroyed = false;
|
||||
this.sent = [];
|
||||
this.opts = opts;
|
||||
FakeProvider.instances.push(this);
|
||||
// Real HocuspocusProvider fires onSynced asynchronously after the handshake;
|
||||
// a microtask reproduces that without depending on timers (so mock.timers on
|
||||
// setTimeout does not stall the connect).
|
||||
queueMicrotask(() => {
|
||||
if (this.destroyed) return;
|
||||
this.config.onConnect?.();
|
||||
this.synced = true;
|
||||
this.config.onSynced?.();
|
||||
});
|
||||
}
|
||||
|
||||
destroy() {
|
||||
this.destroyed = true;
|
||||
}
|
||||
|
||||
sendStateless(payload) {
|
||||
this.sent.push(payload);
|
||||
const reply = this.opts.reply;
|
||||
if (!reply) return; // silent → the caller times out, or the test drives it
|
||||
// Broadcast the configured server reply on a microtask, mirroring
|
||||
// document.broadcastStateless landing back on this provider's `stateless` event.
|
||||
queueMicrotask(() => {
|
||||
if (this.destroyed) return;
|
||||
this.emit("stateless", { payload: JSON.stringify(reply) });
|
||||
});
|
||||
}
|
||||
|
||||
// --- test drivers ---
|
||||
_emitStateless(obj) {
|
||||
this.emit("stateless", { payload: JSON.stringify(obj) });
|
||||
}
|
||||
_disconnect() {
|
||||
this.config.onDisconnect?.();
|
||||
}
|
||||
}
|
||||
|
||||
const ENV_KEYS = [
|
||||
"MCP_COLLAB_SESSION_IDLE_MS",
|
||||
"MCP_COLLAB_SESSION_MAX_AGE_MS",
|
||||
"MCP_COLLAB_SESSION_MAX_ENTRIES",
|
||||
"MCP_COLLAB_TOKEN_TTL_MS",
|
||||
];
|
||||
let savedEnv;
|
||||
|
||||
beforeEach(() => {
|
||||
savedEnv = {};
|
||||
for (const k of ENV_KEYS) savedEnv[k] = process.env[k];
|
||||
FakeProvider.reset();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
destroyAllSessions();
|
||||
__setCollabProviderFactory(null);
|
||||
mock.timers.reset();
|
||||
for (const k of ENV_KEYS) {
|
||||
if (savedEnv[k] === undefined) delete process.env[k];
|
||||
else process.env[k] = savedEnv[k];
|
||||
}
|
||||
});
|
||||
|
||||
const PAGE_A = "11111111-1111-4111-8111-111111111111";
|
||||
const PAGE_B = "22222222-2222-4222-8222-222222222222";
|
||||
const PAGE_C = "33333333-3333-4333-8333-333333333333";
|
||||
const PAGE_D = "44444444-4444-4444-8444-444444444444";
|
||||
const PAGE_E = "55555555-5555-4555-8555-555555555555";
|
||||
const PAGE_F = "66666666-6666-4666-8666-666666666666";
|
||||
|
||||
// --- happy paths (resolve with the server's terminal reply) -----------------
|
||||
|
||||
test("saved: resolves { saved:true, historyId, kind, alreadySaved }", async () => {
|
||||
__setCollabProviderFactory(
|
||||
(config) =>
|
||||
new FakeProvider(config, {
|
||||
reply: {
|
||||
type: "version.saved",
|
||||
historyId: "hist-42",
|
||||
kind: "agent",
|
||||
alreadySaved: false,
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
const result = await savePageVersionRealtime(PAGE_A, "collab-tok", "http://host/api");
|
||||
|
||||
assert.deepEqual(result, {
|
||||
saved: true,
|
||||
historyId: "hist-42",
|
||||
kind: "agent",
|
||||
alreadySaved: false,
|
||||
});
|
||||
// The client actually asked the server to save a version (the stateless send).
|
||||
assert.deepEqual(
|
||||
FakeProvider.last().sent.map((p) => JSON.parse(p)),
|
||||
[{ type: "save-version" }],
|
||||
);
|
||||
});
|
||||
|
||||
test("saved: surfaces alreadySaved=true (promote-not-dup / no-op save)", async () => {
|
||||
__setCollabProviderFactory(
|
||||
(config) =>
|
||||
new FakeProvider(config, {
|
||||
reply: {
|
||||
type: "version.saved",
|
||||
historyId: "hist-7",
|
||||
kind: "manual",
|
||||
alreadySaved: true,
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
const result = await savePageVersionRealtime(PAGE_B, "collab-tok", "http://host/api");
|
||||
|
||||
assert.deepEqual(result, {
|
||||
saved: true,
|
||||
historyId: "hist-7",
|
||||
kind: "manual",
|
||||
alreadySaved: true,
|
||||
});
|
||||
});
|
||||
|
||||
// --- F2: terminal SKIP replies (no timeout, no false "unreachable") ---------
|
||||
|
||||
test("F2 skip: an empty-page reply resolves to a clean { saved:false, skipped:true, reason:'empty' } — no timeout", async () => {
|
||||
__setCollabProviderFactory(
|
||||
(config) =>
|
||||
new FakeProvider(config, {
|
||||
reply: { type: "version.skipped", reason: "empty" },
|
||||
}),
|
||||
);
|
||||
|
||||
const result = await savePageVersionRealtime(PAGE_C, "collab-tok", "http://host/api");
|
||||
|
||||
assert.deepEqual(result, { saved: false, skipped: true, reason: "empty" });
|
||||
});
|
||||
|
||||
test("F2 skip: a page-not-found reply throws an immediate, truthful error (not the health-timeout path)", async () => {
|
||||
__setCollabProviderFactory(
|
||||
(config) =>
|
||||
new FakeProvider(config, {
|
||||
reply: { type: "version.skipped", reason: "page-not-found" },
|
||||
}),
|
||||
);
|
||||
|
||||
await assert.rejects(
|
||||
savePageVersionRealtime(PAGE_D, "collab-tok", "http://host/api"),
|
||||
/was not found on the collaboration server/,
|
||||
);
|
||||
});
|
||||
|
||||
// --- F1 (a): predicate filters unrelated messages, resolves on the real reply ---
|
||||
|
||||
test("F1(a) unrelated-then-real: an unrelated stateless message does NOT resolve; the later version.saved does", async () => {
|
||||
__setCollabProviderFactory((config) => new FakeProvider(config, {})); // manual drive
|
||||
|
||||
const p = savePageVersionRealtime(PAGE_E, "collab-tok", "http://host/api");
|
||||
// Let acquire's handshake + the stateless send complete so the listener is armed.
|
||||
await new Promise((r) => setImmediate(r));
|
||||
const provider = FakeProvider.last();
|
||||
|
||||
// Noise first: an unrelated stateless type (e.g. another feature's broadcast).
|
||||
provider._emitStateless({ type: "something-else", historyId: "NOPE" });
|
||||
// The wait must still be pending — a microtask turn would have resolved it if the
|
||||
// predicate wrongly matched.
|
||||
let settledEarly = false;
|
||||
p.then(() => (settledEarly = true)).catch(() => (settledEarly = true));
|
||||
await new Promise((r) => setImmediate(r));
|
||||
assert.equal(settledEarly, false, "resolved on an unrelated stateless message");
|
||||
|
||||
// Now the real reply arrives → it resolves with that payload.
|
||||
provider._emitStateless({
|
||||
type: "version.saved",
|
||||
historyId: "h-real",
|
||||
kind: "agent",
|
||||
alreadySaved: false,
|
||||
});
|
||||
const result = await p;
|
||||
assert.deepEqual(result, {
|
||||
saved: true,
|
||||
historyId: "h-real",
|
||||
kind: "agent",
|
||||
alreadySaved: false,
|
||||
});
|
||||
});
|
||||
|
||||
// --- F1 (b): teardown mid-wait rejects AND removes the stateless listener -----
|
||||
|
||||
test("F1(b) teardown mid-wait: a disconnect rejects the wait with the connection-loss error AND removes the stateless listener (no leak)", async () => {
|
||||
__setCollabProviderFactory((config) => new FakeProvider(config, {})); // silent
|
||||
|
||||
const session = await acquireCollabSession(PAGE_F, "collab-tok", "http://host/api");
|
||||
const provider = FakeProvider.last();
|
||||
|
||||
const p = session.sendStatelessAndAwait(
|
||||
JSON.stringify({ type: "save-version" }),
|
||||
(m) => (m && m.type === "version.saved" ? m : undefined),
|
||||
20000,
|
||||
);
|
||||
// Swallow the eventual rejection so it does not race the assertions below.
|
||||
p.catch(() => {});
|
||||
await new Promise((r) => setImmediate(r));
|
||||
|
||||
// The wait armed exactly one stateless listener.
|
||||
assert.equal(
|
||||
provider.listenerCount("stateless"),
|
||||
1,
|
||||
"the wait must register one stateless listener",
|
||||
);
|
||||
|
||||
// A disconnect tears the session down mid-wait.
|
||||
provider._disconnect();
|
||||
|
||||
await assert.rejects(p, /connection closed before the update was persisted/);
|
||||
// The listener was removed on teardown — dropping provider.off() reds this.
|
||||
assert.equal(
|
||||
provider.listenerCount("stateless"),
|
||||
0,
|
||||
"the stateless listener leaked after teardown",
|
||||
);
|
||||
});
|
||||
|
||||
// --- F1 (c): concurrent-in-flight guard fails fast, no cross-wiring -----------
|
||||
|
||||
test("F1(c) concurrent guard: a second sendStatelessAndAwait while one is in flight fails fast", async () => {
|
||||
__setCollabProviderFactory((config) => new FakeProvider(config, {})); // silent
|
||||
|
||||
const session = await acquireCollabSession(PAGE_A, "collab-tok", "http://host/api");
|
||||
|
||||
const first = session.sendStatelessAndAwait(
|
||||
JSON.stringify({ type: "save-version" }),
|
||||
(m) => (m && m.type === "version.saved" ? m : undefined),
|
||||
20000,
|
||||
);
|
||||
first.catch(() => {}); // it stays pending until afterEach destroys the session
|
||||
|
||||
await assert.rejects(
|
||||
session.sendStatelessAndAwait(
|
||||
JSON.stringify({ type: "save-version" }),
|
||||
(m) => (m && m.type === "version.saved" ? m : undefined),
|
||||
20000,
|
||||
),
|
||||
/already in-flight/,
|
||||
);
|
||||
});
|
||||
|
||||
// --- timeout: NO reply at all is the genuine "server unreachable" path --------
|
||||
|
||||
test("timeout: rejects on a bounded timeout when the server never replies", async () => {
|
||||
__setCollabProviderFactory((config) => new FakeProvider(config, {})); // silent
|
||||
|
||||
mock.timers.enable({ apis: ["setTimeout"] });
|
||||
const p = savePageVersionRealtime(PAGE_B, "collab-tok", "http://host/api");
|
||||
// Swallow the eventual rejection so an unhandled-rejection does not race the tick.
|
||||
p.catch(() => {});
|
||||
|
||||
// Flush the (microtask-driven) connect handshake + the stateless send so the
|
||||
// ack timer is armed before we advance the clock.
|
||||
await new Promise((r) => setImmediate(r));
|
||||
// Advance well past the 20s ack window; the connect timer (25s) does not fire.
|
||||
mock.timers.tick(20000 + 100);
|
||||
|
||||
await assert.rejects(p, /Timeout waiting for a stateless reply/);
|
||||
});
|
||||
Reference in New Issue
Block a user