From 88c83b4d18d458399599bf1b0b1a8d24d383023f Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 3 Nov 2026 15:10:06 -0600 Subject: [PATCH] Preserve live ingestion acceptance and independent database verification --- scripts/native-ingestion/README-live.md | 148 +++++++++++ scripts/native-ingestion/live-browser.mjs | 171 ++++++++++++ scripts/native-ingestion/live-launch.mjs | 61 +++++ scripts/native-ingestion/verify-database.mjs | 262 +++++++++++++++++++ 4 files changed, 642 insertions(+) create mode 100644 scripts/native-ingestion/README-live.md create mode 100644 scripts/native-ingestion/live-browser.mjs create mode 100644 scripts/native-ingestion/live-launch.mjs create mode 100644 scripts/native-ingestion/verify-database.mjs diff --git a/scripts/native-ingestion/README-live.md b/scripts/native-ingestion/README-live.md new file mode 100644 index 00000000000..1d7f74862d3 --- /dev/null +++ b/scripts/native-ingestion/README-live.md @@ -0,0 +1,148 @@ +# Live-provider ingestion acceptance + +These helpers accompany the separate live-provider functional journey. They do +not generate performance acceptance measurements. The deterministic paired +experiment and profile-only tools are documented in the shared `README.md`. + +## Exact helper provenance + +The following files are byte-for-byte copies of the successful November 3, 2026 +helpers. Paths in this table describe their original temporary locations. + +| File | Original in `/private/tmp/` | SHA256 | +| --- | --- | --- | +| `live-launch.mjs` | `native-ingestion-live-launch.mjs` | `353a1b1cf2ee38551349617e91c95cdfb4d0021abf55be62e7058319b8e5b6bd` | +| `live-browser.mjs` | `native-ingestion-live-browser.mjs` | `cf34b0f7bb8a4f84786a38f1c90b750e7a336b126e6e2bc25647095c1eac9fab` | + +The successful environment was macOS arm64, Node 26.4.0, compiled revision +`6afeba4edafd47120e8117f736641d5424b27d4a`, local embedded PostgreSQL, +OpenCode 1.18.34, and `openrouter/deepseek/deepseek-v4-flash-0731`. The launcher +deliberately retains its macOS `/private/tmp/` data-directory prefix and that +provider/model selection. It is not a qualified portable launcher. It reads +`PAPERCLIP_PI_QUALIFICATION_OPENROUTER_API_KEY` from the environment and provisions +an encrypted instance secret through the normal test-drive API bootstrap. +Supply the credential privately; do not put it in arguments, logs or evidence. + +## Repeating the functional journey + +Use an independently installed and normally built application checkout. Its +source revision and compiled build stamp must equal the explicit expected +revision. The launcher uses the normal onboarding configuration boundary and +explicitly starts `server/dist/index.js` with the shipped static UI. It imports +source CLI configuration/bootstrap helpers through the existing TSX loader; it +does not start the source server or Vite. + +Choose a new owned evidence directory. `TOOLS_CHECKOUT` contains these helpers; +`APP_CHECKOUT` contains the compiled application being tested. In the first +terminal: + +```sh +node --import "$APP_CHECKOUT/server/node_modules/tsx/dist/loader.mjs" \ + "$TOOLS_CHECKOUT/scripts/native-ingestion/live-launch.mjs" \ + "$APP_CHECKOUT" "$EXPECTED_REVISION" "$LIVE_EVIDENCE_DIR" +``` + +Wait for its ready receipt. In a second terminal: + +```sh +node --import "$APP_CHECKOUT/server/node_modules/tsx/dist/loader.mjs" \ + "$TOOLS_CHECKOUT/scripts/native-ingestion/live-browser.mjs" \ + "$LIVE_EVIDENCE_DIR/live-launch.json" +``` + +The browser uses ordinary visible controls to create a task, requires an +ordinary stored delta to match visible text while the run is running, observes +a native tool/result, reloads, and independently reads the run events through +the API. A second task clicks Stop during a running `bash` command, requires +cancellation acknowledgement, and reloads to check settlement. Trace, video +and screenshots are intentional functional evidence here; none enter the paired +performance measurements. + +While the owned app/database remain available, run the independent verifier +below. Afterwards, gracefully stop only the launcher process, confirm its HTTP +and database ports and provider processes are gone, and remove only its recorded +credential-bearing `dataDir`. Keep cleanup results separately. The exact copied +launcher retains the directory, so process exit alone is not data cleanup. +Before publishing retained browser artifacts, scan the files and decompressed +trace archive for the actual qualification credential without printing it. + +The historical successful journey verified success run +`212cf2b8-8e8f-42e0-9fa3-4139d84d4e11` and Stop run +`9607febf-864e-4ddd-b82a-95ea4420821e`; both temporary databases are now removed. +Its direct database proof recorded 105/73 global rows and 84 canonical source +digests. The observed 1,978 ms Stop acknowledgement is one functional sample. +The earlier observer failure remains excluded and retained in the task evidence. + +## Reconstructed independent database verifier + +`verify-database.mjs` is a new, independently implemented reproducible utility. +It is not the historical command that produced those 84 digest checks and cannot +retroactively verify a deleted database. It imports only the raw `postgres` +dependency resolved from the supplied repository. No application service, +Drizzle schema, production payload decoder or digest helper is imported. + +Supply `PAPERCLIP_NATIVE_INGESTION_VERIFY_DATABASE_URL` privately with a bare +`postgres://` or `postgresql://` URL for the disposable loopback database. Only +`127.0.0.1`, `localhost` and `::1` are accepted. Query-string/fragment overrides +are rejected, preventing a URL option from replacing that host. The URL is +never printed. The utility expects the additive cursor schema to be installed. + +```sh +node "$TOOLS_CHECKOUT/scripts/native-ingestion/verify-database.mjs" \ + --repository="$APP_CHECKOUT" --company-id="$COMPANY_ID" \ + --run-id="$SUCCESS_RUN_ID" --run-id="$STOP_RUN_ID" \ + --require-current-projection --output="$LIVE_EVIDENCE_DIR/independent-database-proof-new.json" +``` + +All SQL is parameterized and scoped to the explicit company and run bindings. +The utility opens one `READ ONLY`, `REPEATABLE READ` transaction and confirms +those settings before reading. It reads the raw JSONB representation with +`payload::text`, independently restores the lossless reserved envelope, and +recomputes each PRP event's sorted-key canonical SHA256. It checks source/global +identity uniqueness, column/envelope agreement, run/agent/company binding, and +the complete global sequence allocation. It does not use API-returned objects +as its database evidence. + +When the projection marker equals `next_event_seq - 1`, the utility reconstructs +each source's contiguous prefix and disjoint received ranges from stored positive +source sequences and compares both derived tables. With a stale marker it skips +both projection queries/reconstruction and reports `projectionChecked: false`; +`--require-current-projection` makes that situation fail. A sparse Stop source +may correctly retain a gap and a higher pending range. That does not imply all +raw wire-acknowledged output was durably committed. + +This utility expects the complete history of a disposable fixture. Retention or +manual deletion can legitimately remove historical rows and makes this complete +fixture proof unsuitable. It reads all selected event rows within one snapshot; +it is a verification tool, not a bounded ingestion worker or a performance test. +Output contains counts, bindings and cursor/range summaries, not raw transcript +payloads or connection credentials. Successful output is emitted only after +the database connection is closed; an output file is created exclusively with +mode `0600` and never overwritten. Controlled validation codes are printed on +failure; unexpected driver error details are suppressed to protect connection +information. + +The embedded self-test exercises independently specified canonical digest +fixtures, NUL/reserved-key restoration, numeric object-key ordering, multiple +sparse sources, stale-marker behavior, identity conflicts and projection +conflicts without opening a database: + +```sh +node "$TOOLS_CHECKOUT/scripts/native-ingestion/verify-database.mjs" --self-test +``` + +Its self-test and syntax checks passed during packaging. The actual CLI was also +qualified against an owned minimal PostgreSQL 18.1 fixture: five global rows, +four losslessly encoded canonical events, and two sparse sources verified; a +persisted bad digest and a persisted bad cursor each caused the expected +failure. PostgreSQL's own log recorded `BEGIN ISOLATION LEVEL REPEATABLE READ +READ ONLY`. Restoring the fixture passed again. The cluster was stopped, its +port independently verified closed, and its data directory removed. + +The initial synthetic fixture accidentally passed an already serialized JSON +string to an inferred JSONB parameter and the verifier rejected it as +`payload_not_an_object`. That failed receipt remains retained. Correcting the +private fixture to use the raw driver's explicit JSON object wrapper produced +the successful qualification without changing verifier code. This is fixture +qualification evidence, not a repeated live-provider journey. Every future real +database invocation must retain its own revision/environment/cleanup receipt. diff --git a/scripts/native-ingestion/live-browser.mjs b/scripts/native-ingestion/live-browser.mjs new file mode 100644 index 00000000000..e446a1b31a8 --- /dev/null +++ b/scripts/native-ingestion/live-browser.mjs @@ -0,0 +1,171 @@ +import { createRequire } from "node:module"; +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { pathToFileURL } from "node:url"; + +// Private acceptance helper. Configuration contains only an owned loopback +// origin, company/agent IDs and evidence paths; never provider credentials. +const config = JSON.parse(await readFile(process.argv[2], "utf8")); +const origin = new URL(config.origin); +if (!['127.0.0.1', 'localhost', '[::1]'].includes(origin.hostname)) throw new Error("Owned loopback application required"); +const require = createRequire(join(config.repository, "package.json")); +const { chromium, expect } = require("@playwright/test"); +const { createTaskThroughUi } = await import(pathToFileURL(join(config.repository, "tests/runner-e2e/user-actions.ts")).href); +await mkdir(config.evidenceDir, { recursive: true }); +const evidence = { startedAt: new Date().toISOString(), config, journeys: [], status: "running" }; +const save = () => writeFile(join(config.evidenceDir, "live-provider-result.json"), JSON.stringify(evidence, null, 2)); +const api = async route => { + const response = await fetch(new URL(route, origin)); + if (!response.ok) throw new Error(`Read ${route}: HTTP ${response.status}`); + return response.json(); +}; +const pause = ms => new Promise(resolve => setTimeout(resolve, ms)); +async function poll(label, load, accept, deadlineMs = 180_000) { + const deadline = Date.now() + deadlineMs; + while (true) { + const value = await load(); + if (accept(value)) return value; + if (Date.now() >= deadline) throw new Error(`Timed out waiting for ${label}`); + await pause(250); + } +} +async function eventsFor(runId) { + const result = []; + while (true) { + const page = await api(`/api/heartbeat-runs/${runId}/events?afterSeq=${result.at(-1)?.seq ?? 0}&limit=1000`); + result.push(...page); + if (page.length < 1000) return result; + } +} +const runSummary = run => ({ id: run.id, status: run.status, runtimeMode: run.runtimeMode, + driverKind: run.driverKind, providerSessionId: run.providerSessionId ?? run.sessionIdAfter ?? run.runnerProfileJson?.sessionCheckpoint?.providerSessionId, + nativeSessionId: run.nativeSessionId, + startedAt: run.startedAt, finishedAt: run.finishedAt, error: run.error, + nativePhase: run.nativePhase, provider: run.runnerProfileJson?.nativeExecutionInput?.provider ?? run.runnerProfileJson?.provider, + cancellation: run.resultJson?.nativeCancellation }); +async function taskRuns(issueId) { + const rows = await api(`/api/companies/${config.company.id}/heartbeat-runs?agentId=${config.agent.id}&limit=100`); + const details = await Promise.all(rows.map(run => api(`/api/heartbeat-runs/${run.id}`))); + return details.filter(run => run.nativeIssueId === issueId || run.contextSnapshot?.issueId === issueId); +} +const browser = await chromium.launch({ headless: true }); +const context = await browser.newContext({ viewport: { width: 1440, height: 1000 }, + recordVideo: { dir: join(config.evidenceDir, "video"), size: { width: 1440, height: 1000 } } }); +await context.tracing.start({ screenshots: true, snapshots: true }); +const page = await context.newPage(); +const visibleAgentText = async () => (await page.locator('[data-conversation-message="agent"] [data-conversation-text]:visible, [data-testid="task-chat-live-self-talk"]:visible').allInnerTexts()).join("\n"); +const normalizeVisible = value => String(value).replace(/[*_`#]/g, "").replace(/\s+/g, " ").trim(); +let activeIssueId; +try { + const health = await api("/api/health"); + const agent = await api(`/api/agents/${config.agent.id}`); + if (agent.companyId !== config.company.id || agent.adapterType !== "paperclip_runner" || agent.adapterConfig?.provider !== "opencode") + throw new Error("Configured agent is not the owned native OpenCode agent"); + evidence.health = { status: health.status, gitCommit: health.gitCommit ?? health.commit ?? health.version }; + evidence.provider = { adapterType: agent.adapterType, provider: agent.adapterConfig.provider, model: agent.adapterConfig.model }; + const marker = `LIVE-INGESTION-${Date.now()}`; + const journey = { kind: "stream-tool-result-reload", marker, status: "running" }; + evidence.journeys.push(journey); + const created = await createTaskThroughUi({ page, origin: origin.href, issuePrefix: config.company.issuePrefix, + agentName: config.agent.name, title: `Native live ingestion ${marker}`, workMode: "standard", requireExplicitTitle: true, + prompt: `Verify this isolated test task. First send a short progress message with ${marker}-START. Use the native Paperclip list_task_documents tool to inspect this task's documents and report the count. Then provide twelve numbered sentences about what this check proves. Include ${marker}-DONE in the final response and finish the task. Do not modify files or access unrelated tasks or services.` }); + activeIssueId = created.issueId; journey.issueId = created.issueId; + await page.goto(new URL(`/${config.company.issuePrefix}/issues/${created.issueId}`, origin).href); + const streaming = await poll("ordinary assistant delta visibly rendered before terminal", async () => { + const runs = await taskRuns(created.issueId); + const active = runs.find(run => run.status === "running"); + if (!active) return null; + const text = normalizeVisible(await visibleAgentText()); + const rows = await eventsFor(active.id); + const delta = rows.map(row => row.payload?.prpEvent).find(event => event?.eventType === "item.delta" + && event.payload.kind === "agentMessage" && normalizeVisible(event.payload.text).length >= 12 + && text.includes(normalizeVisible(event.payload.text))); + if (!delta) return null; + // Re-read status after observing the matching UI text. A completed run + // discovered on the next poll cannot establish streaming visibility. + if ((await api(`/api/heartbeat-runs/${active.id}`)).status !== "running") return null; + return { runId: active.id, sourceEventId: delta.sourceEventId, sourceSeq: delta.sourceSeq, + emittedAt: delta.emittedAt, visibleText: text.slice(-1000) }; + }, Boolean); + journey.visibleDelta = streaming; + journey.firstVisibleAt = Date.now(); + const during = await taskRuns(created.issueId); + journey.streamingRunStatuses = during.map(runSummary); + journey.visibleBeforeTerminal = during.some(run => ["queued", "running"].includes(run.status)); + expect(journey.visibleBeforeTerminal).toBe(true); + await page.screenshot({ path: join(config.evidenceDir, "live-streaming.png"), fullPage: true }); + const finished = await poll("completed native run", () => taskRuns(created.issueId), runs => + runs.length > 0 && runs.every(run => !["queued", "running", "scheduled_retry"].includes(run.status))); + journey.runs = finished.map(runSummary); + if (finished.length !== 1 || finished[0].status !== "succeeded" || finished[0].runtimeMode !== "native") + throw new Error(`Live run did not succeed: ${JSON.stringify(journey.runs)}`); + const rows = await eventsFor(finished[0].id); + const prp = rows.map(row => row.payload?.prpEvent).filter(Boolean); + journey.eventCounts = Object.fromEntries([...new Set(prp.map(event => event.eventType))].map(type => [type, prp.filter(event => event.eventType === type).length])); + journey.toolEvidence = prp.filter(event => event.eventType.startsWith("tool.") + && JSON.stringify(event.payload).includes("list_task_documents")) + .map(event => ({ eventType: event.eventType, sourceEventId: event.sourceEventId, sourceSeq: event.sourceSeq, + name: event.payload.name, operation: event.payload.operation, status: event.payload.status })); + expect(prp.some(event => event.eventType === "item.delta" && event.payload.kind === "agentMessage")).toBe(true); + expect(journey.toolEvidence.some(event => event.eventType === "tool.execution.completed")).toBe(true); + expect(prp.some(event => event.eventType === "run.result.accepted")).toBe(true); + await expect.poll(visibleAgentText, { timeout: 30_000 }).toContain(`${marker}-DONE`); + await page.screenshot({ path: join(config.evidenceDir, "live-result.png"), fullPage: true }); + await page.reload({ waitUntil: "domcontentloaded" }); + await expect.poll(visibleAgentText, { timeout: 30_000 }).toContain(`${marker}-DONE`); + await page.screenshot({ path: join(config.evidenceDir, "live-reloaded.png"), fullPage: true }); + journey.persistedAfterReload = true; journey.status = "passed"; + activeIssueId = undefined; + await save(); + + const stopped = { kind: "stop-live-provider", status: "running" }; + evidence.journeys.push(stopped); + const stopTask = await createTaskThroughUi({ page, origin: origin.href, issuePrefix: config.company.issuePrefix, + agentName: config.agent.name, title: `Native live Stop ${marker}`, workMode: "standard", requireExplicitTitle: true, + prompt: `This isolated task verifies the Stop control. First send a short progress message containing ${marker}-STOP-START. Then use your native shell tool to run exactly sleep 60 in the task workspace. Wait for the command and report only after it finishes. Do not modify files or use unrelated resources.` }); + activeIssueId = stopTask.issueId; stopped.issueId = stopTask.issueId; + await page.goto(new URL(`/${config.company.issuePrefix}/issues/${stopTask.issueId}`, origin).href); + const active = await poll("live provider tool execution", async () => { + const runs = await taskRuns(stopTask.issueId); + if (runs.length !== 1 || runs[0].status !== "running") return null; + const events = await eventsFor(runs[0].id); + const started = events.map(row => row.payload?.prpEvent).filter(event => event?.eventType === "tool.execution.started"); + stopped.startedTools = started.map(event => ({ name: event.payload.name, operation: event.payload.operation, + sourceEventId: event.sourceEventId })); + return started.some(event => /shell|bash|execute/i.test(String(event.payload.name) + String(event.payload.operation))) ? runs[0] : null; + }, Boolean); + stopped.activeRunId = active.id; + await page.screenshot({ path: join(config.evidenceDir, "before-live-stop.png"), fullPage: true }); + stopped.requestedAt = Date.now(); + await page.getByTestId("task-chat-composer-stop").last().click(); + const afterStop = await poll("native cancellation acknowledgement", async () => { + const runs = await taskRuns(stopTask.issueId); + return runs.length === 1 ? await api(`/api/heartbeat-runs/${runs[0].id}`) : null; + }, run => run?.status === "cancelled" && run.resultJson?.nativeCancellation?.dispatchState === "acknowledged"); + stopped.acknowledgedAt = Date.now(); stopped.run = runSummary(afterStop); + expect(afterStop.id).toBe(active.id); + await page.reload({ waitUntil: "domcontentloaded" }); + await pause(1000); + stopped.runIdsAfterReload = (await taskRuns(stopTask.issueId)).map(run => run.id); + expect(stopped.runIdsAfterReload).toEqual([active.id]); + await page.screenshot({ path: join(config.evidenceDir, "after-live-stop.png"), fullPage: true }); + stopped.status = "passed"; activeIssueId = undefined; + evidence.status = "passed"; +} catch (error) { + evidence.status = "failed"; + evidence.error = error instanceof Error ? { name: error.name, message: error.message, stack: error.stack } : String(error); + await page.screenshot({ path: join(config.evidenceDir, "live-failure.png"), fullPage: true }).catch(() => undefined); + if (activeIssueId) { + // Cleanup uses the same visible Stop action; failure remains in evidence. + const stop = page.getByTestId("task-chat-composer-stop").last(); + if (await stop.isVisible().catch(() => false)) await stop.click().catch(() => undefined); + } +} finally { + evidence.finishedAt = new Date().toISOString(); + await save(); + await context.tracing.stop({ path: join(config.evidenceDir, "live-provider-trace.zip") }); + await context.close(); await browser.close(); +} +console.log(JSON.stringify({ status: evidence.status, evidenceDir: config.evidenceDir, journeys: evidence.journeys.map(journey => + ({ kind: journey.kind, status: journey.status, issueId: journey.issueId, runs: journey.runs ?? journey.run })) })); +if (evidence.status !== "passed") process.exitCode = 1; diff --git a/scripts/native-ingestion/live-launch.mjs b/scripts/native-ingestion/live-launch.mjs new file mode 100644 index 00000000000..200358b8e5c --- /dev/null +++ b/scripts/native-ingestion/live-launch.mjs @@ -0,0 +1,61 @@ +import { mkdtemp, mkdir, writeFile, readFile } from "node:fs/promises"; +import { join } from "node:path"; +import { pathToFileURL } from "node:url"; +import { createHash } from "node:crypto"; +import { execFileSync } from "node:child_process"; + +const repository = process.argv[2]; +const expectedRevision = process.argv[3]; +const evidenceDir = process.argv[4]; +if (!repository || !expectedRevision || !evidenceDir) throw new Error("repository, expected revision and evidence directory are required"); +const revision = execFileSync("git", ["rev-parse", "HEAD"], { cwd: repository, encoding: "utf8" }).trim(); +if (revision !== expectedRevision) throw new Error("Source revision changed before live acceptance"); +const buildStamp = JSON.parse(await readFile(join(repository, "server/dist/build-info.json"), "utf8")); +if (buildStamp.commit !== expectedRevision) throw new Error("Compiled server revision does not match live acceptance source"); +const credentialName = "PAPERCLIP_PI_QUALIFICATION_OPENROUTER_API_KEY"; +if (!process.env[credentialName]) throw new Error("Qualification credential is unavailable"); +const model = "openrouter/deepseek/deepseek-v4-flash-0731"; +await mkdir(evidenceDir, { recursive: true }); +const dataDir = await mkdtemp("/private/tmp/paperclip-native-live-data-"); +let server; +try { + const { prepareTestDriveEnvironment, bootstrapTestDrive } = await import(pathToFileURL(join(repository, "cli/src/commands/test-drive.ts")).href); + const options = { dataDir, companyName: "Native ingestion live acceptance", agentName: "Live ingestion agent", + harness: "opencode", model, apiKeyEnv: credentialName, browser: false }; + await prepareTestDriveEnvironment(options, repository); + process.env.PAPERCLIP_UI_DEV_MIDDLEWARE = "false"; + process.env.NODE_ENV = "production"; + process.env.SERVE_UI = "true"; + const { onboard } = await import(pathToFileURL(join(repository, "cli/src/commands/onboard.ts")).href); + await onboard({ config: process.env.PAPERCLIP_CONFIG, yes: true, bind: "loopback", invokedByRun: true, installService: false }); + const { doctor } = await import(pathToFileURL(join(repository, "cli/src/commands/doctor.ts")).href); + const checked = await doctor({ config: process.env.PAPERCLIP_CONFIG, repair: true, yes: true }); + if (checked.failed > 0) throw new Error(`Doctor failed ${checked.failed} checks`); + const { startServer } = await import(pathToFileURL(join(repository, "server/dist/index.js")).href); + server = await startServer(); + const origin = `http://${server.host}:${server.listenPort}`; + const { PaperclipApiClient } = await import(pathToFileURL(join(repository, "cli/src/client/http.ts")).href); + const api = new PaperclipApiClient({ apiBase: origin }); + const fixture = await bootstrapTestDrive({ api, options, linkedWorktree: true, instanceId: "default", env: process.env }); + if (fixture.reused || !fixture.agent) throw new Error("Fresh live-provider fixture was not created"); + await api.patch("/api/instance/settings/experimental", { enableNativeRunner: true }); + const agent = await api.patch(`/api/agents/${fixture.agent.id}`, { + adapterType: "paperclip_runner", + adapterConfig: { ...fixture.agent.adapterConfig, provider: "opencode", model, lifecycleMode: "per_turn", + opencodePermissionMode: "allow", timeoutSec: 180 }, + runtimeConfig: { heartbeat: { enabled: false } }, + }); + if (agent?.adapterType !== "paperclip_runner" || agent.adapterConfig?.provider !== "opencode") throw new Error("Native agent setup did not persist"); + const config = { repository, origin, company: { id: fixture.company.id, issuePrefix: fixture.company.issuePrefix }, + agent: { id: agent.id, name: agent.name }, evidenceDir, revision, dataDir, pid: process.pid, + build: { commit: buildStamp.commit, serverSha256: createHash("sha256").update(await readFile(join(repository, "server/dist/index.js"))).digest("hex"), + uiSha256: createHash("sha256").update(await readFile(join(repository, "ui/dist/index.html"))).digest("hex") }, + platform: { node: process.version, os: process.platform, architecture: process.arch }, provider: { kind: "opencode", model } }; + await writeFile(join(evidenceDir, "live-launch.json"), JSON.stringify(config, null, 2)); + console.log(JSON.stringify({ ready: true, origin, revision, evidenceDir, configPath: join(evidenceDir, "live-launch.json"), pid: process.pid })); +} catch (error) { + await server?.shutdown?.("SIGTERM"); + const message = error instanceof Error ? error.stack ?? error.message : String(error); + console.error(message.replaceAll(process.env[credentialName] ?? "\0private-credential-unavailable", "***REDACTED***")); + process.exitCode = 1; +} diff --git a/scripts/native-ingestion/verify-database.mjs b/scripts/native-ingestion/verify-database.mjs new file mode 100644 index 00000000000..7c2b5335e0e --- /dev/null +++ b/scripts/native-ingestion/verify-database.mjs @@ -0,0 +1,262 @@ +import assert from 'node:assert/strict'; +import { createHash } from 'node:crypto'; +import { createRequire } from 'node:module'; +import { writeFile } from 'node:fs/promises'; +import path from 'node:path'; +import { pathToFileURL } from 'node:url'; + +// Reconstructed independent verifier, not the historical live-proof command. +// No application service, Drizzle schema, payload decoder or digest helper is imported. +const DATABASE_ENV = 'PAPERCLIP_NATIVE_INGESTION_VERIFY_DATABASE_URL'; +const ORIGINAL_JSON_KEY = '$paperclipRunEventJsonV1'; +const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; +class VerificationError extends Error { + constructor(code) { super(code); this.name = 'VerificationError'; this.code = code; } +} +const requireCondition = (value, code) => { if (!value) throw new VerificationError(code); }; +const object = value => value !== null && typeof value === 'object' && !Array.isArray(value); +function integer(value, code, minimum = 0) { + requireCondition(typeof value === 'number' || typeof value === 'string' && /^\d+$/.test(value), code); + const number = Number(value); + requireCondition(Number.isSafeInteger(number) && number >= minimum, code); + return number; +} + +export function decodeStoredPayload(raw) { + let decoded; + try { + decoded = typeof raw === 'string' ? JSON.parse(raw) : raw; + requireCondition(object(decoded), 'payload_not_an_object'); + if (Object.hasOwn(decoded, ORIGINAL_JSON_KEY)) { + requireCondition(typeof decoded[ORIGINAL_JSON_KEY] === 'string', 'invalid_encoded_payload'); + decoded = JSON.parse(decoded[ORIGINAL_JSON_KEY]); + requireCondition(object(decoded), 'invalid_encoded_payload'); + } + } catch (error) { + if (error instanceof VerificationError) throw error; + throw new VerificationError('invalid_payload_json'); + } + return decoded; +} + +export function canonicalJson(value) { + if (Array.isArray(value)) return '[' + value.map(canonicalJson).join(',') + ']'; + if (object(value)) { + const fields = Object.entries(value).filter(([, entry]) => entry !== undefined); + fields.sort(([left], [right]) => left < right ? -1 : left > right ? 1 : 0); + return '{' + fields.map(([key, entry]) => JSON.stringify(key) + ':' + canonicalJson(entry)).join(',') + '}'; + } + return JSON.stringify(value) ?? 'null'; +} +export const canonicalSha256 = value => createHash('sha256').update(canonicalJson(value), 'utf8').digest('hex'); + +function receivedProjection(sequences) { + const sorted = [...sequences].sort((left, right) => left - right); + let contiguous = 0; + const ranges = []; + for (const sequence of sorted) { + if (sequence === contiguous + 1 && ranges.length === 0) contiguous = sequence; + else if (ranges.at(-1)?.endSeq === sequence - 1) ranges.at(-1).endSeq = sequence; + else ranges.push({ startSeq: sequence, endSeq: sequence }); + } + return { contiguousSourceSeq: contiguous, ranges }; +} + +export function verifySnapshot({ companyId, run, events, cursors, ranges }) { + requireCondition(run?.company_id === companyId, 'run_company_binding_mismatch'); + const nextEventSeq = integer(run.next_event_seq, 'invalid_next_event_seq', 1); + const marker = integer(run.event_cursor_projection_seq, 'invalid_projection_marker'); + requireCondition(marker < nextEventSeq, 'projection_marker_beyond_run'); + const projectionCurrent = marker === nextEventSeq - 1; + const globalSequences = new Set(), sourceEvents = new Set(), sources = new Map(); + let canonicalPayloadDigestsVerified = 0, encodedPayloadsDecoded = 0; + for (const row of events) { + requireCondition(row.company_id === companyId && row.run_id === run.id && row.agent_id === run.agent_id, + 'event_run_binding_mismatch'); + const sequence = integer(row.seq, 'invalid_global_sequence', 1); + requireCondition(sequence < nextEventSeq && !globalSequences.has(sequence), 'global_sequence_conflict'); + globalSequences.add(sequence); + const metadata = [row.source_instance_id, row.source_event_id, row.source_seq, + row.source_payload_sha256, row.protocol_schema_version]; + if (metadata.every(value => value === null || value === undefined)) continue; + requireCondition(metadata.every(value => value !== null && value !== undefined), 'partial_source_metadata'); + requireCondition(typeof row.source_instance_id === 'string' && row.source_instance_id.length > 0 + && typeof row.source_event_id === 'string' && row.source_event_id.length > 0, 'invalid_source_identity'); + const sourceSeq = integer(row.source_seq, 'invalid_source_sequence', 1); + const version = integer(row.protocol_schema_version, 'invalid_protocol_version', 1); + requireCondition(!sourceEvents.has(row.source_event_id), 'source_event_identity_conflict'); + sourceEvents.add(row.source_event_id); + let source = sources.get(row.source_instance_id); + if (!source) sources.set(row.source_instance_id, source = new Set()); + requireCondition(!source.has(sourceSeq), 'source_sequence_identity_conflict'); + source.add(sourceSeq); + const raw = typeof row.payload_text === 'string' ? row.payload_text : row.payload; + const projected = typeof raw === 'string' ? (() => { try { return JSON.parse(raw); } catch { return null; } })() : raw; + if (object(projected) && Object.hasOwn(projected, ORIGINAL_JSON_KEY)) encodedPayloadsDecoded++; + const event = decodeStoredPayload(raw).prpEvent; + requireCondition(object(event), 'missing_prp_envelope'); + requireCondition(event.runId === run.id && event.sourceInstanceId === row.source_instance_id + && event.sourceEventId === row.source_event_id && event.sourceSeq === sourceSeq + && event.schemaVersion === version && event.eventType === row.event_type, 'envelope_metadata_mismatch'); + requireCondition(typeof row.source_payload_sha256 === 'string' + && canonicalSha256(event) === row.source_payload_sha256, 'canonical_digest_mismatch'); + canonicalPayloadDigestsVerified++; + } + requireCondition(globalSequences.size === nextEventSeq - 1, 'global_sequence_gap_in_complete_fixture'); + const sourceProofs = [...sources].map(([sourceInstanceId, sequences]) => ({ sourceInstanceId, events: sequences.size })); + if (projectionCurrent) { + const storedCursors = new Map(), storedRanges = new Map(); + for (const row of cursors) { + requireCondition(row.company_id === companyId && row.run_id === run.id, 'cursor_run_binding_mismatch'); + requireCondition(!storedCursors.has(row.source_instance_id), 'duplicate_source_cursor'); + storedCursors.set(row.source_instance_id, { + contiguousSourceSeq: integer(row.contiguous_source_seq, 'invalid_stored_cursor'), + pendingRangeCount: integer(row.pending_range_count, 'invalid_stored_range_count'), + }); + } + for (const row of ranges) { + requireCondition(row.company_id === companyId && row.run_id === run.id, 'range_run_binding_mismatch'); + requireCondition(storedCursors.has(row.source_instance_id), 'range_without_cursor'); + const range = { startSeq: integer(row.start_seq, 'invalid_range_start', 1), endSeq: integer(row.end_seq, 'invalid_range_end', 1) }; + requireCondition(range.endSeq >= range.startSeq, 'invalid_stored_range'); + const list = storedRanges.get(row.source_instance_id) ?? []; + list.push(range); storedRanges.set(row.source_instance_id, list); + } + for (const sourceInstanceId of new Set([...sources.keys(), ...storedCursors.keys()])) { + const expected = receivedProjection(sources.get(sourceInstanceId) ?? new Set()); + const actual = storedCursors.get(sourceInstanceId); + requireCondition(actual?.contiguousSourceSeq === expected.contiguousSourceSeq + && actual?.pendingRangeCount === expected.ranges.length, 'source_cursor_mismatch'); + const actualRanges = (storedRanges.get(sourceInstanceId) ?? []).sort((left, right) => left.startSeq - right.startSeq); + requireCondition(JSON.stringify(actualRanges) === JSON.stringify(expected.ranges), 'source_ranges_mismatch'); + let proof = sourceProofs.find(value => value.sourceInstanceId === sourceInstanceId); + if (!proof) sourceProofs.push(proof = { sourceInstanceId, events: 0 }); + Object.assign(proof, { contiguousSourceSeq: expected.contiguousSourceSeq, pendingRanges: expected.ranges }); + } + } + return { + runId: run.id, companyId, status: run.status, nextEventSeq, eventCursorProjectionSeq: marker, + eventCursorRebuildPending: run.event_cursor_rebuild_pending, globalEvents: globalSequences.size, + canonicalPayloadDigestsVerified, encodedPayloadsDecoded, + projectionChecked: projectionCurrent, + projectionStatus: projectionCurrent ? 'matches_retained_complete_fixture_history' : 'not_checked_stale_marker', + sources: sourceProofs.sort((left, right) => left.sourceInstanceId.localeCompare(right.sourceInstanceId)), + }; +} + +function selfTest() { + const companyId = '00000000-0000-0000-0000-000000000001'; + const run = { id: '00000000-0000-0000-0000-000000000002', company_id: companyId, + agent_id: '00000000-0000-0000-0000-000000000003', next_event_seq: '4', + event_cursor_projection_seq: '3', event_cursor_rebuild_pending: false, status: 'succeeded' }; + const events = [1, 3, 4].map((sourceSeq, index) => { + const event = { runId: run.id, sourceInstanceId: 'source-a', sourceEventId: 'event-' + sourceSeq, + sourceSeq, schemaVersion: 1, eventType: 'item.delta', payload: { text: 'text\u0000with nul', '10': 'ten', '2': 'two' } }; + // Literal canonical text is independently specified, not generated by the encoder under test. + const expected = '{"eventType":"item.delta","payload":{"10":"ten","2":"two","text":"text\\u0000with nul"},"runId":"' + run.id + + '","schemaVersion":1,"sourceEventId":"event-' + sourceSeq + '","sourceInstanceId":"source-a","sourceSeq":' + sourceSeq + '}'; + const payload = { prpEvent: event, [ORIGINAL_JSON_KEY]: 'an original user value' }; + return { company_id: companyId, run_id: run.id, agent_id: run.agent_id, seq: String(index + 1), + event_type: event.eventType, source_instance_id: event.sourceInstanceId, source_event_id: event.sourceEventId, + source_seq: String(sourceSeq), protocol_schema_version: 1, + source_payload_sha256: createHash('sha256').update(expected).digest('hex'), + payload_text: JSON.stringify({ prpEvent: { displayProjection: true }, [ORIGINAL_JSON_KEY]: JSON.stringify(payload) }) }; + }); + const cursor = { company_id: companyId, run_id: run.id, source_instance_id: 'source-a', contiguous_source_seq: '1', pending_range_count: 1 }; + const range = { company_id: companyId, run_id: run.id, source_instance_id: 'source-a', start_seq: '3', end_seq: '4' }; + const fixture = { companyId, run, events, cursors: [cursor], ranges: [range] }; + assert.equal(verifySnapshot(fixture).canonicalPayloadDigestsVerified, 3); + assert.equal(verifySnapshot(fixture).encodedPayloadsDecoded, 3); + assert.equal(decodeStoredPayload(events[0].payload_text)[ORIGINAL_JSON_KEY], 'an original user value'); + assert.equal(canonicalJson({ b: [2, 1], a: 'é', '2': 2, '10': 10 }), '{"10":10,"2":2,"a":"é","b":[2,1]}'); + const fails = (snapshot, code) => assert.throws(() => verifySnapshot(snapshot), error => error.code === code); + fails({ ...fixture, companyId: 'another-company' }, 'run_company_binding_mismatch'); + fails({ ...fixture, events: [{ ...events[0], company_id: 'another-company' }, ...events.slice(1)] }, 'event_run_binding_mismatch'); + fails({ ...fixture, events: [{ ...events[0], source_payload_sha256: 'bad' }, ...events.slice(1)] }, 'canonical_digest_mismatch'); + fails({ ...fixture, events: [...events, events[0]] }, 'global_sequence_conflict'); + fails({ ...fixture, events: events.slice(1) }, 'global_sequence_gap_in_complete_fixture'); + fails({ ...fixture, cursors: [{ ...cursor, contiguous_source_seq: 4 }] }, 'source_cursor_mismatch'); + fails({ ...fixture, ranges: [{ ...range, end_seq: 5 }] }, 'source_ranges_mismatch'); + const fourth = { runId: run.id, sourceInstanceId: 'source-b', sourceEventId: 'other-event', + sourceSeq: 2, schemaVersion: 1, eventType: 'item.delta', payload: { text: 'other' } }; + const fourthCanonical = '{"eventType":"item.delta","payload":{"text":"other"},"runId":"' + run.id + + '","schemaVersion":1,"sourceEventId":"other-event","sourceInstanceId":"source-b","sourceSeq":2}'; + const fourthRow = { ...events[0], seq: 4, source_instance_id: 'source-b', source_event_id: 'other-event', source_seq: 2, + source_payload_sha256: createHash('sha256').update(fourthCanonical).digest('hex'), payload_text: JSON.stringify({ prpEvent: fourth }) }; + const multipleSources = { ...fixture, run: { ...run, next_event_seq: 5, event_cursor_projection_seq: 4 }, + events: [...events, fourthRow], + cursors: [cursor, { ...cursor, source_instance_id: 'source-b', contiguous_source_seq: 0 }], + ranges: [range, { ...range, source_instance_id: 'source-b', start_seq: 2, end_seq: 2 }] }; + assert.equal(verifySnapshot(multipleSources).sources.length, 2); + fails({ ...multipleSources, events: [...events, { ...fourthRow, source_event_id: events[0].source_event_id }] }, 'source_event_identity_conflict'); + fails({ ...multipleSources, events: [...events, { ...fourthRow, source_instance_id: 'source-a', source_seq: 1 }] }, 'source_sequence_identity_conflict'); + assert.equal(verifySnapshot({ ...fixture, run: { ...run, event_cursor_projection_seq: 0 }, cursors: [], ranges: [] }).projectionChecked, false); + assert.throws(() => decodeStoredPayload({ [ORIGINAL_JSON_KEY]: 1 }), error => error.code === 'invalid_encoded_payload'); + assert.throws(() => decodeStoredPayload({ [ORIGINAL_JSON_KEY]: '[]' }), error => error.code === 'invalid_encoded_payload'); + return { selfTest: 'passed', databaseAccess: false, covered: ['NUL and reserved-key decoding', 'canonical ordering and array order', + 'multiple sparse sources', 'company binding', 'hash conflict', 'global and both source identity conflicts', 'projection conflicts', 'stale-marker skip'] }; +} + +async function main() { + if (process.argv.includes('--self-test')) { console.log(JSON.stringify(selfTest(), null, 2)); return; } + const options = name => process.argv.filter(value => value.startsWith('--' + name + '=')).map(value => value.slice(name.length + 3)); + const one = name => { const values = options(name); requireCondition(values.length === 1, 'supply_one_' + name); return values[0]; }; + const repository = one('repository'), companyId = one('company-id').toLowerCase(), runIds = options('run-id').map(value => value.toLowerCase()); + requireCondition(path.isAbsolute(repository), 'repository_must_be_absolute'); + requireCondition(UUID.test(companyId) && runIds.length > 0 && runIds.every(value => UUID.test(value)) + && new Set(runIds).size === runIds.length, 'supply_company_and_unique_run_uuids'); + let url; + try { url = new URL(process.env[DATABASE_ENV]); } catch { throw new VerificationError('supply_' + DATABASE_ENV); } + requireCondition(['postgres:', 'postgresql:'].includes(url.protocol) + && ['127.0.0.1', 'localhost', '[::1]'].includes(url.hostname) && !url.search && !url.hash, + 'database_must_be_explicit_loopback_without_query_overrides'); + requireCondition(url.pathname.length > 1, 'database_name_required'); + const require = createRequire(path.join(repository, 'packages/db/package.json')); + const postgres = require('postgres'); + const sql = postgres({ host: url.hostname.replace(/^\[|\]$/g, ''), port: Number(url.port || 5432), + database: decodeURIComponent(url.pathname.slice(1)), username: decodeURIComponent(url.username), + password: decodeURIComponent(url.password), ssl: false, max: 1, connect_timeout: 10, idle_timeout: 5, + connection: { application_name: 'paperclip_native_ingestion_independent_verify' } }); + let output; + try { + const proofs = await sql.begin('isolation level repeatable read read only', async tx => { + const settings = await tx`select current_setting('transaction_read_only') as read_only, + current_setting('transaction_isolation') as isolation`; + requireCondition(settings[0]?.read_only === 'on' && settings[0]?.isolation === 'repeatable read', 'read_only_snapshot_required'); + const result = []; + for (const runId of runIds) { + const runs = await tx`select id, company_id, agent_id, status, next_event_seq, event_cursor_projection_seq, + event_cursor_rebuild_pending from heartbeat_runs where company_id = ${companyId} and id = ${runId}`; + requireCondition(runs.length === 1, 'bound_run_not_found'); + const events = await tx`select company_id, run_id, agent_id, seq, event_type, payload::text as payload_text, + source_instance_id, source_event_id, source_seq, source_payload_sha256, protocol_schema_version + from heartbeat_run_events where company_id = ${companyId} and run_id = ${runId} order by seq`; + const run = runs[0], current = integer(run.event_cursor_projection_seq, 'invalid_projection_marker') + === integer(run.next_event_seq, 'invalid_next_event_seq', 1) - 1; + const cursors = current ? await tx`select company_id, run_id, source_instance_id, contiguous_source_seq, + pending_range_count from heartbeat_run_event_cursors where company_id = ${companyId} and run_id = ${runId}` : []; + const ranges = current ? await tx`select company_id, run_id, source_instance_id, start_seq, end_seq + from heartbeat_run_event_source_ranges where company_id = ${companyId} and run_id = ${runId} order by source_instance_id, start_seq` : []; + const proof = verifySnapshot({ companyId, run, events, cursors, ranges }); + if (process.argv.includes('--require-current-projection')) requireCondition(proof.projectionChecked, 'projection_is_stale'); + result.push(proof); + } + return result; + }); + output = { verified: true, reconstructedVerifier: true, checkedAt: new Date().toISOString(), + database: 'Supplied disposable loopback PostgreSQL; read-only repeatable-read snapshot', + limitation: 'Cursor reconstruction requires the complete retained fixture history; no production or raw wire ACK durability claim', proofs }; + } finally { await sql.end({ timeout: 5 }); } + const files = options('output'); + requireCondition(files.length <= 1, 'supply_at_most_one_output'); + if (files.length) await writeFile(files[0], JSON.stringify(output, null, 2) + '\n', { flag: 'wx', mode: 0o600 }); + console.log(JSON.stringify(output, null, 2)); +} + +if (process.argv[1] && import.meta.url === pathToFileURL(path.resolve(process.argv[1])).href) { + main().catch(error => { + // Driver errors can contain connection details. Only controlled codes leave this process. + console.error(JSON.stringify({ verified: false, error: error instanceof VerificationError ? error.code : 'verifier_failed_details_suppressed' })); + process.exitCode = 1; + }); +}