From 4d0b6835fb9ee9cd802e7cff120e4ea564f7f1cd Mon Sep 17 00:00:00 2001 From: AgentLinkOps Date: Wed, 7 Oct 2026 03:38:32 -0400 Subject: [PATCH] Fix sync acknowledgements and prepare CLI 0.6.12 --- .claude-plugin/plugin.json | 2 +- .codex-plugin/plugin.json | 2 +- .github/workflows/release.yml | 9 +- CHANGELOG.md | 6 + cli/client.js | 2 +- cli/main.js | 7 + cli/mcp-client.js | 4 +- cli/sync.js | 25 +- package-lock.json | 4 +- package.json | 2 +- .../references/agentlinkops.md | 2 +- test/backend-sync-large-import.test.js | 222 ++++++++++++++++++ test/cli-client-resilience.test.js | 46 +++- test/cli-sync.test.js | 71 ++++++ 14 files changed, 385 insertions(+), 19 deletions(-) create mode 100644 test/backend-sync-large-import.test.js diff --git a/.claude-plugin/plugin.json b/.claude-plugin/plugin.json index 7ade711..e8c9b92 100644 --- a/.claude-plugin/plugin.json +++ b/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "agentlinkops", - "version": "0.6.11", + "version": "0.6.12", "description": "Connect an agent to AgentLinkOps and run backlink work with it: the connect skill (views, the three discovery verbs, evidence rules), local ledger and campaign skills, a portable CRM, the source-cited content pipeline, user-managed disavow lists, HTML reports and monitoring-event sync.", "author": { "name": "AgentLinkOps" diff --git a/.codex-plugin/plugin.json b/.codex-plugin/plugin.json index fe26d1e..ff322fe 100644 --- a/.codex-plugin/plugin.json +++ b/.codex-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "agentlinkops", - "version": "0.6.11", + "version": "0.6.12", "description": "Connect an agent to AgentLinkOps and run backlink work with it: the connect skill (views, the three discovery verbs, evidence rules), local ledger and campaign skills, a portable CRM, the source-cited content pipeline, user-managed disavow lists, HTML reports and monitoring-event sync.", "author": { "name": "AgentLinkOps" diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index b6048a4..f95874a 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -1,8 +1,5 @@ -# One tagged version reaches npm from here, with provenance, under the AgentLinkOps identity: -# push a tag `vX.Y.Z` that matches package.json, and this workflow tests, packs and publishes -# through npm trusted publishing (OIDC; no token in the repository). The GitHub release is -# created from the tag with the CHANGELOG entry. `workflow_dispatch` with dry_run rehearses -# everything except the final upload. +# Releases are qualified, packed and published locally under the owner deployment policy. +# The retained workflow is disabled so pushing a tag cannot build or publish a second artifact. name: Release on: push: @@ -18,6 +15,8 @@ permissions: id-token: write jobs: release: + # Owner policy: qualify and publish locally; tags must not launch a release job. + if: ${{ false }} runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 diff --git a/CHANGELOG.md b/CHANGELOG.md index d936f96..70f0a60 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,12 @@ All notable changes to the `agentlinkops` package are recorded here. The version follows the plugin manifests in `.claude-plugin/` and `.codex-plugin/`. +## 0.6.12 (2026-10-07) + +- Check every watch import result before updating sync mappings or reporting progress. Reject incomplete, duplicate or out-of-range results without resending the request. +- Show inspection and read-only cursor recovery steps after a sync timeout or lost connection. Keep the project ledger and history while the write outcome is unknown. +- Test multi-batch imports and uncertain responses in the assembled CLI package. + ## 0.6.11 (2026-10-06) - Keep `check --json` stdout limited to observation rows. Send empty-ledger, not-due and filter notices to stderr. diff --git a/cli/client.js b/cli/client.js index a6cc2a9..a631b6b 100644 --- a/cli/client.js +++ b/cli/client.js @@ -5,7 +5,7 @@ export {CloudError} from './cloud-error.js'; // A small HTTP client for the cloud. Scoped keys only; no human session ever reaches here. // DP-0036-T13: the CLI names itself so the usage counters can tell it from other REST callers. // The version follows the plugin manifests; bump it with them. -export const CLI_VERSION = '0.6.11'; +export const CLI_VERSION = '0.6.12'; export const CLI_USER_AGENT = `agentlinkops-cli/${CLI_VERSION}`; const MAX_ATTEMPTS = 3; const MAX_RETRY_WAIT_MS = 5_000; diff --git a/cli/main.js b/cli/main.js index 857cca1..42f21b8 100644 --- a/cli/main.js +++ b/cli/main.js @@ -625,6 +625,13 @@ export async function main(argv = process.argv.slice(2), { cwd = process.cwd(), err(JSON.stringify({error: {code: error.code, status: error.status, next: 'Keep this project ledger and its history. Resolve authorized read access for the configured project, then retry agentlinkops sync --pull-only --recover-cursors. Do not broaden grants or replay remote writes.'}})); return 2; } + if (command === 'sync' && error instanceof CloudError && error.status === 0 && + ['CLOUD_TIMEOUT', 'CLOUD_NETWORK_ERROR'].includes(error.code)) { + // A local transport failure cannot establish whether a write committed. + // Print trusted recovery guidance, never attached exception/server text. + err(`agentlinkops: ${error.code}\nThe request outcome is unknown. Check the operation result before sending it again. Keep this project ledger and its history. Inspect accessible watches and their local references with agentlinkops sync --pull-only --recover-cursors. Do not broaden grants or replay remote writes.`); + return 2; + } err(`agentlinkops: ${error?.message ?? error}`); return 2; } finally { diff --git a/cli/mcp-client.js b/cli/mcp-client.js index 17ad9a3..fac0922 100644 --- a/cli/mcp-client.js +++ b/cli/mcp-client.js @@ -62,7 +62,7 @@ function retryDelay(error, attempt, now) { async function sdkClient({url, token, workspaceId, fetchImpl}) { const {Client, StreamableHTTPClientTransport} = await import('@modelcontextprotocol/client'); - const client = new Client({name: 'agentlinkops-cli', version: '0.6.11'}); + const client = new Client({name: 'agentlinkops-cli', version: '0.6.12'}); const destination = new URL(url); const transport = new StreamableHTTPClientTransport(destination, { authProvider: {token: async () => token}, @@ -71,7 +71,7 @@ async function sdkClient({url, token, workspaceId, fetchImpl}) { if (target.href !== destination.href) fail('OAUTH_TOKEN_DESTINATION_MISMATCH'); const headers = new Headers(init.headers ?? (input instanceof Request ? input.headers : undefined)); if (workspaceId) headers.set('X-Workspace-ID', workspaceId); - headers.set('User-Agent', 'agentlinkops-cli/0.6.11'); + headers.set('User-Agent', 'agentlinkops-cli/0.6.12'); const response = await fetchImpl(input, {...init, headers, redirect: 'error', signal: AbortSignal.timeout(30000)}); if (response.redirected || response.status >= 300 && response.status < 400 || response.url && response.url !== destination.href) fail('OAUTH_TOKEN_DESTINATION_MISMATCH'); diff --git a/cli/sync.js b/cli/sync.js index ebb94ff..2ef59bf 100644 --- a/cli/sync.js +++ b/cli/sync.js @@ -12,6 +12,24 @@ import { CloudError } from './client.js'; const PUSH_BATCH = 100; +function completeImportRows(result, expected) { + const record = value => value !== null && typeof value === 'object' && !Array.isArray(value); + const text = value => typeof value === 'string' && value.trim().length > 0; + const refuse = () => { throw new CloudError('INVALID_IMPORT_RESPONSE', 502, null, + 'The cloud returned an incomplete or invalid watch import result. Some watches may already have been stored. Inspect accessible watches and their local references before retrying sync; this request was not automatically replayed.'); }; + if (!record(result) || !Array.isArray(result.rows) || result.rows.length !== expected) refuse(); + const seen = new Set(); + for (const row of result.rows) { + if (!record(row) || !Number.isInteger(row.index) || row.index < 0 || row.index >= expected || seen.has(row.index)) refuse(); + seen.add(row.index); + const watch = Object.hasOwn(row, 'watch'), error = Object.hasOwn(row, 'error'); + if (watch === error) refuse(); + if (watch && (!record(row.watch) || !text(row.watch.id) || typeof row.watch.created !== 'boolean')) refuse(); + if (error && (!record(row.error) || !text(row.error.code) || !text(row.error.message))) refuse(); + } + return result.rows; +} + export function syncPlan(entries, state, { includeWanted = false, ledgerIds = null } = {}) { if (ledgerIds !== null && (!Array.isArray(ledgerIds) || ledgerIds.some(id => typeof id !== 'string') || ledgerIds.some(id => !entries.some(entry => entry.id === id)))) throw new CloudError('INVALID_SYNC_SELECTION', 0); @@ -38,12 +56,13 @@ export async function pushExpectations(client, entries, state, { projectId, onPr for (let index = 0; index < watched.length; index += PUSH_BATCH) { const batch = watched.slice(index, index + PUSH_BATCH); const result = await client.importWatches(projectId, batch.map(toWatchInput)); - for (const row of result.rows ?? []) { + // Validate the complete acknowledgement before adopting any of this batch. + // Earlier POSTs may have committed; a malformed response never authorizes a replay. + for (const row of completeImportRows(result, batch.length)) { const entry = batch[row.index]; - if (!entry) continue; if (row.error) { failed.push({ id: entry.id, code: row.error.code, message: row.error.message }); continue; } mapping[entry.id] = row.watch.id; - if (row.watch.created !== false) created.push(entry.id); + if (row.watch.created) created.push(entry.id); } onProgress?.(Math.min(index + PUSH_BATCH, watched.length), watched.length); } diff --git a/package-lock.json b/package-lock.json index 76149e0..3eae9c3 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "agentlinkops", - "version": "0.6.11", + "version": "0.6.12", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "agentlinkops", - "version": "0.6.11", + "version": "0.6.12", "license": "Apache-2.0", "dependencies": { "@modelcontextprotocol/client": "2.2.0", diff --git a/package.json b/package.json index 0a210f3..cd065c4 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "agentlinkops", - "version": "0.6.11", + "version": "0.6.12", "description": "Backlink checks with dated evidence and portable results, optional repository ledgers, supplier-export imports and sync to AgentLinkOps.", "author": "AgentLinkOps (https://agentlinkops.com)", "license": "Apache-2.0", diff --git a/skills/agentlinkops-connect/references/agentlinkops.md b/skills/agentlinkops-connect/references/agentlinkops.md index 1685e76..cce2c08 100644 --- a/skills/agentlinkops-connect/references/agentlinkops.md +++ b/skills/agentlinkops-connect/references/agentlinkops.md @@ -9,7 +9,7 @@ records. This page is the complete agent reference: the same text prints from full once per session, then use `describe` (or `exec describe` on MCP) for any command's exact schema. -Pack version 0.6.11. 139 commands in 15 toolsets. +Pack version 0.6.12. 139 commands in 15 toolsets. ## Install and connect (once per machine) diff --git a/test/backend-sync-large-import.test.js b/test/backend-sync-large-import.test.js new file mode 100644 index 0000000..2a42e17 --- /dev/null +++ b/test/backend-sync-large-import.test.js @@ -0,0 +1,222 @@ +import test, {mock} from 'node:test'; +import assert from 'node:assert/strict'; +import {mkdtemp, mkdir, writeFile, readFile, readdir, rm} from 'node:fs/promises'; +import {tmpdir} from 'node:os'; +import {join} from 'node:path'; + +// Only advisory acquisition is doubled in the module-mock lane. Ordinary package +// tests use the actual native binding. This file exercises the CLI and its files; +// supplied HTTP replies remain an offline fixture, not a hosted soak or billing proof. +if (typeof mock.module === 'function') mock.module('../cli/native-lock.js', { + namedExports: {nativeLock: async () => ({acquire: () => true})}, +}); +const {main} = await import('../cli/main.js'); + +const COUNT = 2003; +const ORIGIN = 'https://large-sync.fixture.invalid'; +const PROJECT = 'pr_large_fixture'; +const localId = n => `lk_large${String(n).padStart(7, '0')}`; +const watchId = id => `wat_${id}`; +const RETIRED = 'lk_retired000001'; +const authoredEntry = n => ({ + id: localId(n), intent: 'expected', source: `https://publisher.fixture.invalid/${n}`, + target: 'https://customer.fixture.invalid/guide', scope: 'exact', cadence: 'weekly', + ref: `external/row/${n}`, note: `private note ${n}`, tags: ['customer-tag'], + customer: {originalRow: n, retain: true}, +}); + +async function ledgerFixture(t) { + const cwd = await mkdtemp(join(tmpdir(), 'alo-large-sync-')); + t.after(() => rm(cwd, {recursive: true, force: true})); + const dir = join(cwd, '.agentlinkops'); + await mkdir(dir, {mode: 0o700}); + const entries = Array.from({length: COUNT}, (_, n) => authoredEntry(n)); + const wanted = Array.from({length: 3}, (_, n) => ({ + ...authoredEntry(COUNT + n), id: `lk_wanted${String(n).padStart(7, '0')}`, intent: 'wanted', + })); + const retired = {...authoredEntry(COUNT + 3), id: RETIRED, intent: 'retired'}; + const ledger = [...entries, ...wanted, retired].map(row => JSON.stringify(row)).join('\n') + '\n'; + const config = JSON.stringify({project: {id: PROJECT}, cloud: {origin: ORIGIN, token: 'offline-fixture'}}); + const state = { + v: 2, entries: {}, watches: {[RETIRED]: 'wat_retired', lk_preserved0001: 'wat_preserved'}, + cursors: {events: 'old-source', target_events: 'old-target'}, + customerCheckpoint: {externalRevision: 42, notes: 'retain original state'}, + }; + const stateBytes = JSON.stringify(state) + '\n'; + await writeFile(join(dir, 'links.jsonl'), ledger); + await writeFile(join(dir, 'config.json'), config); + await writeFile(join(dir, 'state.json'), stateBytes); + return {cwd, dir, entries, wanted, ledger, config, state, stateBytes}; +} + +function requestLog(url, init, calls) { + const parsed = new URL(url); + assert.equal(parsed.origin, ORIGIN, 'Every request stays inside the injected fixture'); + const row = { + method: init.method, path: parsed.pathname, cursor: parsed.searchParams.get('cursor'), + project: parsed.searchParams.get('projectId'), body: init.body ? JSON.parse(init.body) : null, + }; + calls.push(row); + return row; +} + +async function unchangedAuthored(f) { + assert.equal(await readFile(join(f.dir, 'links.jsonl'), 'utf8'), f.ledger); + assert.equal(await readFile(join(f.dir, 'config.json'), 'utf8'), f.config); +} + +test('actual CLI persists all acknowledged positions of a 2,003-row upload across reordered chunks and row refusals', async t => { + const f = await ledgerFixture(t), calls = [], output = [], errors = []; + const rejected = new Set([localId(137), localId(COUNT - 1)]); + const existing = new Set([localId(0), localId(1700)]); + const supplied = new Set(); + const fetchImpl = async (url, init) => { + const request = requestLog(url, init, calls); + if (request.method === 'PATCH') { + assert.equal(request.path, '/v1/watches/wat_retired'); + assert.deepEqual(request.body, {status: 'paused'}); + return Response.json({id: 'wat_retired', status: 'paused'}); + } + assert.equal(request.method, 'POST'); + assert.equal(request.path, '/v1/watches/import'); + assert.equal(request.body.projectId, PROJECT); + const rows = request.body.watches.map((watch, index) => { + const id = watch.localReference.split(' ')[0]; + assert.equal(supplied.has(id), false, 'An input position is submitted only once'); + supplied.add(id); + assert.ok(!JSON.stringify(watch).includes('private note')); + assert.ok(!JSON.stringify(watch).includes('customer-tag')); + if (rejected.has(id)) return {index, error: {code: 'WATCH_LIMIT_REACHED', message: 'No remaining watch capacity.'}}; + return {index, watch: {id: watchId(id), created: !existing.has(id)}}; + }).reverse(); + return Response.json({status: rows.some(row => row.error) ? 'partial' : 'succeeded', rows}); + }; + assert.equal(await main(['sync', '--push-only'], {cwd: f.cwd, env: {}, fetchImpl, + out: value => output.push(value), err: value => errors.push(value)}), 2); + assert.deepEqual(errors, []); + assert.equal(output[0], 'pushed 2003 (1999 new, 1 retired, 3 kept local)'); + assert.ok(output.some(line => line.includes(`${localId(137)}: WATCH_LIMIT_REACHED`))); + assert.ok(output.some(line => line.includes(`${localId(COUNT - 1)}: WATCH_LIMIT_REACHED`))); + const uploads = calls.filter(row => row.method === 'POST'); + assert.deepEqual(uploads.map(row => row.body.watches.length), [...Array(20).fill(100), 3]); + assert.equal(supplied.size, COUNT); + assert.equal(calls.length, 22, 'The only other request deliberately pauses the retired watch'); + assert.equal(calls.at(-1).method, 'PATCH'); + const state = JSON.parse(await readFile(join(f.dir, 'state.json'), 'utf8')); + for (const entry of f.entries) assert.equal(state.watches[entry.id], rejected.has(entry.id) ? undefined : watchId(entry.id)); + for (const entry of f.wanted) assert.equal(state.watches[entry.id], undefined); + assert.equal(state.watches[RETIRED], 'wat_retired'); + assert.equal(state.watches.lk_preserved0001, 'wat_preserved'); + assert.equal(Object.keys(state.watches).length, COUNT, 'Every success and both preexisting mappings survive; rejects get no guessed ID'); + assert.deepEqual(state.cursors, f.state.cursors, 'A push cannot advance either read cursor'); + assert.deepEqual(state.customerCheckpoint, f.state.customerCheckpoint); + await unchangedAuthored(f); + assert.ok(!(await readdir(f.dir)).includes('observations.jsonl'), 'An import acknowledgement does not invent observations'); +}); + +test('a dropped final large-import acknowledgement preserves local state and recovers by paged reads without another POST', async t => { + const f = await ledgerFixture(t), calls = [], committed = new Map(), uploadOutput = [], uploadErrors = []; + const upload = async (url, init) => { + const request = requestLog(url, init, calls); + assert.equal(request.method, 'POST'); + assert.equal(request.path, '/v1/watches/import'); + assert.equal(request.body.projectId, PROJECT); + const rows = request.body.watches.map((watch, index) => { + const id = watch.localReference.split(' ')[0]; + assert.equal(committed.has(id), false, 'A committed write is never automatically replayed'); + committed.set(id, {...watch, id: watchId(id), project_id: PROJECT, local_reference: watch.localReference, + source_url: watch.sourceUrl, target_url: watch.targetUrl, target_scope: watch.targetScope}); + return {index, watch: {id: watchId(id), created: true}}; + }); + // The fixture accepted every last position before its response disappeared. + // The client cannot infer that outcome from the transport exception. + if (committed.size === COUNT) throw new DOMException('Injected dropped committed response', 'TimeoutError'); + return Response.json({status: 'succeeded', rows: rows.reverse()}); + }; + assert.equal(await main(['sync'], {cwd: f.cwd, env: {}, fetchImpl: upload, + out: value => uploadOutput.push(value), err: value => uploadErrors.push(value)}), 2); + assert.match(uploadErrors.join('\n'), /Check the operation result before sending it again/); + assert.deepEqual(uploadOutput, [], 'The CLI must not claim the full push finished'); + assert.equal(committed.size, COUNT, 'Only the fixture knows the ambiguous POST committed'); + assert.deepEqual(calls.map(row => row.body.watches.length), [...Array(20).fill(100), 3]); + assert.ok(calls.every(row => row.method === 'POST'), 'No subsequent pause, check, event read or replay follows the ambiguous acknowledgement'); + assert.equal(await readFile(join(f.dir, 'state.json'), 'utf8'), f.stateBytes); + await unchangedAuthored(f); + assert.ok(!(await readdir(f.dir)).includes('events.jsonl')); + + const checkedAt = '2026-10-06T12:00:00.000Z'; + const current = {state: 'unknown', uncertain: true, checked_at: checkedAt, + latestAttempt: {state: 'unknown', reason: 'timeout', evidence: {complete: false}}}; + const watches = [...committed.values()].map(watch => ({...watch, observation_state: current})); + const echo = n => ({id: `event_large_${n}`, watch_id: watchId(localId(n)), project_id: PROJECT, + data: {after: {state: 'unknown', uncertain: true, checked_at: checkedAt, reason: 'timeout', + source_url: f.entries[n].source, target_url: f.entries[n].target, target_scope: 'exact'}}}); + const readOnly = async (url, init) => { + const request = requestLog(url, init, calls); + assert.equal(request.method, 'GET', 'Recovery reconciles results without replaying remote writes or requesting checks'); + assert.equal(request.project, PROJECT, 'Read recovery retains the configured project scope'); + if (request.path === '/v1/events') { + if (request.cursor === 'old-source') return Response.json({error: {code: 'INVALID_CURSOR', + message: 'The saved binding cannot be used.', details: {resume_cursor: 'untrusted-resume'}}}, {status: 400}); + if (request.cursor === 'source-done') return Response.json({events: [], next_cursor: 'source-done', has_more: false}); + if (request.cursor === null) return Response.json({events: [echo(0)], next_cursor: 'source-page2', has_more: true}); + assert.equal(request.cursor, 'source-page2'); + return Response.json({events: [echo(COUNT - 1)], next_cursor: 'source-done', has_more: false}); + } + if (request.path === '/v1/target-events') { + assert.ok(['old-target', 'target-done'].includes(request.cursor)); + return Response.json({events: [], next_cursor: 'target-done', has_more: false}); + } + assert.ok(['/v1/exports/watches', '/v1/watches'].includes(request.path)); + const offset = request.cursor === null ? 0 : Number(request.cursor.slice('offset-'.length)); + assert.ok(Number.isSafeInteger(offset) && offset >= 0 && offset < COUNT); + return Response.json({items: watches.slice(offset, offset + 100), next_cursor: offset + 100 < COUNT ? `offset-${offset + 100}` : null}); + }; + const recover = extra => main(['sync', '--pull-only', '--recover-cursors'], {cwd: f.cwd, env: {}, + fetchImpl: readOnly, out: () => {}, err: () => {}, ...extra}); + const beforeRecovery = calls.length; + let checkpoint = false; + assert.equal(await recover({afterHistoryPersisted: () => {checkpoint = true; throw new Error('Injected history-before-state interruption');}}), 2); + assert.equal(checkpoint, true, 'The interruption occurred after actual history persistence'); + assert.equal(await readFile(join(f.dir, 'state.json'), 'utf8'), f.stateBytes); + const observationBytes = await readFile(join(f.dir, 'observations.jsonl'), 'utf8'); + const eventBytes = await readFile(join(f.dir, 'events.jsonl'), 'utf8'); + const rows = observationBytes.trim().split('\n').map(JSON.parse); + assert.equal(rows.length, COUNT); + assert.equal(new Set(rows.map(row => row.id)).size, COUNT); + assert.ok(rows.every(row => row.state === 'unknown' && row.complete === false && row.reason === 'timeout')); + assert.ok(rows.every(row => row.evidence_key === null && row.result.evidence.sha256 === null)); + for (const row of rows) assert.equal(row.result.sourceUrl, f.entries.find(entry => entry.id === row.id).source); + assert.equal(eventBytes.trim().split('\n').length, COUNT + 2, 'The complete snapshot and both retained feed events survive'); + const firstReads = calls.slice(beforeRecovery); + assert.equal(firstReads.filter(row => row.path === '/v1/exports/watches').length, 21); + assert.equal(firstReads.filter(row => row.path === '/v1/watches').length, 21); + assert.deepEqual(firstReads.filter(row => row.path === '/v1/events').map(row => row.cursor), ['old-source', null, 'source-page2']); + assert.ok(firstReads.findIndex(row => row.path === '/v1/events' && row.cursor === null) > + firstReads.findLastIndex(row => row.path === '/v1/exports/watches'), 'The complete snapshot precedes cursorless replay'); + + assert.equal(await recover(), 0); + assert.equal(await readFile(join(f.dir, 'observations.jsonl'), 'utf8'), observationBytes, 'History replay appends no duplicate observations'); + assert.equal(await readFile(join(f.dir, 'events.jsonl'), 'utf8'), eventBytes, 'History replay appends no duplicate events'); + const state = JSON.parse(await readFile(join(f.dir, 'state.json'), 'utf8')); + assert.deepEqual(state.cursors, {events: 'source-done', target_events: 'target-done'}); + assert.deepEqual(state.gaps.map(row => [row.feed, row.reason, row.resumed_from]), [['events', 'cursor_invalid', null]]); + assert.deepEqual(state.watches, f.state.watches, 'Read recovery does not fabricate import acknowledgements'); + assert.deepEqual(state.customerCheckpoint, f.state.customerCheckpoint); + const snapshots = (await readdir(f.dir)).filter(name => name.startsWith('snapshot-events-')); + assert.equal(snapshots.length, 2, 'Both interruption and recovery snapshots remain inspectable'); + for (const name of snapshots) { + const snapshot = JSON.parse(await readFile(join(f.dir, name), 'utf8')); + assert.equal(snapshot.items.length, COUNT); + assert.equal(snapshot.next_cursor, null); + } + const savedBytes = await readFile(join(f.dir, 'state.json'), 'utf8'), beforeSettledPull = calls.length; + assert.equal(await recover(), 0); + assert.equal(await readFile(join(f.dir, 'state.json'), 'utf8'), savedBytes); + assert.equal(await readFile(join(f.dir, 'observations.jsonl'), 'utf8'), observationBytes); + assert.equal(await readFile(join(f.dir, 'events.jsonl'), 'utf8'), eventBytes); + assert.deepEqual(calls.slice(beforeSettledPull).filter(row => row.path.includes('events')).map(row => row.cursor), ['source-done', 'target-done']); + assert.ok(calls.slice(beforeRecovery).every(row => row.method === 'GET' && row.project === PROJECT)); + assert.equal(calls.filter(row => row.method === 'POST').length, 21, 'No acknowledgement failure or recovery causes an extra write'); + await unchangedAuthored(f); +}); diff --git a/test/cli-client-resilience.test.js b/test/cli-client-resilience.test.js index 8f58527..da32e86 100644 --- a/test/cli-client-resilience.test.js +++ b/test/cli-client-resilience.test.js @@ -1,6 +1,9 @@ import test from 'node:test'; import assert from 'node:assert/strict'; import { createClient, CloudError } from '../cli/client.js'; +import {mkdtemp,mkdir,writeFile,readFile,readdir,rm} from 'node:fs/promises'; +import {tmpdir} from 'node:os'; +import {join} from 'node:path'; const ORIGIN='https://app.example.test', TOKEN='lt_fixture'; const page=()=>Response.json({items:[],next_cursor:null}); @@ -8,14 +11,14 @@ const unavailable=(status,headers={})=>Response.json({error:{code:status===429?' {status,headers:{'X-Request-ID':`req-${status}`,...headers}}); const client=(fetchImpl,options={})=>createClient({origin:ORIGIN,token:TOKEN,fetchImpl,...options}); -test('the REST client sends the 0.6.11 release User-Agent', async () => { +test('the REST client sends the 0.6.12 release User-Agent', async () => { const requests = []; const result = await client(async (url, init) => { const request = new Request(url, init); requests.push(request); assert.equal(request.url, ORIGIN + '/v1/watches'); assert.equal(request.method, 'GET'); - assert.equal(request.headers.get('user-agent'), 'agentlinkops-cli/0.6.11'); + assert.equal(request.headers.get('user-agent'), 'agentlinkops-cli/0.6.12'); assert.equal(request.headers.get('authorization'), 'Bearer ' + TOKEN); return page(); }).listWatches(); @@ -114,3 +117,42 @@ test('timeouts and exhausted network failures have actionable classifications',a assert.match(error.publicError.message,/Check the operation result before sending it again/);return true; }); }); + +for(const [code,failure]of [ + ['CLOUD_TIMEOUT',()=>new DOMException('PRIVATE_EXCEPTION_SENTINEL','TimeoutError')], + ['CLOUD_NETWORK_ERROR',()=>new TypeError('PRIVATE_EXCEPTION_SENTINEL')], +])test(`actual sync CLI gives trusted ${code} recovery without replay or local adoption`,async t=>{ + // Exercise actual main/client/push/state code and the real local sync lock. + // The supplied HTTP transport is an offline ambiguous-write fixture only. + const cwd=await mkdtemp(join(tmpdir(),'alo-sync-transport-')); + t.after(()=>rm(cwd,{recursive:true,force:true})); + const dir=join(cwd,'.agentlinkops');await mkdir(dir,{mode:0o700}); + const ledger=JSON.stringify({id:'lk_netdrop000001',intent:'expected', + source:'https://publisher.fixture.invalid/drop',target:'https://customer.fixture.invalid/guide',scope:'exact'})+'\n'; + const config=JSON.stringify({project:{id:'pr_transport_fixture'},cloud:{origin:ORIGIN,token:TOKEN}}); + const state=JSON.stringify({v:2,entries:{},watches:{},cursors:{events:'source-before',target_events:'target-before'},customerCheckpoint:{retain:true}})+'\n'; + await writeFile(join(dir,'links.jsonl'),ledger); + await writeFile(join(dir,'config.json'),config); + await writeFile(join(dir,'state.json'),state); + const {main}=await import('../cli/main.js'),out=[],err=[];let calls=0,committed=false; + const result=await main(['sync'],{cwd,env:{},out:value=>out.push(value),err:value=>err.push(value), + fetchImpl:async(url,init)=>{ + assert.equal(url,ORIGIN+'/v1/watches/import');assert.equal(init.method,'POST'); + const body=JSON.parse(init.body);assert.equal(body.projectId,'pr_transport_fixture');assert.equal(body.watches.length,1); + calls++;committed=true; + const error=failure();error.serverMessage='PRIVATE_SERVER_SENTINEL';error.details={token:'PRIVATE_TOKEN_SENTINEL'}; + throw error; + }}); + assert.equal(result,2);assert.equal(calls,1);assert.equal(committed,true,'only the fixture knows its write committed'); + assert.deepEqual(out,[],'unknown write outcome cannot claim a completed push'); + assert.equal(err.length,1);assert.ok(err[0].startsWith(`agentlinkops: ${code}\n`)); + assert.match(err[0],/Check the operation result before sending it again/); + assert.match(err[0],/sync --pull-only --recover-cursors/); + assert.match(err[0],/Do not broaden grants or replay remote writes/); + assert.doesNotMatch(err[0],/PRIVATE_(?:EXCEPTION|SERVER|TOKEN)_SENTINEL/); + assert.equal(await readFile(join(dir,'links.jsonl'),'utf8'),ledger); + assert.equal(await readFile(join(dir,'config.json'),'utf8'),config); + assert.equal(await readFile(join(dir,'state.json'),'utf8'),state); + const files=await readdir(dir); + assert.ok(!files.includes('events.jsonl'));assert.ok(!files.includes('observations.jsonl')); +}); diff --git a/test/cli-sync.test.js b/test/cli-sync.test.js index 5bc8812..5851db4 100644 --- a/test/cli-sync.test.js +++ b/test/cli-sync.test.js @@ -71,6 +71,77 @@ test('a row the cloud refuses is reported, and the rest still land', async () => assert.deepEqual(result.failed, [{ id: 'lk_ffffffff', code: 'WATCH_LIMIT_REACHED', message: 'The plan allows 100 watches.' }]); }); +test('sync import acknowledgement refuses incomplete, ambiguous and malformed batch results without replay or local mutation', async t => { + const ok = index => ({index, watch: {id: `wat_ack_${index}`, created: true}}); + const cases = [ + ['missing rows', {status: 'succeeded'}], + ['missing position', {status: 'succeeded', rows: [ok(0)]}], + ['duplicate position', {status: 'succeeded', rows: [ok(0), ok(0)]}], + ['out of range position', {status: 'succeeded', rows: [ok(0), ok(2)]}], + ['negative position', {status: 'succeeded', rows: [ok(0), ok(-1)]}], + ['fractional position', {status: 'succeeded', rows: [ok(0), ok(0.5)]}], + ['string position', {status: 'succeeded', rows: [ok(0), ok('1')]}], + ['null row', {status: 'succeeded', rows: [ok(0), null]}], + ['nonarray rows', {status: 'succeeded', rows: {0: ok(0), 1: ok(1)}}], + ['missing outcome', {status: 'succeeded', rows: [ok(0), {index: 1}]}], + ['both outcomes', {status: 'partial', rows: [ok(0), {...ok(1), error: {code: 'REFUSED', message: 'Refused.'}}]}], + ['missing watch identity', {status: 'succeeded', rows: [ok(0), {index: 1, watch: {created: true}}]}], + ['nonboolean creation', {status: 'succeeded', rows: [ok(0), {index: 1, watch: {id: 'wat_ack_1', created: 'false'}}]}], + ['malformed row rejection', {status: 'partial', rows: [ok(0), {index: 1, error: {message: 'Refused.'}}]}], + ]; + for (const [name, response] of cases) await t.test(name, async () => { + const entries = [entry('lk_ackfirst'), entry('lk_acksecond'), entry('lk_ackretired', {intent: 'retired'})]; + const state = {watches: {lk_retained: 'wat_retained', lk_ackretired: 'wat_retired'}, cursors: {events: 'saved_cursor'}}; + const originalEntries = structuredClone(entries), originalState = structuredClone(state), progress = []; + const {client, calls} = cloud({'POST /v1/watches/import': () => response}); + await assert.rejects(pushExpectations(client, entries, state, {projectId: 'pr_1', onProgress: (...values) => progress.push(values)}), error => { + assert.ok(error instanceof CloudError); + assert.equal(error.code, 'INVALID_IMPORT_RESPONSE'); + assert.equal(error.status, 502); + assert.match(error.serverMessage, /result|watch/i); + assert.match(error.serverMessage, /retry|again/i); + return true; + }); + assert.equal(calls.length, 1, 'An ambiguous POST is not automatically replayed or followed by a pause'); + assert.equal(calls[0].key, 'POST /v1/watches/import'); + assert.deepEqual(entries, originalEntries, 'Customer ledger values remain unchanged'); + assert.deepEqual(state, originalState, 'Previously adopted mappings and cursors remain unchanged'); + assert.deepEqual(progress, [], 'A batch without complete acknowledgements cannot be reported complete'); + }); +}); + +test('sync import acknowledgement accepts reordered complete results and legitimate per-row refusals', async () => { + const entries = [entry('lk_ackfirst'), entry('lk_acksecond'), entry('lk_ackthird')]; + const {client, calls} = cloud({'POST /v1/watches/import': () => ({status: 'partial', rows: [ + {index: 2, watch: {id: 'wat_existing', created: false}}, + {index: 0, error: {code: 'WATCH_LIMIT_REACHED', message: 'The plan allows 100 watches.'}}, + {index: 1, watch: {id: 'wat_new', created: true}}, + ]})}); + const progress = [], state = {watches: {lk_retained: 'wat_retained'}}; + const result = await pushExpectations(client, entries, state, {projectId: 'pr_1', onProgress: (...values) => progress.push(values)}); + assert.equal(calls.length, 1); + assert.equal(result.pushed, 3); + assert.equal(result.created, 1); + assert.deepEqual(result.watches, {lk_retained: 'wat_retained', lk_acksecond: 'wat_new', lk_ackthird: 'wat_existing'}); + assert.deepEqual(result.failed, [{id: 'lk_ackfirst', code: 'WATCH_LIMIT_REACHED', message: 'The plan allows 100 watches.'}]); + assert.deepEqual(progress, [[3, 3]]); + assert.deepEqual(state, {watches: {lk_retained: 'wat_retained'}}); +}); + +test('sync import acknowledgement stops at a later invalid batch without replay, final progress or customer-state adoption', async () => { + const entries = Array.from({length: 101}, (_, index) => entry(`lk_ack${index}`)); + entries.push(entry('lk_ackretired', {intent: 'retired'})); + const state = {watches: {lk_ackretired: 'wat_retired'}, cursors: {events: 'saved_cursor'}}; + const original = structuredClone({entries, state}), progress = []; + const {client, calls} = cloud({'POST /v1/watches/import': (parsed, init, seen) => seen.length === 1 ? imported(parsed, init) : {status: 'succeeded', rows: []}}); + await assert.rejects(pushExpectations(client, entries, state, {projectId: 'pr_1', onProgress: (...values) => progress.push(values)}), + {code: 'INVALID_IMPORT_RESPONSE', status: 502}); + assert.equal(calls.length, 2, 'Only the first two POSTs occurred; no automatic replay or later pause'); + assert.deepEqual(calls.map(call => call.body.watches.length), [100, 1]); + assert.deepEqual(progress, [[100, 101]], 'Earlier complete batch progress is retained without declaring final success'); + assert.deepEqual({entries, state}, original); +}); + test('a page applies before the cursor advances', async () => { const pages = [ { events: [{ id: 'e1', watch_id: 'w1', type: 'watch.checked', cursor: 'c1', data: {} }], next_cursor: 'c1', has_more: true },