diff --git a/docs/deck/contract.md b/docs/deck/contract.md index bf7520d..c34c3ed 100644 --- a/docs/deck/contract.md +++ b/docs/deck/contract.md @@ -14,7 +14,10 @@ with a current revision and an immutable revision subject. `requestPublication` constructs those facts; it does not publish them. A real curator job must use normal governed conditional writes: advance the current revision atomically, expect the old revision, and expect the new manifest slot to be absent. This -source does not deploy that job or grant it a credential. +source does not deploy that job or grant it a credential. The separate +[conditional curator source](curator.md) implements plan construction and +durable publication custody with explicit adapters; production adapters, +credential restrictions and served admission remain separate gates. The admitted reader supplies the exact graph, request ID, current revision, revision subject and complete manifest. Unknown fields, duplicate JSON keys, diff --git a/docs/deck/curator-demo.md b/docs/deck/curator-demo.md new file mode 100644 index 0000000..c342e8f --- /dev/null +++ b/docs/deck/curator-demo.md @@ -0,0 +1,37 @@ +# Curator publication custody demo + +Trigger: a first publication changes a synthetic fixture, but its reply is +lost. Watch UNKNOWN become CONFIRMED after a new job instance reads the same +durable custody. The send count stays one; a changed payload in that revision +refuses. The adapters in this recording are synthetic, with no native server, +credential or production request. The separate native qualifier exercises the +generated HTTP update and real native commit/recovery. + +Base `b290c515bfb3cda35eba38af3533c6c9200a0003` has the request fact builder but +no publication job. The demo checks that exact tree with a positive module +control. Demonstrated source is +`5727afd66297160c64006558fe82433247c2bac9`; subsequent recording/result files +do not change its implementation. This demonstrates the new library, rather +than an installed effect or a before/after server benchmark. + +Reproduce from a checkout containing that base object: + +```sh +node scripts/demo-deck-curator.mjs +``` + +The small raw [output recording](curator-demo.output) also serves as the readable +transcript. Download it together with its [timing file](curator-demo.timing). +Playback was verified with the installed util-linux `scriptreplay`; the +recording retains real output and timing. This command deliberately slows its +brief output for reading: + +```sh +scriptreplay --log-out curator-demo.output --log-timing curator-demo.timing --divisor 0.05 +``` + +This is a terminal recording, with no inline video-player claim. Both small +files are retained in this repository's existing Deck evidence directory. +Neither the recording nor the synthetic native controls establish production +least-privilege credential enforcement, graph policy, served read cost or +human acceptance. diff --git a/docs/deck/curator-demo.output b/docs/deck/curator-demo.output new file mode 100644 index 0000000..4c823d9 --- /dev/null +++ b/docs/deck/curator-demo.output @@ -0,0 +1,14 @@ +Script started on 2026-10-25 18:22:44-04:00 [COMMAND="node scripts/demo-deck-curator.mjs" ] +Base b290c515bfb3cda35eba38af3533c6c9200a0003 : request fact builder exists; publication job absent. +Demonstrated source HEAD: 5727afd66297160c64006558fe82433247c2bac9 +Target: source library, synthetic transport and reader; no service or credential. +Trigger: first publication; transport loses its reply after the fixture changes. +Before send, durable custody: UNKNOWN +First call: {"state":"UNKNOWN","sent":true,"transaction":null} +Restart with the SAME durable directory; reconcile by reading. +Recovery: {"state":"CONFIRMED","sent":false,"transaction":null} +Send count: 1 +Changed payload in same revision refused: true +Observed: UNKNOWN -> CONFIRMED by read-back; one send retained. + +Script done on 2026-10-25 18:22:44-04:00 [COMMAND_EXIT_CODE="0"] diff --git a/docs/deck/curator-demo.timing b/docs/deck/curator-demo.timing new file mode 100644 index 0000000..b32365d --- /dev/null +++ b/docs/deck/curator-demo.timing @@ -0,0 +1,11 @@ +0.106361 102 +0.002925 68 +0.000018 83 +0.000022 82 +0.013912 39 +0.000146 64 +0.000115 64 +0.001911 65 +0.000533 15 +0.002559 48 +0.000092 65 diff --git a/docs/deck/curator-native-results.json b/docs/deck/curator-native-results.json new file mode 100644 index 0000000..fade116 --- /dev/null +++ b/docs/deck/curator-native-results.json @@ -0,0 +1,95 @@ +{ + "candidate": { + "features": { + "dev-tools": false, + "full": true, + "jemalloc": true, + "lancedb": false, + "mcp": true, + "onnx": true, + "owl": true, + "oxicompare": false, + "reactive-reasoner": true, + "remote": true, + "server": true, + "shacl": true + }, + "git_dirty": false, + "git_sha": "703e2ce83fd42992ded8214ff62320b84204b0d4", + "read_contracts": [ + "quipu-acquisition/v1", + "quipu-acquisition/composed-v1", + "quipu-eligibility/v1" + ], + "version": "0.16.3" + }, + "candidate_sha256": "bdba9adb323696e0b32e10cc36f6cd6a7af3f14462fcd4c4e869f85ffd0ce452", + "synthetic_loopback_only": true, + "real_credential_used": false, + "production_authority_proven": false, + "results": [ + { + "case": "identical-first-replay", + "no_partial_request_mutation": true + }, + { + "case": "initial-and-next", + "exact_old_manifest_retained": true, + "escaped_text_is_data": true + }, + { + "case": "stale-revision", + "no_partial_request_mutation": true + }, + { + "case": "changed-item", + "no_partial_request_mutation": true + }, + { + "case": "foreign-item", + "no_partial_request_mutation": true + }, + { + "case": "ambiguous-item-revision", + "no_partial_request_mutation": true + }, + { + "case": "ambiguous-item-status", + "no_partial_request_mutation": true + }, + { + "case": "closed-item", + "no_partial_request_mutation": true + }, + { + "case": "ambiguous-current-head", + "no_partial_request_mutation": true + }, + { + "case": "occupied-immutable-revision", + "no_partial_request_mutation": true + }, + { + "case": "concurrent-different-payload", + "exactly_one_atomic_publication": true + }, + { + "case": "native-commit-lost-response-restart", + "read_only_recovery": true, + "sends": 1 + }, + { + "case": "graph-mutation-boundary", + "positive_item_control_unchanged": true, + "answer_graph_unchanged": true + } + ], + "qualified_curator_source": "5727afd66297160c64006558fe82433247c2bac9", + "source_sha256": { + "web/deck/curator.mjs": "d6b3fa6a4bfdb7e25b4ec5169150c18714a4ffbf82471d6e4d9f41e9a52c6f8a", + "scripts/deck-curator-job.mjs": "9825094b63e6a403e2a8fc144a2da24fb8dba207ddedbe17cde09a26392a5461", + "scripts/qualify-deck-curator.mjs": "cbc3520daeba6a905161178c06eea8e31ba604e02ecfa02cd08176adb6c17d04" + }, + "home_guard": "command exit0;157 paths and5 UID resources CLEAN", + "owned_server_reaped_and_fixture_removed": true +} diff --git a/docs/deck/curator.md b/docs/deck/curator.md new file mode 100644 index 0000000..d87994a --- /dev/null +++ b/docs/deck/curator.md @@ -0,0 +1,80 @@ +# Conditional curator source + +`publicationPlan` builds one existing SPARQL `/update` operation. A first +publication requires no current revision and revision 1. A later publication +requires the exact previous head and manifest and advances by exactly one. +The new immutable revision subject must have no facts. Both forms check the +canonical item's exact revision and open status, rejecting additional values +in either slot. The only DELETE/INSERT target is the request graph. Previous +manifests, answers and item facts are retained. + +`scripts/deck-curator-job.mjs` supplies `CuratorJob` with explicit injected +inspection, send and read-back adapters. It selects no endpoint, credential, +timer or production store. Before proceeding it checks the configured native +store lineage and retained graph-creation identity against inspection of a +registered committed graph. This separate inspection is not an atomic graph +replacement guard or proof of server-side credential restrictions. + +The job requires an existing owned directory with mode 0700 and no symlinked +path. It creates a mode-0600 custody file exclusively, containing the complete +plan, store lineage, graph pin and their digest. The file and directory are +synced before the one send. The custody slot is keyed by store, request graph, +request ID and revision; a different payload or graph pin conflicts. Partial, +corrupt, foreign-owned or symlinked records refuse. The directory must remain +on the job's durable volume across restarts; using a fresh directory defeats +that local custody guarantee. + +Every retained file stays UNKNOWN, including after successful confirmation. +Subsequent instances may only read it and reconcile; they never resend. A +lost response or read timeout therefore retains the exact publication intent. +`reconcile` can read an expired request's retained intent, while `publish` +refuses to initiate an expired publication. No automatic cleanup or retry is +provided. This is local job custody, not a cross-host exactly-once protocol. +Concurrent distinct publications are additionally controlled by the native +conditional update. + +A response must describe the exact request graph, one positive transaction, +two assertions and zero or one retraction. A zero-change report cannot be +mistaken for a changed transaction. CONFIRMED also requires complete exact +manifest/current-revision read-back from the pinned store. Read-only recovery +may confirm matching state without a transaction ID; it does not attribute +that state to this job or mint a signed receipt. Missing, foreign or changed +evidence remains UNKNOWN. No accepted decision or answered disposition is +created by publishing a request. + +Production adapters are still owed. They must pin the actual server/store, +native graph registration, bounded admitted reader definitions and cost, and +use the sanctioned authenticated write path. A dedicated named credential +must be restricted by the server's admitted policy to the request graph. +The generated query's narrow mutation target is not a security boundary for +a broad bearer. Declared actor/source fields are attribution, not verified +authentication. This source grants no credential, graph creation, protection, +registration, served enablement or enrollment authority. Production profile +union and external-writer/graph-replacement assumptions need admission too. + +## Repeatable qualification + +`just ci` covers query construction, strict transaction/read-back validation, +durable-before-send ordering, changed-payload refusal, restart/lost-response +reconciliation, concurrent job instances and unsafe custody. Negative target +controls reject missing/replaced/overlay graphs and wrong store lineage. + +The explicit qualifier starts one loopback native server with a new disposable +store, private HOME/configuration and a synthetic test bearer. It opens no +installed store and inherits no production credential or provider setting. +Its positive item facts control the graph-mutation checks. It runs the actual +generated update through HTTP, checks native reports and retained manifests, +races two distinct publications, and loses a reply after a real native commit +before read-only recovery by a new job instance. It reaps its child and removes +the fixture in `finally`, with a 90-second server bound. + +```sh +QUIPU_SERVER=/absolute/path/to/pinned/quipu-server \ +QUIPU_SERVER_SHA256= node scripts/qualify-deck-curator.mjs +``` + +[curator-native-results.json](curator-native-results.json) records the clean +release/full candidate and synthetic controls. That candidate is an explicit +source qualification artifact, not a claim about an installed or officially +released server. The fixture's broad synthetic bearer does not demonstrate +least-privilege credential enforcement, real enrollment or production cost. diff --git a/scripts/deck-curator-job.mjs b/scripts/deck-curator-job.mjs new file mode 100644 index 0000000..2e66544 --- /dev/null +++ b/scripts/deck-curator-job.mjs @@ -0,0 +1,69 @@ +// Source-only job, with explicit injected governed transport and reader. +// No endpoint, credential, timer or production adapter is selected here. +import {open, lstat, realpath} from 'node:fs/promises'; +import {constants} from 'node:fs'; +import {resolve, join} from 'node:path'; +import {createHash} from 'node:crypto'; +import {publicationPlan, publicationReport, publicationReadback} from '../web/deck/curator.mjs'; +import {Unknown, iri, text, exact, integer} from '../web/deck/value.mjs'; + +const hash = value => createHash('sha256').update(value).digest('hex'); +export class CuratorJob { + constructor({directory, storeId, registration, inspect, send, readBack}) { + if (typeof directory !== 'string' || typeof inspect !== 'function' || typeof send !== 'function' || typeof readBack !== 'function') throw new Unknown('Explicit custody and governed adapters required'); + exact(registration, ['iri', 'creation']); iri(registration.iri); + exact(registration.creation, ['actor', 'timestamp']); text(registration.creation.timestamp); + if (registration.creation.actor !== null) text(registration.creation.actor); + this.registration = structuredClone(registration); this.inspect = inspect; + this.directory = resolve(directory); this.storeId = iri(storeId); this.send = send; this.readBack = readBack; + } + async publish(input) {return this.run(input, false);} + async reconcile(input) {return this.run(input, true);} + async run(input, readOnly) { + const plan = publicationPlan(readOnly ? {...input, now: 0} : input), payload = JSON.stringify({storeId: this.storeId, registration: this.registration, plan}); + if (this.registration.iri !== plan.boundary.requestGraph) throw new Unknown('Foreign configured graph'); + const admitted = await this.inspect(this.registration.iri); + const graph = admitted.graph; integer(graph?.g); + if (admitted.storeId !== this.storeId || graph.iri !== this.registration.iri || graph.class !== 'committed' || JSON.stringify(graph.creation) !== JSON.stringify(this.registration.creation)) throw new Unknown('Unregistered or replaced publication target'); + const info = await lstat(this.directory); + if (!info.isDirectory() || info.uid !== process.getuid() || (info.mode & 0o777) !== 0o700 || await realpath(this.directory) !== this.directory) throw new Unknown('Owned private custody directory required'); + // Same request revision has ONE custody slot, including changed payloads. + const key = hash(JSON.stringify([this.storeId, plan.boundary.requestGraph, plan.next.id, plan.next.revision])); + const path = join(this.directory, key + '.json'); + const record = {state: 'UNKNOWN', digest: hash(payload), storeId: this.storeId, registration: this.registration, plan}; + const bytes = JSON.stringify(record) + '\n'; + if (Buffer.byteLength(bytes) > 131072) throw new Unknown('Publication custody too large'); + let fresh = false, handle; + try { + if (readOnly) {const error = new Error('Reconciliation requires retained custody'); error.code = 'EEXIST'; throw error;} + handle = await open(path, constants.O_WRONLY | constants.O_CREAT | constants.O_EXCL | constants.O_NOFOLLOW, 0o600); + await handle.writeFile(bytes); await handle.sync(); + await handle.close(); handle = null; + const dir = await open(this.directory, constants.O_RDONLY | constants.O_DIRECTORY); + try {await dir.sync();} finally {await dir.close();} + fresh = true; + } catch (error) { + if (handle) await handle.close(); + if (error.code !== 'EEXIST') throw error; + const retained = await open(path, constants.O_RDONLY | constants.O_NOFOLLOW); + let existing; + try { + const stat = await retained.stat(); + if (!stat.isFile() || stat.uid !== process.getuid() || (stat.mode & 0o777) !== 0o600 || stat.size > 131072) throw new Unknown('Invalid retained custody'); + existing = JSON.parse(await retained.readFile('utf8')); + } finally {await retained.close();} + if (existing.state !== 'UNKNOWN' || existing.digest !== record.digest || JSON.stringify({storeId: existing.storeId, registration: existing.registration, plan: existing.plan}) !== payload) throw new Unknown('Changed or incomplete publication custody'); + } + let report = null; + if (fresh) { + try {report = await this.send(plan.query); publicationReport(report, plan);} catch {return {state: 'UNKNOWN', sent: true, transaction: null};} + } + // Retained UNKNOWN never resends. Read-only reconciliation is repeatable. + try { + const observed = await this.readBack(plan.next.id); + if (observed.storeId !== this.storeId) throw new Unknown('Foreign store readback'); + publicationReadback(observed.requestRow, plan); + return {state: 'CONFIRMED', sent: fresh, transaction: report?.tx ?? null}; + } catch {return {state: 'UNKNOWN', sent: fresh, transaction: null};} + } +} diff --git a/scripts/demo-deck-curator.mjs b/scripts/demo-deck-curator.mjs new file mode 100644 index 0000000..7592cad --- /dev/null +++ b/scripts/demo-deck-curator.mjs @@ -0,0 +1,41 @@ +// Focused library demo with synthetic adapters, not a native or served proof. +import assert from 'node:assert/strict'; +import {mkdtemp, chmod, rm, readdir, readFile} from 'node:fs/promises'; +import {join} from 'node:path'; +import {tmpdir} from 'node:os'; +import {spawnSync, execFileSync} from 'node:child_process'; +import {CuratorJob} from './deck-curator-job.mjs'; +import {requestPublication, revisionIri} from '../web/deck/model.mjs'; +import {ask, boundary} from '../tests/deck-fixture.mjs'; + +const directory = await mkdtemp(join(tmpdir(), 'rikuy-curator-demo-')); await chmod(directory, 0o700); +const next = {...ask, revision: 1}, input = {boundary, next, previous: null, now: 1}; +const storeId = 'urn:uuid:synthetic-demo-store'; +const creation = {actor: 'synthetic-curator', timestamp: '2026-10-25T00:00:00Z'}; +let sends = 0, present = false; +const options = {directory, storeId, registration: {iri: boundary.requestGraph, creation}, inspect: async () => ({storeId, graph: {iri: boundary.requestGraph, g: 1, class: 'committed', creation}}), send: async () => { + const file = (await readdir(directory))[0]; + const saved = JSON.parse(await readFile(join(directory, file), 'utf8')); + console.log('Before send, durable custody:', saved.state); assert.equal(saved.state, 'UNKNOWN'); + sends++; present = true; throw Error('Synthetic lost reply after fixture mutation'); +}, readBack: async () => { + assert(present); return {storeId, requestRow: {graph: boundary.requestGraph, requestId: next.id, currentRevision: 1, revisionIri: revisionIri(next), manifest: requestPublication(next, boundary).datums[1].o.value, complete: true}}; +}}; +try { + const base = 'b290c515bfb3cda35eba38af3533c6c9200a0003'; + assert.equal(spawnSync('git', ['cat-file', '-e', base + ':web/deck/model.mjs'], {stdio: 'ignore'}).status, 0); + const before = spawnSync('git', ['cat-file', '-e', base + ':scripts/deck-curator-job.mjs'], {stdio: 'ignore'}).status; + assert.equal(before, 128); + console.log('Base', base, ': request fact builder exists; publication job absent.'); + console.log('Demonstrated source HEAD:', execFileSync('git', ['rev-parse', 'HEAD'], {encoding: 'utf8'}).trim()); + console.log('Target: source library, synthetic transport and reader; no service or credential.'); + console.log('Trigger: first publication; transport loses its reply after the fixture changes.'); + const lost = await new CuratorJob(options).publish(input); console.log('First call:', JSON.stringify(lost)); assert.equal(lost.state, 'UNKNOWN'); + console.log('Restart with the SAME durable directory; reconcile by reading.'); + const recovered = await new CuratorJob(options).reconcile(input); console.log('Recovery:', JSON.stringify(recovered)); assert.equal(recovered.state, 'CONFIRMED'); assert.equal(recovered.sent, false); + console.log('Send count:', sends); assert.equal(sends, 1); + let refused = false; + try {await new CuratorJob(options).publish({...input, next: {...next, title: 'Changed payload'}});} catch {refused = true;} + console.log('Changed payload in same revision refused:', refused); assert(refused); assert.equal(sends, 1); + console.log('Observed: UNKNOWN -> CONFIRMED by read-back; one send retained.'); +} finally {await rm(directory, {recursive: true, force: true});} diff --git a/scripts/qualify-deck-curator.mjs b/scripts/qualify-deck-curator.mjs new file mode 100644 index 0000000..1ba98a3 --- /dev/null +++ b/scripts/qualify-deck-curator.mjs @@ -0,0 +1,103 @@ +// Synthetic loopback qualification only. Requires an explicitly pinned binary; +// never opens an installed/default store or reads a real credential. +import assert from 'node:assert/strict'; +import {mkdtemp, mkdir, writeFile, readFile, rm} from 'node:fs/promises'; +import {tmpdir} from 'node:os'; +import {join, isAbsolute} from 'node:path'; +import {createHash} from 'node:crypto'; +import {spawn} from 'node:child_process'; +import {createServer} from 'node:net'; +import {once} from 'node:events'; +import {publicationPlan, publicationReport} from '../web/deck/curator.mjs'; +import {CuratorJob} from './deck-curator-job.mjs'; +import {NS} from '../web/deck/value.mjs'; +import {revisionIri, requestPublication} from '../web/deck/model.mjs'; + +const binary = process.env.QUIPU_SERVER, expectedHash = process.env.QUIPU_SERVER_SHA256; +assert(binary && isAbsolute(binary) && /^[0-9a-f]{64}$/.test(expectedHash || ''), 'Explicit candidate binary and SHA-256 required'); +const hash = createHash('sha256').update(await readFile(binary)).digest('hex'); assert.equal(hash, expectedHash); +const root = await mkdtemp(join(tmpdir(), 'rikuy-curator-native-')); +const socket = createServer(); socket.listen(0, '127.0.0.1'); await once(socket, 'listening'); const port = socket.address().port; await new Promise(resolve => socket.close(resolve)); +const endpoint = `http://127.0.0.1:${port}`, credential = 'synthetic-local-fixture-only'; +const boundary = {requestGraph: 'https://example.test/requests', answerGraph: 'https://example.test/answers', itemGraph: 'https://example.test/items', itemPrefix: 'https://example.test/item/', revisionPredicate: 'https://example.test/revision', statusPredicate: 'https://schema.org/actionStatus', openStatus: 'https://schema.org/ActiveActionStatus'}; +await mkdir(join(root, '.bobbin')); await mkdir(join(root, 'home')); +await writeFile(join(root, '.bobbin/config.toml'), `[quipu.server]\nbind = "127.0.0.1:${port}"\nauth_token = "${credential}"\nread_pool_size = 2\nread_processes = 0\nreserve_point_reads = false\nmemory_budget_mib = 256\n[quipu.embedding]\nauto_embed = false\n`, {mode: 0o600}); +let logs = '', ended = false; +const child = spawn(binary, ['--db', join(root, 'fixture.db'), '--bind', `127.0.0.1:${port}`], {cwd: root, env: {HOME: join(root, 'home'), PATH: '/usr/bin:/bin', LANG: 'C.UTF-8'}, stdio: ['ignore', 'pipe', 'pipe']}); +const exit = once(child, 'exit').then(([code, signal]) => {ended = true; return {code, signal};}); +child.stdout.on('data', data => {logs = (logs + data).slice(-16384);}); child.stderr.on('data', data => {logs = (logs + data).slice(-16384);}); +const budget = setTimeout(() => child.kill('SIGKILL'), 90000); budget.unref(); +const results = []; +async function http(path, body, type = 'application/json', authorized = true) { + const response = await fetch(endpoint + path, {method: body === undefined ? 'GET' : 'POST', headers: {'content-type': type, ...(authorized ? {authorization: `Bearer ${credential}`} : {}), 'X-Quipu-Client': 'agent-adhoc'}, body: body === undefined ? undefined : type === 'application/json' ? JSON.stringify(body) : body, signal: AbortSignal.timeout(3000), redirect: 'error'}); + const text = await response.text(); let value; try {value = JSON.parse(text);} catch {value = text;} + assert(response.ok, `HTTP ${response.status}: ${JSON.stringify(value)}`); return value; +} +const update = query => http('/update', query, 'application/sparql-update'); +const query = async sparql => {const value = await http('/query', {query: sparql}); assert.equal(value.truncated, false); assert.equal(value.rows.length, value.count); return value.rows;}; +const facts = graph => query(`SELECT ?s ?p ?o WHERE { GRAPH <${graph}> { ?s ?p ?o } } ORDER BY ?s ?p ?o LIMIT 100`); +const ask = id => ({id: 'https://example.test/request/' + id, revision: 1, seed: boundary.itemPrefix + 'one', itemRevision: 7, title: 'Synthetic curator request', options: ['Yes', 'No'], actions: ['choose', 'later', 'comment'], expiresAt: 2000000000}); +const plan = (next, previous = null) => publicationPlan({boundary, next, previous, now: 1}); +async function noChange(p, label) {const before = await facts(boundary.requestGraph); assert.equal(publicationReport(await update(p.query), p), 'NO_CHANGE'); assert.deepEqual(await facts(boundary.requestGraph), before); results.push({case: label, no_partial_request_mutation: true});} +try { + let ready = false; + for (let i = 0; i < 60; i++) {assert(!ended, logs); try {await http('/health'); ready = true; break;} catch {await new Promise(resolve => setTimeout(resolve, 100));}} + assert(ready, 'Candidate did not become ready'); + const version = await http('/version'); assert.equal(version.git_dirty, false); + for (const graph of Object.values(boundary).slice(0, 3)) await http('/graph/create', {graph}); + await update(`INSERT DATA { GRAPH <${boundary.itemGraph}> { <${boundary.itemPrefix}one> <${boundary.revisionPredicate}> 7 ; <${boundary.statusPredicate}> <${boundary.openStatus}> . } }`); + const itemBefore = await facts(boundary.itemGraph), answersBefore = await facts(boundary.answerGraph); + assert.equal(itemBefore.length, 2); assert.equal(answersBefore.length, 0); + const first = ask('normal'), initial = plan(first); + assert.equal(publicationReport(await update(initial.query), initial), 'CHANGED'); + await noChange(initial, 'identical-first-replay'); + const next = {...first, revision: 2, title: 'Quotes " and newline\nwith Unicode λ; } INSERT DATA { nope'}; + const advanced = plan(next, first); assert.equal(publicationReport(await update(advanced.query), advanced), 'CHANGED'); + const retained = await facts(boundary.requestGraph); + assert.equal(retained.filter(r => r.s === revisionIri(first)).length, 1); + assert.equal(retained.find(r => r.s === revisionIri(first)).o, requestPublication(first, boundary).datums[1].o.value); + results.push({case: 'initial-and-next', exact_old_manifest_retained: true, escaped_text_is_data: true}); + await noChange(plan({...next, title: 'Competing payload'}, first), 'stale-revision'); + for (const [name, delta] of [['changed-item', {itemRevision: 8}], ['foreign-item', {seed: boundary.itemPrefix + 'absent'}]]) await noChange(plan({...ask(name), ...delta}), name); + await update(`INSERT DATA { GRAPH <${boundary.itemGraph}> { <${first.seed}> <${boundary.revisionPredicate}> 8 . } }`); + await noChange(plan(ask('ambiguous-item')), 'ambiguous-item-revision'); + await update(`DELETE DATA { GRAPH <${boundary.itemGraph}> { <${first.seed}> <${boundary.revisionPredicate}> 8 . } }`); + await update(`INSERT DATA { GRAPH <${boundary.itemGraph}> { <${first.seed}> <${boundary.statusPredicate}> . } }`); + await noChange(plan(ask('ambiguous-status')), 'ambiguous-item-status'); + await update(`DELETE DATA { GRAPH <${boundary.itemGraph}> { <${first.seed}> <${boundary.statusPredicate}> <${boundary.openStatus}> . } }`); + await noChange(plan(ask('closed')), 'closed-item'); + await update(`DELETE { GRAPH <${boundary.itemGraph}> { <${first.seed}> <${boundary.statusPredicate}> . } } INSERT { GRAPH <${boundary.itemGraph}> { <${first.seed}> <${boundary.statusPredicate}> <${boundary.openStatus}> . } } WHERE {}`); + await update(`INSERT DATA { GRAPH <${boundary.requestGraph}> { <${first.id}> <${NS}currentRevision> 999 . } }`); + await noChange(plan({...next, revision: 3}, next), 'ambiguous-current-head'); + const occupied = ask('occupied'); await update(`INSERT DATA { GRAPH <${boundary.requestGraph}> { <${revisionIri(occupied)}> "occupied" . } }`); + await noChange(plan(occupied), 'occupied-immutable-revision'); + const race = ask('race'), race2 = {...race, title: 'Other curator'}; + const outcomes = await Promise.all([plan(race), plan(race2)].map(async p => publicationReport(await update(p.query), p))); + assert.deepEqual(outcomes.sort(), ['CHANGED', 'NO_CHANGE']); results.push({case: 'concurrent-different-payload', exactly_one_atomic_publication: true}); + const custody = join(root, 'custody'); await mkdir(custody, {mode: 0o700}); + const jobRequest = ask('lost-reply'), jobInput = {boundary, next: jobRequest, previous: null, now: 1}; + const storeId = (await http('/watermark')).store_id; let sends = 0; + const registered = (await http('/graphs?graph=' + encodeURIComponent(boundary.requestGraph))).graphs[0]; + const options = {directory: custody, storeId, registration: {iri: registered.iri, creation: registered.creation}, inspect: async graph => { + const [identity, metadata] = await Promise.all([http('/watermark'), http('/graphs?graph=' + encodeURIComponent(graph))]); + assert.equal(metadata.count, 1); assert.equal(metadata.graphs.length, 1); + return {storeId: identity.store_id, graph: metadata.graphs[0]}; + }, send: async text => {sends++; await update(text); throw Error('Synthetic dropped response after native commit');}, readBack: async id => { + const identity = (await http('/watermark')).store_id; + const rows = await query(`SELECT ?revision ?manifest WHERE { GRAPH <${boundary.requestGraph}> { <${id}> <${NS}currentRevision> ?revision . <${revisionIri(jobRequest)}> <${NS}requestManifest> ?manifest . } } LIMIT 3`); + assert.equal(rows.length, 1); + return {storeId: identity, requestRow: {graph: boundary.requestGraph, requestId: id, currentRevision: rows[0].revision, revisionIri: revisionIri(jobRequest), manifest: rows[0].manifest, complete: true}}; + }}; + assert.equal((await new CuratorJob(options).publish(jobInput)).state, 'UNKNOWN'); + const recovered = await new CuratorJob(options).reconcile({...jobInput, now: jobRequest.expiresAt + 1}); + assert.equal(recovered.state, 'CONFIRMED'); assert.equal(recovered.sent, false); assert.equal(sends, 1); + results.push({case: 'native-commit-lost-response-restart', read_only_recovery: true, sends}); + assert.deepEqual(await facts(boundary.itemGraph), itemBefore); assert.deepEqual(await facts(boundary.answerGraph), answersBefore); + results.push({case: 'graph-mutation-boundary', positive_item_control_unchanged: true, answer_graph_unchanged: true}); + console.log(JSON.stringify({candidate: version, candidate_sha256: hash, synthetic_loopback_only: true, real_credential_used: false, production_authority_proven: false, results}, null, 2)); +} catch (error) {console.error(logs); throw error;} +finally { + clearTimeout(budget); if (!ended) child.kill('SIGTERM'); + const escalation = setTimeout(() => {if (!ended) child.kill('SIGKILL');}, 3000); + await exit; clearTimeout(escalation); await rm(root, {recursive: true, force: true}); +} diff --git a/tests/deck-curator.test.mjs b/tests/deck-curator.test.mjs new file mode 100644 index 0000000..f80d5a8 --- /dev/null +++ b/tests/deck-curator.test.mjs @@ -0,0 +1,118 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {mkdtemp, rm, chmod, readdir, readFile, writeFile, symlink} from 'node:fs/promises'; +import {tmpdir} from 'node:os'; +import {join} from 'node:path'; +import {publicationPlan, publicationReport, publicationReadback} from '../web/deck/curator.mjs'; +import {CuratorJob} from '../scripts/deck-curator-job.mjs'; +import {requestPublication, revisionIri} from '../web/deck/model.mjs'; +import {ask, boundary} from './deck-fixture.mjs'; + +const first = {...ask, revision: 1}; +const input = {boundary, next: first, previous: null, now: 1}; +const row = r => ({graph: boundary.requestGraph, requestId: r.id, currentRevision: r.revision, revisionIri: revisionIri(r), manifest: requestPublication(r, boundary).datums[1].o.value, complete: true}); +const report = (previous = null) => ({tx: 7, asserted: 2, retracted: previous ? 1 : 0, graphs: [{graph: boundary.requestGraph, tx: 7, asserted: 2, retracted: previous ? 1 : 0}]}); +async function withJob(fn, adapters = {}) { + const directory = await mkdtemp(join(tmpdir(), 'rikuy-curator-')); await chmod(directory, 0o700); + let sends = 0; + const creation = {actor: 'fixture-curator', timestamp: '2026-10-25T00:00:00Z'}; + const settings = {directory, storeId: 'urn:uuid:fixture-store', registration: {iri: boundary.requestGraph, creation}, inspect: async () => ({storeId: 'urn:uuid:fixture-store', graph: {iri: boundary.requestGraph, g: 1, class: 'committed', creation}}), send: async () => {sends++; return report();}, readBack: async () => ({storeId: 'urn:uuid:fixture-store', requestRow: row(first)}), ...adapters}; + try {await fn({directory, settings, job: new CuratorJob(settings), sends: () => sends});} finally {await rm(directory, {recursive: true, force: true});} +} + +test('first and next publication use one graph-confined conditional update', () => { + const p = publicationPlan(input), advanced = publicationPlan({...input, next: {...first, revision: 2, title: 'Recurated'}, previous: first}); + assert(p.query.startsWith('DELETE { GRAPH <' + boundary.requestGraph + '> { } }')); + assert(p.query.includes('sameTerm(?otherItemRevision')); + assert(p.query.includes('sameTerm(?otherStatus')); + assert(p.query.includes('?occupiedPredicate ?occupiedValue')); + assert(advanced.query.includes('sameTerm(?otherHead')); + assert(advanced.query.includes('sameTerm(?otherManifest')); + assert(!p.query.includes(boundary.answerGraph)); + assert.equal(publicationReport(report(), p), 'CHANGED'); + assert.equal(publicationReport(report(first), advanced), 'CHANGED'); + assert.deepEqual(publicationReadback(row(first), p), first); +}); + +test('foreign, skipped, inexact, expired and reused revisions refuse before a send', () => { + for (const delta of [{next: {...first, revision: 2}}, {next: {...first, expiresAt: 1}}, {next: {...first, revision: 3}, previous: first}, {next: {...first, revision: 2}, previous: {...first, id: 'urn:other'}}, {next: {...first, revision: 2}, previous: {...first, seed: boundary.itemPrefix + 'other'}}, {now: NaN}]) assert.throws(() => publicationPlan({...input, ...delta})); +}); + +test('report must bind exact graph, single transaction and change counts', () => { + const p = publicationPlan(input); + assert.equal(publicationReport({tx: null, asserted: 0, retracted: 0, graphs: []}, p), 'NO_CHANGE'); + for (const mutate of [r => ({...r, asserted: 1}), r => ({...r, tx: null}), r => ({...r, graphs: []}), r => ({...r, graphs: [{...r.graphs[0], graph: boundary.answerGraph}]}), r => ({...r, graphs: [{...r.graphs[0], tx: 8}]}), r => ({...r, unknown: true})]) assert.throws(() => publicationReport(mutate(report()), p)); + for (const r of [{...first, title: 'Changed'}, {...first, revision: 2}]) assert.throws(() => publicationReadback(row(r), p)); +}); + +test('durable UNKNOWN is synced before one send and retained across new job instances', async () => { + await withJob(async ({directory, settings, sends}) => { + const original = settings.send; + settings.send = async query => { + const files = await readdir(directory); assert.equal(files.length, 1); + const retained = JSON.parse(await readFile(join(directory, files[0]), 'utf8')); + assert.equal(retained.state, 'UNKNOWN'); assert.equal(retained.plan.query, query); + return original(query); + }; + assert.deepEqual(await new CuratorJob(settings).publish(input), {state: 'CONFIRMED', sent: true, transaction: 7}); + assert.deepEqual(await new CuratorJob(settings).publish(input), {state: 'CONFIRMED', sent: false, transaction: null}); + assert.equal(sends(), 1); + await assert.rejects(() => new CuratorJob(settings).publish({...input, next: {...first, title: 'Conflicting payload'}})); + assert.equal(sends(), 1); + }); +}); + +test('lost response and read timeout recover by reads, never a second send', async () => { + let writes = 0, available = false; + await withJob(async ({settings}) => { + const job = new CuratorJob(settings); + assert.deepEqual(await job.publish(input), {state: 'UNKNOWN', sent: true, transaction: null}); + assert.deepEqual(await new CuratorJob(settings).reconcile(input), {state: 'UNKNOWN', sent: false, transaction: null}); + available = true; + assert.deepEqual(await new CuratorJob(settings).reconcile({...input, now: first.expiresAt + 1}), {state: 'CONFIRMED', sent: false, transaction: null}); + assert.equal(writes, 1); + }, {send: async () => {writes++; throw Error('Lost reply');}, readBack: async () => {if (!available) throw Error('Read timeout'); return {storeId: 'urn:uuid:fixture-store', requestRow: row(first)};}}); +}); + +test('concurrent publishers produce exactly one send', async () => { + await withJob(async ({settings, sends}) => { + await Promise.all(Array.from({length: 5}, async () => { + try {await new CuratorJob(settings).publish(input);} catch {} // A partially written record stays UNKNOWN. + })); + assert.equal(sends(), 1); + assert.equal((await new CuratorJob(settings).reconcile(input)).state, 'CONFIRMED'); + }); +}); + +test('zero-change, malformed reports, incomplete and foreign-store evidence do not authorize resend', async () => { + for (const adapters of [{send: async () => ({tx: null, asserted: 0, retracted: 0, graphs: []}), readBack: async () => {throw Error('No matching row');}}, {send: async () => ({tx: 8})}, {readBack: async () => ({storeId: 'urn:wrong', requestRow: row(first)})}, {readBack: async () => ({storeId: 'urn:uuid:fixture-store', requestRow: {...row(first), complete: false}})}]) { + await withJob(async ({job}) => {assert.equal((await job.publish(input)).state, 'UNKNOWN'); assert.equal((await job.reconcile(input)).sent, false);}, adapters); + } +}); + +test('unsafe custody modes, symlinks, corrupted and absent retained records refuse', async () => { + await withJob(async ({directory, settings, job, sends}) => { + await assert.rejects(() => job.reconcile(input)); assert.equal(sends(), 0); + await chmod(directory, 0o755); await assert.rejects(() => job.publish(input)); assert.equal(sends(), 0); + await chmod(directory, 0o700); await job.publish(input); + const path = join(directory, (await readdir(directory))[0]); await writeFile(path, '{broken'); + await assert.rejects(() => new CuratorJob(settings).publish(input)); assert.equal(sends(), 1); + await rm(path); await symlink(join(directory, 'missing'), path); + await assert.rejects(() => job.publish(input)); assert.equal(sends(), 1); + }); +}); + +test('native target inspection requires retained creation, committed graph and store identity', async () => { + for (const mutate of [v => ({...v, storeId: 'urn:other'}), v => ({...v, graph: null}), v => ({...v, graph: {...v.graph, g: 0}}), v => ({...v, graph: {...v.graph, class: 'overlay'}}), v => ({...v, graph: {...v.graph, iri: boundary.answerGraph}}), v => ({...v, graph: {...v.graph, creation: {...v.graph.creation, timestamp: 'replaced'}}})]) { + await withJob(async ({settings, sends}) => {const inspect = settings.inspect; settings.inspect = async (...args) => mutate(await inspect(...args)); await assert.rejects(() => new CuratorJob(settings).publish(input)); assert.equal(sends(), 0);}); + } +}); + +test('replacing the retained graph pin cannot reuse an existing custody slot', async () => { + await withJob(async ({settings, job, sends}) => { + await job.publish(input); + const registration = {...settings.registration, creation: {...settings.registration.creation, timestamp: '2026-10-26T00:00:00Z'}}; + const changed = new CuratorJob({...settings, registration, inspect: async () => ({storeId: settings.storeId, graph: {iri: registration.iri, g: 2, class: 'committed', creation: registration.creation}})}); + await assert.rejects(() => changed.reconcile(input)); assert.equal(sends(), 1); + }); +}); diff --git a/web/deck/curator.mjs b/web/deck/curator.mjs new file mode 100644 index 0000000..464c80c --- /dev/null +++ b/web/deck/curator.mjs @@ -0,0 +1,52 @@ +import {request, requestPublication, readRequest, revisionIri, scope} from './model.mjs'; +import {Unknown, NS, exact, integer, utf8} from './value.mjs'; + +const term = value => `<${value}>`; +const literal = value => JSON.stringify(value); +const number = value => `"${value}"^^`; +const slot = (s, p, o) => `${term(s)} ${term(p)} ${o} .`; +function unique(s, p, expected, variable) { + return `${slot(s, p, expected)} FILTER NOT EXISTS { ${term(s)} ${term(p)} ?${variable} . FILTER (!sameTerm(?${variable}, ${expected})) }`; +} + +// One existing /update operation, with writes confined to the request graph. +// This builder neither registers a graph nor grants the caller write authority. +export function publicationPlan({boundary, next, previous, now}) { + const b = scope(boundary), r = request(next, b); + if (!Number.isSafeInteger(now) || now < 0 || now >= r.expiresAt) throw new Unknown('Expired publication'); + const old = previous === null ? null : request(previous, b); + if (old ? old.id !== r.id || old.seed !== r.seed || old.revision === Number.MAX_SAFE_INTEGER || r.revision !== old.revision + 1 : r.revision !== 1) throw new Unknown('Publication must advance exactly one revision'); + const pub = requestPublication(r, b), current = NS + 'currentRevision', manifest = NS + 'requestManifest'; + const removed = old ? slot(r.id, current, number(old.revision)) : ''; + const head = old ? unique(r.id, current, number(old.revision), 'otherHead') : `FILTER NOT EXISTS { ${term(r.id)} ${term(current)} ?head . }`; + const retained = old ? unique(revisionIri(old), manifest, literal(requestPublication(old, b).datums[1].o.value), 'otherManifest') : ''; + const inserted = `${slot(r.id, current, number(r.revision))}\n${slot(revisionIri(r), manifest, literal(pub.datums[1].o.value))}`; + const query = `DELETE { GRAPH ${term(b.requestGraph)} { ${removed} } } +INSERT { GRAPH ${term(b.requestGraph)} { ${inserted} } } +WHERE { + GRAPH ${term(b.itemGraph)} { + ${unique(r.seed, b.revisionPredicate, number(r.itemRevision), 'otherItemRevision')} + ${unique(r.seed, b.statusPredicate, term(b.openStatus), 'otherStatus')} + } + ${old ? `GRAPH ${term(b.requestGraph)} { ${head} ${retained} }` : `FILTER NOT EXISTS { GRAPH ${term(b.requestGraph)} { ${term(r.id)} ${term(current)} ?head . } }`} + FILTER NOT EXISTS { GRAPH ${term(b.requestGraph)} { ${term(revisionIri(r))} ?occupiedPredicate ?occupiedValue . } } +}`; + if (utf8(query).length > 65536) throw new Unknown('Publication update too large'); + return {boundary: b, next: r, previous: old, query}; +} + +export function publicationReport(value, plan) { + exact(value, ['tx', 'asserted', 'retracted', 'graphs']); + if (value.tx === null && value.asserted === 0 && value.retracted === 0 && Array.isArray(value.graphs) && value.graphs.length === 0) return 'NO_CHANGE'; + integer(value.tx); + if (value.asserted !== 2 || value.retracted !== (plan.previous === null ? 0 : 1) || !Array.isArray(value.graphs) || value.graphs.length !== 1) throw new Unknown('Unexpected publication changes'); + const g = value.graphs[0]; exact(g, ['graph', 'tx', 'asserted', 'retracted']); + if (g.graph !== plan.boundary.requestGraph || g.tx !== value.tx || g.asserted !== value.asserted || g.retracted !== value.retracted) throw new Unknown('Foreign publication transaction'); + return 'CHANGED'; +} + +export function publicationReadback(row, plan) { + const observed = readRequest(row, plan.boundary); + if (JSON.stringify(observed) !== JSON.stringify(plan.next)) throw new Unknown('Publication not confirmed'); + return observed; +}