From fd296e455754fdad9e1ad7e5ea60004d4c588d63 Mon Sep 17 00:00:00 2001 From: Andrew Mikofalvy <5668128+amikofalvy@users.noreply.github.com> Date: Tue, 21 Jul 2026 16:49:03 -0700 Subject: [PATCH] feat(server): agent-write-batch endpoint (#2813) * feat(server): agent-write-batch endpoint for bulk document writes * chore(md-audit): regenerate anchors catalog and audit for batch tests * fix(server): keep canonical presence shape in batch handler * test(server): track batch handler in conflict-gate coverage GitOrigin-RevId: 70c5eab7aa9690cf1c18db6a6f285854a3882c5b --- .changeset/agent-write-batch-endpoint.md | 5 + .../integration/agent-write-batch.test.ts | 267 ++++++++++++ .../agent-write-batch.test.ts | 163 ++++++++ .../attribution-sweep-coverage.test.ts | 5 + .../conflict-gate-coverage.test.ts | 5 + packages/core/src/index.ts | 11 + packages/core/src/schemas/api/_shared.ts | 8 + packages/core/src/schemas/api/agent-write.ts | 114 +++++- packages/server/src/api-extension.ts | 380 +++++++++++++++++- 9 files changed, 953 insertions(+), 5 deletions(-) create mode 100644 .changeset/agent-write-batch-endpoint.md create mode 100644 packages/app/tests/integration/agent-write-batch.test.ts create mode 100644 packages/app/tests/integration/api-error-envelope/agent-write-batch.test.ts diff --git a/.changeset/agent-write-batch-endpoint.md b/.changeset/agent-write-batch-endpoint.md new file mode 100644 index 000000000..3ad36d9e7 --- /dev/null +++ b/.changeset/agent-write-batch-endpoint.md @@ -0,0 +1,5 @@ +--- +"@inkeep/open-knowledge": patch +--- + +Add `POST /api/agent-write-batch`: apply up to 100 document writes in one HTTP call. Each entry uses the same write semantics as `POST /api/agent-write-md` (create or update, append/prepend/replace, per-entry summary, `.mdx` extension on create) and every entry is attributed to the calling agent. Outcomes are per-entry — a reserved doc name, merge conflict, or disk failure fails only that entry while its siblings land — and the response reports per-entry results with broken-link validation that resolves links between documents written in the same batch. The batch amortizes what N single calls pay N times: one round-trip, one presence/focus update, one admitted-doc-set scan for link validation, and one coalesced shadow-repo commit. The per-document mermaid render validation and lint advisories stay on the single-write endpoint. diff --git a/packages/app/tests/integration/agent-write-batch.test.ts b/packages/app/tests/integration/agent-write-batch.test.ts new file mode 100644 index 000000000..869242325 --- /dev/null +++ b/packages/app/tests/integration/agent-write-batch.test.ts @@ -0,0 +1,267 @@ +/** + * `POST /api/agent-write-batch` behavior suite. + * + * Pins the batch contract end-to-end against a real server: + * 1. Mixed create/update batches land per-entry (array order, duplicate + * docNames compose like sequential writes) with every write attributed + * to the batch agent, and the batch never forces a shadow commit of its + * own — the L2 debounce coalesces the whole batch (plus adjacent + * single-call writes) into ONE commit, drained by `/api/history`. + * 2. Reserved (system/config) doc names reject per entry; siblings land. + * 3. Broken-link validation sees sibling batch docs regardless of entry + * order (intra-batch links resolve) and still reports genuinely dead + * links. + * 4. Per-entry summaries echo (with the shared truncation shape). + * 5. Undo after a batch write flows through the normal per-session path. + * 6. The request-id envelope conventions hold for the new route. + */ + +import { execFileSync } from 'node:child_process'; +import { readFileSync } from 'node:fs'; +import { join } from 'node:path'; +import { AgentWriteBatchSuccessSchema } from '@inkeep/open-knowledge-core'; +import { resolveShadowDir } from '@inkeep/open-knowledge-core/shadow-repo-layout'; +import { afterEach, describe, expect, test } from 'vitest'; +import { createTestServer, getServerState, type TestServer } from './test-harness'; + +interface BatchResponseBody { + timestamp: string; + results: Array<{ + status: 'written' | 'error'; + docName: string; + summary?: { value: string; truncatedFrom?: number }; + brokenLinks?: Array<{ href: string; resolvedTo: string | null; reason: string }>; + error?: { type: string; title: string; detail?: string }; + }>; + written: number; + failed: number; +} + +interface TimelineResponse { + entries: Array<{ + sha: string; + type: string; + message: string; + contributors: Array<{ id: string; name?: string; docs?: string[] }>; + }>; +} + +let server: TestServer | undefined; + +afterEach(async () => { + await server?.cleanup(); + server = undefined; +}); + +async function postBatch( + port: number, + body: Record, + headers: Record = {}, +): Promise { + return fetch(`http://127.0.0.1:${port}/api/agent-write-batch`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', ...headers }, + body: JSON.stringify(body), + }); +} + +function listWipRefs(contentDir: string): string[] { + const shadowDir = resolveShadowDir(contentDir); + const raw = execFileSync('git', ['for-each-ref', '--format=%(refname)', 'refs/wip/'], { + env: { ...process.env, GIT_DIR: shadowDir }, + encoding: 'utf-8', + }); + return raw.trim().split('\n').filter(Boolean); +} + +function countCommits(contentDir: string, ref: string): number { + const shadowDir = resolveShadowDir(contentDir); + const raw = execFileSync('git', ['rev-list', '--count', ref], { + env: { ...process.env, GIT_DIR: shadowDir }, + encoding: 'utf-8', + }); + return Number(raw.trim()); +} + +describe('agent-write-batch', () => { + test('mixed create/update batch lands per entry, attributed, with one coalesced shadow commit', async () => { + server = await createTestServer({ gitEnabled: true, commitDebounceMs: 600_000 }); + const agentId = 'batch-writer'; + + const res = await postBatch(server.port, { + agentId, + agentName: 'Batch Writer', + docs: [ + { docName: 'batch/alpha', markdown: '# Alpha\n', position: 'replace' }, + { docName: 'batch/beta', markdown: '# Beta\n', position: 'replace' }, + // Duplicate docName: applies after the first alpha entry, in order. + { docName: 'batch/alpha', markdown: 'Second paragraph.\n', position: 'append' }, + ], + }); + expect(res.status).toBe(200); + const body = (await res.json()) as BatchResponseBody; + expect(AgentWriteBatchSuccessSchema.safeParse(body).success).toBe(true); + expect(body.written).toBe(3); + expect(body.failed).toBe(0); + expect(body.results.map((r) => r.status)).toEqual(['written', 'written', 'written']); + expect(body.results.map((r) => r.docName)).toEqual([ + 'batch/alpha', + 'batch/beta', + 'batch/alpha', + ]); + + // Per-doc L1 stores were awaited before the response: disk truth now. + const alpha = readFileSync(join(server.contentDir, 'batch/alpha.md'), 'utf-8'); + expect(alpha).toContain('# Alpha'); + expect(alpha).toContain('Second paragraph.'); + expect(readFileSync(join(server.contentDir, 'batch/beta.md'), 'utf-8')).toContain('# Beta'); + + // The batch armed the L2 debounce but never forced a commit of its own. + expect(listWipRefs(server.contentDir)).toEqual([]); + + // /api/history drains the pending commit; the whole batch coalesced into + // ONE shadow commit on the batch writer's wip ref, attributed to it. + const hist = (await ( + await fetch(`${server.baseUrl}/api/history?docName=${encodeURIComponent('batch/alpha')}`) + ).json()) as TimelineResponse; + expect(hist.entries.length).toBeGreaterThanOrEqual(1); + const contributorIds = new Set(hist.entries.flatMap((e) => e.contributors.map((c) => c.id))); + expect([...contributorIds]).toEqual([`agent-${agentId}`]); + const batchContributor = hist.entries[0].contributors.find((c) => c.id === `agent-${agentId}`); + expect(batchContributor?.docs).toContain('batch/alpha'); + expect(batchContributor?.docs).toContain('batch/beta'); + + const refs = listWipRefs(server.contentDir); + expect(refs.length).toBe(1); + expect(countCommits(server.contentDir, refs[0])).toBe(1); + }, 30_000); + + test('system/config docs reject per entry; siblings land unaffected', async () => { + server = await createTestServer(); + + const res = await postBatch(server.port, { + agentId: 'batch-gate', + docs: [ + { docName: 'batch-gate/ok-doc', markdown: '# Fine\n', position: 'replace' }, + { docName: '__system__', markdown: '# Nope\n', position: 'replace' }, + { docName: '__config__/project', markdown: '# Nope\n', position: 'replace' }, + { docName: 'batch-gate/also-ok', markdown: '# Also fine\n', position: 'replace' }, + ], + }); + expect(res.status).toBe(200); + const body = (await res.json()) as BatchResponseBody; + expect(body.written).toBe(2); + expect(body.failed).toBe(2); + expect(body.results[0].status).toBe('written'); + expect(body.results[1].status).toBe('error'); + expect(body.results[1].error?.type).toBe('urn:ok:error:reserved-doc-name'); + expect(body.results[2].status).toBe('error'); + expect(body.results[2].error?.type).toBe('urn:ok:error:reserved-doc-name'); + expect(body.results[3].status).toBe('written'); + + expect(readFileSync(join(server.contentDir, 'batch-gate/ok-doc.md'), 'utf-8')).toContain( + '# Fine', + ); + expect(readFileSync(join(server.contentDir, 'batch-gate/also-ok.md'), 'utf-8')).toContain( + '# Also fine', + ); + }, 30_000); + + test('broken-link validation admits sibling batch docs and reports dead links', async () => { + server = await createTestServer(); + + const res = await postBatch(server.port, { + agentId: 'batch-links', + docs: [ + // Links to a sibling written LATER in the same batch — must resolve. + { + docName: 'batch-links/a', + markdown: '# A\n\n[to b](./b.md)\n', + position: 'replace', + }, + { docName: 'batch-links/b', markdown: '# B\n', position: 'replace' }, + { + docName: 'batch-links/c', + markdown: '# C\n\n[dead](./no-such-doc.md)\n', + position: 'replace', + }, + ], + }); + expect(res.status).toBe(200); + const body = (await res.json()) as BatchResponseBody; + expect(body.results[0].status).toBe('written'); + expect(body.results[0].brokenLinks).toEqual([]); + expect(body.results[1].brokenLinks).toEqual([]); + expect(body.results[2].status).toBe('written'); + expect(body.results[2].brokenLinks?.length).toBe(1); + expect(body.results[2].brokenLinks?.[0].reason).toBe('no-such-doc'); + }, 30_000); + + test('per-entry summaries echo with the shared truncation shape', async () => { + server = await createTestServer(); + + const longSummary = 'x'.repeat(120); + const res = await postBatch(server.port, { + agentId: 'batch-summaries', + docs: [ + { + docName: 'batch-summaries/one', + markdown: '# One\n', + position: 'replace', + summary: 'Wrote one', + }, + { + docName: 'batch-summaries/two', + markdown: '# Two\n', + position: 'replace', + summary: longSummary, + }, + { docName: 'batch-summaries/three', markdown: '# Three\n', position: 'replace' }, + ], + }); + expect(res.status).toBe(200); + const body = (await res.json()) as BatchResponseBody; + expect(body.results[0].summary?.value).toBe('Wrote one'); + expect(body.results[1].summary?.value).toBe(`${'x'.repeat(79)}…`); + expect(body.results[1].summary?.truncatedFrom).toBe(120); + expect(body.results[2].summary).toBeUndefined(); + }, 30_000); + + test('undo after a batch write flows through the normal per-session path', async () => { + server = await createTestServer(); + const agentId = 'batch-undo'; + const docName = 'batch-undo/target'; + + const writeRes = await postBatch(server.port, { + agentId, + docs: [{ docName, markdown: '# Undo target\n', position: 'replace' }], + }); + expect(writeRes.status).toBe(200); + expect(getServerState(server, docName)?.ytext.toString()).toContain('# Undo target'); + + const undoRes = await fetch(`${server.baseUrl}/api/agent-undo`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ docName, connectionId: `agent-${agentId}`, scope: 'session' }), + }); + expect(undoRes.status).toBe(200); + const undoBody = (await undoRes.json()) as { undone: boolean }; + expect(undoBody.undone).toBe(true); + expect(getServerState(server, docName)?.ytext.toString()).not.toContain('# Undo target'); + }, 30_000); + + test('response echoes a well-formed incoming x-request-id', async () => { + server = await createTestServer(); + + const res = await postBatch( + server.port, + { + agentId: 'batch-reqid', + docs: [{ docName: 'batch-reqid/doc', markdown: '# Hi\n', position: 'replace' }], + }, + { 'x-request-id': 'batch-req-1' }, + ); + expect(res.status).toBe(200); + expect(res.headers.get('x-request-id')).toBe('batch-req-1'); + }, 30_000); +}); diff --git a/packages/app/tests/integration/api-error-envelope/agent-write-batch.test.ts b/packages/app/tests/integration/api-error-envelope/agent-write-batch.test.ts new file mode 100644 index 000000000..e27db0bb2 --- /dev/null +++ b/packages/app/tests/integration/api-error-envelope/agent-write-batch.test.ts @@ -0,0 +1,163 @@ +/** + * Per-handler narrow-integration smoke test for `handleAgentWriteBatch`. + * + * Asserts the canonical RFC 9457 wire shape: + * - happy path: status 200, `Content-Type: application/json`, body parses + * against `AgentWriteBatchSuccessSchema` (timestamp + per-entry results + + * written/failed counts), no `ok: true` discriminator. + * - missing docs / empty docs / over-cap docs / malformed per-entry docName + * → `urn:ok:error:invalid-request` (pre-identity body-shape rejection + * from `withValidation`; the whole batch rejects at the schema boundary). + * - reserved docname → per-entry error INSIDE a 200 body (semantic, + * post-identity, partial success by design). + * - method-not-allowed on GET. + */ + +import { + AGENT_WRITE_BATCH_MAX_DOCS, + AgentWriteBatchSuccessSchema, + ProblemDetailsSchema, +} from '@inkeep/open-knowledge-core'; +import { afterAll, beforeAll, describe, expect, test } from 'vitest'; +import { HARNESS_BOOT_TIMEOUT_MS } from '../harness-boot-timeout'; +import { createTestServer, type TestServer } from '../test-harness'; + +let server: TestServer; + +beforeAll(async () => { + server = await createTestServer(); +}, HARNESS_BOOT_TIMEOUT_MS); + +afterAll(async () => { + await server.cleanup(); +}); + +const UUID_RE = /^urn:uuid:[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; + +async function postBatch(body: Record): Promise { + return fetch(`http://127.0.0.1:${server.port}/api/agent-write-batch`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(body), + }); +} + +describe('agent-write-batch envelope (RFC 9457)', () => { + test('happy path emits flat success body with application/json', async () => { + const prefix = `batch-env-${crypto.randomUUID().slice(0, 8)}`; + const res = await postBatch({ + docs: [ + { docName: `${prefix}/one`, markdown: '# One\n', position: 'replace' }, + { docName: `${prefix}/two`, markdown: '# Two\n', position: 'replace' }, + ], + }); + expect(res.status).toBe(200); + expect(res.headers.get('content-type')).toBe('application/json'); + + const body = await res.json(); + const parsed = AgentWriteBatchSuccessSchema.safeParse(body); + expect(parsed.success).toBe(true); + if (parsed.success) { + expect(parsed.data.timestamp.length).toBeGreaterThan(0); + expect(parsed.data.results.length).toBe(2); + expect(parsed.data.written).toBe(2); + expect(parsed.data.failed).toBe(0); + } + expect((body as Record).ok).toBeUndefined(); + }); + + test('missing docs emits urn:ok:error:invalid-request', async () => { + const res = await postBatch({}); + expect(res.status).toBe(400); + expect(res.headers.get('content-type')).toBe('application/problem+json'); + + const body = await res.json(); + const parsed = ProblemDetailsSchema.safeParse(body); + expect(parsed.success).toBe(true); + if (parsed.success) { + expect(parsed.data.type).toBe('urn:ok:error:invalid-request'); + expect(parsed.data.status).toBe(400); + expect(parsed.data.instance).toBeDefined(); + if (parsed.data.instance) expect(parsed.data.instance).toMatch(UUID_RE); + } + }); + + test('empty docs array emits urn:ok:error:invalid-request', async () => { + const res = await postBatch({ docs: [] }); + expect(res.status).toBe(400); + const body = await res.json(); + const parsed = ProblemDetailsSchema.safeParse(body); + expect(parsed.success).toBe(true); + if (parsed.success) expect(parsed.data.type).toBe('urn:ok:error:invalid-request'); + }); + + test('over-cap batch rejects wholesale with urn:ok:error:invalid-request', async () => { + const docs = Array.from({ length: AGENT_WRITE_BATCH_MAX_DOCS + 1 }, (_, i) => ({ + docName: `batch-cap/doc-${i}`, + markdown: '# Over cap\n', + position: 'replace', + })); + const res = await postBatch({ docs }); + expect(res.status).toBe(400); + expect(res.headers.get('content-type')).toBe('application/problem+json'); + const body = await res.json(); + const parsed = ProblemDetailsSchema.safeParse(body); + expect(parsed.success).toBe(true); + if (parsed.success) { + expect(parsed.data.type).toBe('urn:ok:error:invalid-request'); + expect(parsed.data.status).toBe(400); + } + }); + + test('malformed per-entry docName rejects the whole batch as invalid-request', async () => { + const res = await postBatch({ + docs: [ + { docName: 'batch-shape/fine', markdown: '# Fine\n', position: 'replace' }, + { docName: '../escape', markdown: '# Nope\n', position: 'replace' }, + ], + }); + expect(res.status).toBe(400); + const body = await res.json(); + const parsed = ProblemDetailsSchema.safeParse(body); + expect(parsed.success).toBe(true); + if (parsed.success) expect(parsed.data.type).toBe('urn:ok:error:invalid-request'); + }); + + test('reserved docname is a per-entry error inside a 200 body', async () => { + const prefix = `batch-env-reserved-${crypto.randomUUID().slice(0, 8)}`; + const res = await postBatch({ + docs: [ + { docName: `${prefix}/fine`, markdown: '# Fine\n', position: 'replace' }, + { docName: '__system__', markdown: '# Nope\n', position: 'replace' }, + ], + }); + expect(res.status).toBe(200); + const body = await res.json(); + const parsed = AgentWriteBatchSuccessSchema.safeParse(body); + expect(parsed.success).toBe(true); + if (parsed.success) { + expect(parsed.data.written).toBe(1); + expect(parsed.data.failed).toBe(1); + const failedEntry = parsed.data.results[1]; + expect(failedEntry.status).toBe('error'); + if (failedEntry.status === 'error') { + expect(failedEntry.error.type).toBe('urn:ok:error:reserved-doc-name'); + } + } + }); + + test('method-not-allowed on GET emits problem+json with Allow: POST', async () => { + const res = await fetch(`http://127.0.0.1:${server.port}/api/agent-write-batch`, { + method: 'GET', + }); + expect(res.status).toBe(405); + expect(res.headers.get('allow')).toBe('POST'); + + const body = await res.json(); + const parsed = ProblemDetailsSchema.safeParse(body); + expect(parsed.success).toBe(true); + if (parsed.success) { + expect(parsed.data.type).toBe('urn:ok:error:method-not-allowed'); + } + }); +}); diff --git a/packages/app/tests/integration/attribution-sweep-coverage.test.ts b/packages/app/tests/integration/attribution-sweep-coverage.test.ts index b7230e229..83087f796 100644 --- a/packages/app/tests/integration/attribution-sweep-coverage.test.ts +++ b/packages/app/tests/integration/attribution-sweep-coverage.test.ts @@ -34,6 +34,11 @@ const actorHelperSource = readFileSync(ACTOR_HELPER_PATH, 'utf8'); const REQUIRED_HANDLERS = [ 'handleAgentWrite', 'handleAgentWriteMd', + // Batch sibling of handleAgentWriteMd — one identity extraction attributes + // every entry in the batch; per-entry failures are response-body results, + // not wire-level errorResponse emits, so the ordering check sees only the + // post-identity catch-all. + 'handleAgentWriteBatch', 'handleAgentPatch', // Per-key frontmatter mutation via JSON Merge Patch. Writes through the same // session-frozen origin as the other agent-write handlers; attribution is diff --git a/packages/app/tests/integration/conflict-gate-coverage.test.ts b/packages/app/tests/integration/conflict-gate-coverage.test.ts index 1fad3d63e..d4f03c500 100644 --- a/packages/app/tests/integration/conflict-gate-coverage.test.ts +++ b/packages/app/tests/integration/conflict-gate-coverage.test.ts @@ -38,6 +38,11 @@ const source = readFileSync(API_EXT_PATH, 'utf8'); const REQUIRED_HANDLERS = [ 'handleAgentWrite', 'handleAgentWriteMd', + // Batch sibling of handleAgentWriteMd — every entry routes through the + // gated applyAgentMarkdownWrite spine; a per-entry DocInConflictError is + // translated to that entry's result (with the doc-in-conflict-write-refused + // event re-emitted) instead of a wire-level 409, so siblings keep landing. + 'handleAgentWriteBatch', 'handleAgentPatch', 'handleAgentUndo', 'handleRollback', diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 3b9fee46c..0eace3e83 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -562,6 +562,7 @@ export { type AdvisoryWarning, AdvisoryWarningSchema, AdvisoryWarningsSchema, + AGENT_WRITE_BATCH_MAX_DOCS, type AgentActivitySuccess, AgentActivitySuccessSchema, type AgentBurstDiffSuccess, @@ -580,6 +581,14 @@ export { AgentUndoRequestSchema, type AgentUndoSuccess, AgentUndoSuccessSchema, + type AgentWriteBatchEntry, + AgentWriteBatchEntrySchema, + type AgentWriteBatchRequest, + AgentWriteBatchRequestSchema, + type AgentWriteBatchResult, + AgentWriteBatchResultSchema, + type AgentWriteBatchSuccess, + AgentWriteBatchSuccessSchema, type AgentWriteMdRequest, AgentWriteMdRequestSchema, type AgentWriteMdSuccess, @@ -597,6 +606,8 @@ export { BacklinkEntrySchema, type BacklinksSuccess, BacklinksSuccessSchema, + type BatchEntryError, + BatchEntryErrorSchema, BROKEN_LINK_REASONS, type BranchInfoResponse, BranchInfoResponseSchema, diff --git a/packages/core/src/schemas/api/_shared.ts b/packages/core/src/schemas/api/_shared.ts index f8ed3ac1d..5cfb7af8a 100644 --- a/packages/core/src/schemas/api/_shared.ts +++ b/packages/core/src/schemas/api/_shared.ts @@ -45,6 +45,14 @@ function checkDocName(value: string, ctx: z.RefinementCtx): void { export const safeDocNameField = z.string().superRefine(checkDocName).optional(); +/** + * Required variant of `safeDocNameField` — same structural docName contract, + * but the key must be present. Batch write entries use this: a per-entry + * docName cannot fall back to a handler-side default the way the single-write + * handlers' optional field can. + */ +export const requiredSafeDocNameField = z.string().superRefine(checkDocName); + /** * Identity fields shared by every mutating handler. All optional — * `extractAgentIdentity` in `api-extension.ts` carries the default-agent diff --git a/packages/core/src/schemas/api/agent-write.ts b/packages/core/src/schemas/api/agent-write.ts index 602275734..c99738746 100644 --- a/packages/core/src/schemas/api/agent-write.ts +++ b/packages/core/src/schemas/api/agent-write.ts @@ -16,7 +16,13 @@ import { z } from 'zod'; import { SUPPORTED_DOC_EXTENSIONS } from '../../constants/doc-extensions.ts'; import { FRONTMATTER_TYPES, FrontmatterValueSchema } from '../../frontmatter/schema.ts'; -import { agentIdentityFields, safeDocNameField, summaryField } from './_shared.ts'; +import { ProblemTypeSchema } from './_envelope.ts'; +import { + agentIdentityFields, + requiredSafeDocNameField, + safeDocNameField, + summaryField, +} from './_shared.ts'; /** * Request body for `POST /api/agent-write`. Free-text content append (the @@ -437,6 +443,112 @@ export const AgentUndoSuccessSchema = z .loose() satisfies StandardSchemaV1; export type AgentUndoSuccess = z.infer; +/** + * Per-call ceiling on `POST /api/agent-write-batch` entries. Sized against + * three independent bounds: the 1 MB request-body cap (100 docs leaves ~10 KB + * of markdown each), the 256-session `AgentSessionManager` cap (a batch must + * not be able to exhaust it in one call), and bounded per-request latency + * (each entry pays a CRDT transact + an awaited per-doc disk store). A larger + * import splits into multiple calls. + */ +export const AGENT_WRITE_BATCH_MAX_DOCS = 100; + +/** + * One entry of a `POST /api/agent-write-batch` request. Mirrors the + * `AgentWriteMdRequestSchema` per-doc fields exactly (same write semantics per + * entry); identity fields live once at the batch level instead. + */ +export const AgentWriteBatchEntrySchema = z + .object({ + docName: requiredSafeDocNameField, + markdown: z.string(), + position: z.enum(['append', 'prepend', 'replace']).optional(), + extension: z.enum(SUPPORTED_DOC_EXTENSIONS).optional(), + summary: summaryField, + }) + .loose() satisfies StandardSchemaV1; +export type AgentWriteBatchEntry = z.infer; + +/** + * Request body for `POST /api/agent-write-batch`. One identity for the whole + * batch (every entry is attributed to the same agent); entries apply in array + * order, so duplicate docNames are legal and behave like sequential writes to + * that doc. Exceeding `AGENT_WRITE_BATCH_MAX_DOCS` rejects the whole request + * at the schema boundary (400 `urn:ok:error:invalid-request`). + */ +export const AgentWriteBatchRequestSchema = z + .object({ + docs: z.array(AgentWriteBatchEntrySchema).min(1).max(AGENT_WRITE_BATCH_MAX_DOCS), + ...agentIdentityFields, + }) + .loose() satisfies StandardSchemaV1; +export type AgentWriteBatchRequest = z.infer; + +/** + * Per-entry error payload inside a batch response. Reuses the closed + * `ProblemTypeSchema` URN vocabulary so SDK consumers branch on the same + * tokens as the single-write endpoints, but deliberately omits the RFC 9457 + * `status`/`instance` fields — the entry is not an HTTP response; the batch + * response's HTTP layer is the 200 envelope around all entries. + */ +export const BatchEntryErrorSchema = z + .object({ + type: ProblemTypeSchema, + title: z.string(), + detail: z.string().optional(), + }) + .loose() satisfies StandardSchemaV1; +export type BatchEntryError = z.infer; + +/** + * Per-entry outcome of a batch write, positionally aligned with the request's + * `docs` array (`results[i]` answers `docs[i]`). `docName` echoes the entry's + * alias-resolved canonical docName. Outcomes are independent — a failed entry + * never rolls back or blocks its siblings. + * + * `written` entries carry the same advisory channels as the single-write + * success body, minus the deliberately-omitted per-doc passes (mermaid render + * validation, lint, orphan hints) — those stay on `POST /api/agent-write-md`, + * whose per-write cost is exactly what a batch amortizes away. + */ +export const AgentWriteBatchResultSchema = z.discriminatedUnion('status', [ + z + .object({ + status: z.literal('written'), + docName: z.string().min(1), + summary: SummaryResponseFieldSchema.optional(), + warnings: AdvisoryWarningsSchema.optional(), + /** Write-time outbound-link validation. Always present, `[]` when all links resolve. */ + brokenLinks: BrokenLinksSchema, + }) + .loose(), + z + .object({ + status: z.literal('error'), + docName: z.string().min(1), + error: BatchEntryErrorSchema, + }) + .loose(), +]); +export type AgentWriteBatchResult = z.infer; + +/** + * Success body for `POST /api/agent-write-batch`. HTTP 200 whenever the batch + * request itself was well-formed — per-entry failures (reserved doc name, + * conflict, disk failure) report inside `results`, so a partially-failed + * batch is still a 200 with a non-zero `failed` count. Wire-level errors are + * reserved for whole-request faults (schema rejection, payload cap, method). + */ +export const AgentWriteBatchSuccessSchema = z + .object({ + timestamp: z.string().min(1), + results: z.array(AgentWriteBatchResultSchema), + written: z.number().int().nonnegative(), + failed: z.number().int().nonnegative(), + }) + .loose() satisfies StandardSchemaV1; +export type AgentWriteBatchSuccess = z.infer; + /** * Request body for `POST /api/frontmatter-patch`. Merge-patch semantics apply * at the TOP-LEVEL key only: a top-level key mapped to `null` deletes that diff --git a/packages/server/src/api-extension.ts b/packages/server/src/api-extension.ts index acb33394f..4e9463ed8 100644 --- a/packages/server/src/api-extension.ts +++ b/packages/server/src/api-extension.ts @@ -32,6 +32,7 @@ import { pipeline } from 'node:stream/promises'; import { setTimeout as wait } from 'node:timers/promises'; import type { Document, Extension, Hocuspocus } from '@hocuspocus/server'; import { + type AdvisoryWarning, AGENT_ICON_COLORS, AgentActivitySuccessSchema, AgentBurstDiffSuccessSchema, @@ -39,6 +40,8 @@ import { AgentPatchSuccessSchema, AgentUndoRequestSchema, AgentUndoSuccessSchema, + AgentWriteBatchRequestSchema, + AgentWriteBatchSuccessSchema, AgentWriteMdRequestSchema, AgentWriteMdSuccessSchema, AgentWriteRequestSchema, @@ -48,6 +51,7 @@ import { applyPatchToFm, BacklinkCountsSuccessSchema, BacklinksSuccessSchema, + type BatchEntryError, BranchInfoResponseSchema, CheckoutRequestSchema, CheckoutResponseSchema, @@ -335,6 +339,7 @@ import { type SemanticSearchService, } from './embeddings/index.ts'; import { + classifyParseError, FrontmatterMalformedError, respondFrontmatterMalformed, } from './frontmatter-malformed-error.ts'; @@ -701,13 +706,13 @@ function renameAttributionCounter(): ReturnType['cre // denominator (every gated agent write); `content_divergence_total` the // numerator (writes whose converged Y.Text diverged from intent). The ratio is // the production divergence rate. Bounded label set: -// handler ∈ {agent-write-md, agent-patch, rollback} + bounded divergence_type. +// handler ∈ {agent-write-md, agent-write-batch, agent-patch, rollback} + bounded divergence_type. let _agentWriteGateFiredCounter: ReturnType['createCounter']> | null = null; function agentWriteGateFiredCounter(): ReturnType['createCounter']> { _agentWriteGateFiredCounter ||= getMeter().createCounter('ok.agent_write.gate_fired_total', { description: - 'Count of agent writes that ran the Site A content-divergence gate (denominator for the divergence rate). Bounded label: handler ∈ {agent-write-md, agent-patch, rollback}.', + 'Count of agent writes that ran the Site A content-divergence gate (denominator for the divergence rate). Bounded label: handler ∈ {agent-write-md, agent-write-batch, agent-patch, rollback}.', }); return _agentWriteGateFiredCounter; } @@ -722,7 +727,7 @@ function agentWriteContentDivergenceCounter(): ReturnType< 'ok.agent_write.content_divergence_total', { description: - 'Count of agent writes whose converged Y.Text diverged from the composed intent (numerator for the divergence rate). Bounded labels: handler ∈ {agent-write-md, agent-patch, rollback}, divergence_type.', + 'Count of agent writes whose converged Y.Text diverged from the composed intent (numerator for the divergence rate). Bounded labels: handler ∈ {agent-write-md, agent-write-batch, agent-patch, rollback}, divergence_type.', }, ); return _agentWriteContentDivergenceCounter; @@ -759,7 +764,7 @@ function searchCorpusUpdateCounter(): ReturnType['cr } /** Bounded handler label for the content-divergence counters. */ -type DivergenceHandler = 'agent-write-md' | 'agent-patch' | 'rollback'; +type DivergenceHandler = 'agent-write-md' | 'agent-write-batch' | 'agent-patch' | 'rollback'; /** * Record a gated agent write: always bump the denominator; bump the numerator @@ -5598,6 +5603,371 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { { handler: 'agent-write-md', method: 'POST' }, ); + /** + * `POST /api/agent-write-batch` — N document writes in one HTTP call, for + * bulk create/import flows that would otherwise pay one round-trip plus the + * full per-write advisory suite per document. + * + * Semantics per entry are identical to `handleAgentWriteMd`: same + * `applyAgentMarkdownWrite` spine under the entry doc's per-session frozen + * origin (no new CRDT write path, no second undo surface — each doc's + * UndoManager sees the write exactly as a single-call write), same + * reserved-name gate re-checked per doc, same L1 reconcile-before-apply, + * same per-doc awaited disk store so an ENOSPC-class failure or an L3 + * disk-divergence revert reports as that entry's error, never a false + * success. + * + * What a batch amortizes relative to N single calls: + * - one HTTP round-trip, one identity extraction, one presence + * set/idle pair and one focus broadcast instead of N; + * - one `collectAdmittedDocNames()` file-index scan feeding every + * entry's broken-link validation; + * - one coalesced L2 shadow commit via the store path's existing commit + * debounce (per-doc L1 stores arm the same timer — the batch never + * forces per-doc commits, mirroring the single-write coalescing + * contract pinned by `agent-write-commit-coalescing.test.ts`); + * - the search corpus stays fingerprint-invalidated (mtime-keyed cache in + * `getWorkspaceSearchCorpus`), so N writes cost one rebuild on the next + * search, not N eager invalidations. + * + * The per-doc mermaid render validation, lint pass, and orphan hints are + * deliberately NOT run here — their per-write cost is what a batch exists + * to avoid. Callers who want those advisories use the single-write + * endpoint; `brokenLinks` (cheap once the admitted set is shared) is kept + * per entry. + * + * Outcomes are per-entry and independent (partial success by design; no + * atomic mode): `results[i]` answers `docs[i]`, failures carry the same + * closed URN vocabulary as the single-write endpoints, and the batch + * response is HTTP 200 whenever the request itself was well-formed. + */ + const handleAgentWriteBatch = withValidation( + AgentWriteBatchRequestSchema, + async (_req, res, body) => { + try { + const { agentId, agentName, colorSeed, clientName, clientVersion, label } = + extractAgentIdentity(body); + + const timestamp = new Date().toISOString(); + + // Handler-internal result shapes; the wire contract is enforced at + // emit time by `successResponse`'s safeParse against + // `AgentWriteBatchSuccessSchema` (the `.loose()` index signatures the + // z.infer types carry don't unify with interface-typed helpers like + // `BrokenOutboundLink`). + interface BatchErrorResult { + status: 'error'; + docName: string; + error: { type: BatchEntryError['type']; title: string; detail?: string }; + } + interface BatchWrittenResult { + status: 'written'; + docName: string; + summary?: SummaryResponse; + warnings?: AdvisoryWarning[]; + brokenLinks: ReturnType; + } + type BatchResult = BatchWrittenResult | BatchErrorResult; + + const entryError = ( + docName: string, + type: BatchEntryError['type'], + title: string, + detail?: string, + ): BatchErrorResult => ({ + status: 'error', + docName, + error: { type, title, ...(detail !== undefined ? { detail } : {}) }, + }); + + /** Failure translation mirroring the single-write catch arms — same + * URN per error class, demoted from a wire-level response to a + * per-entry result so siblings keep processing. The conflict / + * malformed-FM branches re-emit the structured refusal events whose + * single-site emission normally lives in the `respond*` helpers, so + * the grouped-by-handler refusal counters see batch refusals too. */ + const classifyEntryFailure = (docName: string, e: unknown): BatchErrorResult => { + if (e instanceof DocInConflictError) { + console.warn( + JSON.stringify({ + event: 'doc-in-conflict-write-refused', + handler: 'agent-write-batch', + 'doc.name': docName, + }), + ); + return entryError( + docName, + 'urn:ok:error:doc-in-conflict', + 'Document is in conflict.', + 'The document is in a merge-conflict state. Call conflicts({ kind: "content" }) + resolve_conflict before retrying.', + ); + } + if (e instanceof FrontmatterMalformedError) { + console.warn( + JSON.stringify({ + event: 'frontmatter-malformed-write-refused', + handler: 'agent-write-batch', + class: classifyParseError(e.parseError), + 'doc.name': docName, + parseError: e.parseError, + }), + ); + return entryError( + docName, + 'urn:ok:error:frontmatter-malformed', + 'Frontmatter YAML is malformed.', + e.parseError, + ); + } + if (e instanceof AgentSessionCapacityError) { + return entryError( + docName, + 'urn:ok:error:too-many-agent-sessions', + 'Too many agent sessions.', + ); + } + log.error({ err: e, docName }, '[agent-write-batch] entry failed'); + return entryError( + docName, + 'urn:ok:error:internal-server-error', + 'Internal server error.', + ); + }; + + interface PendingEntry { + index: number; + docName: string; + session: Awaited>; + summaryResponse?: SummaryResponse; + warnings: AdvisoryWarning[]; + } + + const results: (BatchResult | undefined)[] = new Array(body.docs.length); + const pending: PendingEntry[] = []; + + // One presence set/idle pair for the whole batch — per-entry + // setPresence would spam `__system__` awareness N times for one + // logical operation. The focus broadcast at the end navigates + // followers to the last written doc. + try { + const icon = iconFromClientName(clientName); + const color = AGENT_ICON_COLORS[icon] ?? colorFromSeed(colorSeed ?? agentId); + agentPresenceBroadcaster?.setPresence(agentId, { + displayName: agentName, + icon, + color, + currentDoc: resolveAlias(body.docs[0].docName), + mode: 'writing', + ts: Date.now(), + }); + + // Apply phase: entries in array order, so duplicate docNames behave + // like sequential writes to that doc. + for (let i = 0; i < body.docs.length; i++) { + const entry = body.docs[i]; + const resolvedDocName = resolveAlias(entry.docName); + + if (isSystemDoc(resolvedDocName) || isConfigDoc(resolvedDocName)) { + results[i] = entryError( + resolvedDocName, + 'urn:ok:error:reserved-doc-name', + `'${resolvedDocName}' is a reserved document name.`, + ); + continue; + } + + try { + if ( + entry.extension !== undefined && + !docNameExistsWithAnySupportedExtension(contentDir, resolvedDocName) + ) { + registerDocExtension(resolvedDocName, entry.extension); + } + + const normalizedSummary = normalizeSummary(entry.summary); + const { response: summaryResponse, stored: storedSummary } = + summaryResponseFields(normalizedSummary); + const session = await sessionManager.getSession(resolvedDocName, agentId, { + displayName: agentName, + colorSeed, + clientName, + }); + + const reconcile = reconcileDiskBeforeAgentWrite( + hocuspocus, + resolvedDocName, + contentDir, + options.resolveEmbed, + ); + + let writeDivergence: AgentWriteContentDivergence | undefined; + // Register one-shot observer BEFORE write transact so YTextEvent.delta is captured + captureEffect(session.dc.document.getText('source'), agentId, colorSeed, clientName); + // Use per-session origin, not shared AGENT_WRITE_ORIGIN (STOP rule) + session.dc.document.transact(() => { + const beforeBlocks = snapshotBlocks(session.dc.document); + writeDivergence = applyAgentMarkdownWrite( + session.dc.document, + entry.markdown, + entry.position ?? 'append', + options.resolveEmbed + ? { resolveEmbed: options.resolveEmbed, sourcePath: resolvedDocName } + : undefined, + ); + + const changedBlocks = + changedBlockRange(beforeBlocks, snapshotBlocks(session.dc.document)) ?? undefined; + const activityMap = session.dc.document.getMap('agent-flash'); + activityMap.set(agentId, { + agentId, + timestamp: Date.now(), + type: 'insert', + description: `Added (${agentName}): ${entry.markdown.trim().slice(0, 50)}`, + ...(changedBlocks !== undefined ? { changedBlocks } : {}), + }); + }, session.origin); + + recordContentDivergenceGate('agent-write-batch', writeDivergence); + recordContributor( + resolvedDocName, + agentId, + agentName, + colorSeed, + undefined, + buildAgentActor({ clientName, clientVersion, label }), + storedSummary, + ); + incrementAgentWriteCalls(); + countNormalizedSummary(normalizedSummary); + + const reconcileWarning = buildReconcileWarning(reconcile); + const warnings: AdvisoryWarning[] = [ + ...(writeDivergence !== undefined + ? [toContentDivergenceWarning(writeDivergence)] + : []), + ...(reconcileWarning ? [reconcileWarning] : []), + ]; + pending.push({ + index: i, + docName: resolvedDocName, + session, + summaryResponse, + warnings, + }); + } catch (e) { + results[i] = classifyEntryFailure(resolvedDocName, e); + } + } + + // Flush phase: one awaited L1 disk store per unique doc, after every + // CRDT write has applied — disk truth per entry without interleaving + // store cycles between writes. Each store arms the shared L2 commit + // debounce; the batch never forces a per-doc shadow commit. + const flushErrors = new Map(); + for (const p of pending) { + if (flushErrors.has(p.docName)) continue; + const flushOutcome = await flushDiskAndDetectOutcome(p.docName); + if (flushOutcome?.kind === 'failure') { + const reason = classifyUploadErrno({ + code: flushOutcome.failure.code, + } as NodeJS.ErrnoException); + flushErrors.set(p.docName, { + type: reason, + title: 'Write applied in memory but failed to persist to disk.', + detail: `${flushOutcome.failure.code ?? 'unknown error'}: ${flushOutcome.failure.message}. The content was NOT saved and will be lost if the server restarts.`, + }); + } else if (flushOutcome?.kind === 'divergence') { + flushErrors.set(p.docName, { + type: 'urn:ok:error:disk-divergence', + title: + 'The document changed on disk after your edit was prepared; your edit was NOT applied. Re-read the document and retry.', + }); + } else { + flushErrors.set(p.docName, undefined); + } + } + + // Advisory phase: broken-link validation from the just-written + // source bytes, with the O(corpus) admitted-set scan paid once for + // the whole batch. Every flushed batch doc joins the admitted set + // first so intra-batch links (doc A -> doc B written in the same + // call) validate regardless of entry order. + const admittedForLinks = collectAdmittedDocNames(); + for (const p of pending) { + if (flushErrors.get(p.docName) === undefined) admittedForLinks.add(p.docName); + } + + let lastWrittenDoc: string | undefined; + for (const p of pending) { + const flushError = flushErrors.get(p.docName); + if (flushError !== undefined) { + results[p.index] = { status: 'error', docName: p.docName, error: flushError }; + continue; + } + const writtenSource = p.session.dc.document.getText('source').toString(); + registerWrittenDocInFileIndex(p.docName, writtenSource); + const brokenLinks = computeBrokenOutboundLinks( + writtenSource, + p.docName, + admittedForLinks, + linkedFileExists, + ); + results[p.index] = { + status: 'written', + docName: p.docName, + ...(p.summaryResponse ? { summary: p.summaryResponse } : {}), + ...(p.warnings.length > 0 ? { warnings: p.warnings } : {}), + brokenLinks, + }; + lastWrittenDoc = p.docName; + } + + if (lastWrittenDoc !== undefined) { + agentFocusBroadcaster?.setFocus(agentId, { + agentName, + currentDoc: lastWrittenDoc, + writeKind: 'write', + ts: Date.now(), + }); + onAgentWrite?.(); + } + } finally { + agentPresenceBroadcaster?.touchMode(agentId, 'idle'); + } + + const finalResults: BatchResult[] = results.map( + (r, i) => + r ?? + entryError( + body.docs[i].docName, + 'urn:ok:error:internal-server-error', + 'Internal server error.', + ), + ); + const written = finalResults.filter((r) => r.status === 'written').length; + successResponse( + res, + 200, + AgentWriteBatchSuccessSchema, + { + timestamp, + results: finalResults, + written, + failed: finalResults.length - written, + }, + { handler: 'agent-write-batch' }, + ); + } catch (e) { + log.error({ err: e }, '[agent-write-batch] handler failed'); + errorResponse(res, 500, 'urn:ok:error:internal-server-error', 'Internal server error.', { + handler: 'agent-write-batch', + cause: e, + }); + } + }, + { handler: 'agent-write-batch', method: 'POST' }, + ); + /** * `POST /api/frontmatter-patch` — JSON Merge Patch (RFC 7396) for the YAML * region of `Y.Text('source')`. Mirrors `handleAgentWriteMd`'s session + @@ -19139,6 +19509,7 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { '/api/upload': handleUploadAsset, '/api/agent-write': handleAgentWrite, '/api/agent-write-md': handleAgentWriteMd, + '/api/agent-write-batch': handleAgentWriteBatch, '/api/frontmatter-patch': handleFrontmatterPatch, '/api/agent-patch': handleAgentPatch, '/api/agent-undo': handleAgentUndo, @@ -19217,6 +19588,7 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { '/api/trash/cleanup', '/api/agent-write', '/api/agent-write-md', + '/api/agent-write-batch', '/api/frontmatter-patch', '/api/agent-patch', '/api/agent-undo',