diff --git a/tests/e2e/fixtures/native-ingestion-blocked-stop-authority.ts b/tests/e2e/fixtures/native-ingestion-blocked-stop-authority.ts new file mode 100644 index 00000000000..7d29183bc10 --- /dev/null +++ b/tests/e2e/fixtures/native-ingestion-blocked-stop-authority.ts @@ -0,0 +1,131 @@ +// Bounded verbatim copy of the H5c authority/process proof, e431ec63; frozen measured source stays unchanged. +import { createHash } from "node:crypto"; +import { createRequire } from "node:module"; +import { lstat, readFile, readdir, realpath } from "node:fs/promises"; +import { hostname } from "node:os"; +import path from "node:path"; +import { expect } from "@playwright/test"; +export async function runtimeAuthority(input: { + appRoot: string; companyId: string; environmentId: string; issueIds: string[]; transport: "local" | "sandbox-remote"; + afterStop?: boolean; recordedFailureRunIds?: string[]; +}) { + const configuration = process.env.PAPERCLIP_E2E_SERVER_CONFIG; + if (!configuration) throw new Error("Missing owned server configuration"); + const config = JSON.parse(await readFile(configuration, "utf8")); + expect(config.database.mode).toBe("embedded-postgres"); + const dataDirectory = await realpath(config.database.embeddedPostgresDataDir); + const ownedHome = await realpath(path.resolve(path.dirname(configuration), "../..")); + expect(dataDirectory.startsWith(`${ownedHome}${path.sep}`)).toBe(true); + const owner = (await readFile(path.join(dataDirectory, "postmaster.pid"), "utf8")).split(/\r?\n/); + expect(await realpath(owner[1])).toBe(dataDirectory); + const port = Number(owner[3]); + expect(Number.isSafeInteger(port) && port > 0 && port <= 65535).toBe(true); + const require = createRequire(path.join(input.appRoot, "packages/db/package.json")); + const postgres = require("postgres"); + const sql = postgres({ host: "127.0.0.1", port, database: "paperclip", username: "paperclip", password: "paperclip", + max: 1, connect_timeout: 10, connection: { application_name: "native_ingestion_remote_authority_proof" } }); + let proof; + try { + proof = await sql.begin("isolation level repeatable read read only", async (tx: any) => { + const settings = await tx`select current_setting('transaction_read_only') as read_only, + current_setting('transaction_isolation') as isolation`; + expect(settings[0].read_only).toBe("on"); + expect(settings[0].isolation).toBe("repeatable read"); + if (input.recordedFailureRunIds) { + const queryStartedAtMs = Date.now(); + const rows = await tx`select r.id as run_id, r.company_id, r.agent_id, r.runtime_mode, r.status, r.native_phase, + r.error_code as current_error_code, r.error is not null as current_has_error, + jsonb_typeof(r.result_json->'recoveredExecutionFailure') as recovered_type, + r.result_json->'recoveredExecutionFailure'->>'schema' as recovered_schema, + r.result_json->'recoveredExecutionFailure'->>'errorCode' as recovered_error_code, + r.result_json->'recoveredExecutionFailure'->>'observedAt' as recovered_observed_at, + f.attempt as finalization_attempt + from heartbeat_runs r left join native_run_finalizations f on f.run_id = r.id and f.company_id = r.company_id + where r.company_id = ${input.companyId} and r.id = any(${input.recordedFailureRunIds}::uuid[])`; + return { kind: "recorded-run-failure-codes", queryStartedAtMs, checkedAtMs: Date.now(), + transaction: "READ ONLY REPEATABLE READ", observedReadOnly: settings[0].read_only, + observedIsolation: settings[0].isolation, rows }; + } + const rows = await tx`select r.id as run_id, r.company_id, r.agent_id, r.native_issue_id as issue_id, r.runtime_mode, + r.native_session_id, r.runner_instance_id, r.status, r.process_pid, r.process_group_id, r.process_started_at, + r.runner_profile_json->'nativeExecutionInput'->'binding'->>'executionWorkspaceId' as admitted_execution_workspace_id, + r.result_json->'nativeCancellation'->>'dispatchState' as cancellation_dispatch_state, + i.seq as process_identity_seq, i.payload as process_identity, i.source_event_id as process_identity_source_event_id, + s.seq as process_stopped_seq, s.payload as process_stopped, s.source_event_id as process_stopped_source_event_id, + l.id as lease_id, l.environment_id, l.provider, l.provider_lease_id, + c.part as checkpoint_part, octet_length(c.material::text) as checkpoint_header_bytes + from heartbeat_runs r + join environment_leases l on l.heartbeat_run_id = r.id and l.company_id = r.company_id + left join native_runner_checkpoints c on c.run_id = r.id and c.company_id = r.company_id + and c.issue_id = r.native_issue_id and c.part = 'header' + left join lateral (select seq, payload, source_event_id from heartbeat_run_events e where e.run_id = r.id + and e.company_id = r.company_id and e.event_type = 'native.process_identity_recorded' + and e.source_event_id is null order by seq desc limit 1) i on true + left join lateral (select seq, payload, source_event_id from heartbeat_run_events e where e.run_id = r.id + and e.company_id = r.company_id and e.event_type = 'native.local_process_stopped' + and e.source_event_id is null order by seq desc limit 1) s on true + where r.company_id = ${input.companyId} and r.native_issue_id = any(${input.issueIds}::uuid[])`; + expect(rows).toHaveLength(input.issueIds.length); + expect(new Set(rows.map((row: any) => row.issue_id)).size).toBe(input.issueIds.length); + for (const row of rows) { + expect(row.company_id).toBe(input.companyId); + expect(row.runtime_mode).toBe("native"); + expect(row.environment_id).toBe(input.environmentId); + if (input.transport === "sandbox-remote") { + expect(row.provider).toBe("fake-plugin"); + expect(row.provider_lease_id).toMatch(/^fake-plugin:\/\//); + expect(row.checkpoint_part).toBe("header"); + expect(Number(row.checkpoint_header_bytes)).toBeGreaterThan(0); + } else { + expect(row.provider).toBe("local"); expect(row.provider_lease_id).toBeNull(); + expect(row.checkpoint_part, "Local raw transport keeps its existing host-file journal authority").toBeNull(); + expect(row.process_identity?.targetKind).toBe("local"); expect(row.process_identity?.hostname).toBe(hostname()); + expect(row.process_identity_source_event_id).toBeNull(); + expect(BigInt(String(row.process_identity_seq))).toBeGreaterThan(0n); + expect(row.admitted_execution_workspace_id).toMatch(/^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i); + if (row.process_started_at) expect(row.process_identity.processStartedAt).toBe(new Date(row.process_started_at).toISOString()); + } + } + return { kind: input.transport === "local" ? "host-file-journal" : "database-controller-checkpoint", + checkedBeforeMeasuredEpoch: !input.afterStop, checkedAtMs: Date.now(), transaction: "READ ONLY REPEATABLE READ", + observedReadOnly: settings[0].read_only, observedIsolation: settings[0].isolation, rows }; + }); + } finally { await sql.end({ timeout: 5 }); } + const connectionClosedAtMs = Date.now(); + let localJournal; + if (input.transport === "local" && !input.afterStop && !input.recordedFailureRunIds) { + expect(input.issueIds).toHaveLength(1); + const base = await realpath(path.join(path.dirname(configuration), "runtime/paperclip-runner/durable-sessions")); + expect(base.startsWith(`${ownedHome}${path.sep}`)).toBe(true); + const directories = (await readdir(base, { withFileTypes: true })).filter(entry => entry.isDirectory() && /^[a-f0-9]{64}$/.test(entry.name)); + expect(directories.length, "Bound the owned fixture journal metadata search").toBeLessThanOrEqual(256); + for (const directory of directories) { + const file = path.join(base, directory.name, "control-plane/control-plane-state.json"); + const metadata = await lstat(file).catch((error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") return null; throw error; + }); + if (!metadata) continue; + expect(metadata.isFile() && !metadata.isSymbolicLink() && metadata.size <= 16 * 1024 * 1024).toBe(true); + expect((await realpath(file)).startsWith(`${base}${path.sep}`)).toBe(true); + const bytes = await readFile(file), state = JSON.parse(bytes.toString("utf8")); + if (state.identity?.runId !== proof!.rows[0].run_id) continue; + expect(localJournal, "Exactly one owned local journal must match the actual run").toBeUndefined(); + expect(state.schema).toBe("paperclip.runner.durable.control-plane-state.v1"); + expect(state.checkpointEventPolicy).not.toBe("control-v1"); + expect(state.identity.runnerInstanceId).toBe(proof!.rows[0].runner_instance_id); + expect(state.identity.normalizedSessionId).toBe(proof!.rows[0].native_session_id); + expect(state.identity.environmentLeaseId).toBe(proof!.rows[0].admitted_execution_workspace_id); + localJournal = { file, sha256: createHash("sha256").update(bytes).digest("hex"), bytes: bytes.length, + schema: state.schema, identity: Object.fromEntries(["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId", "turnId", "itemId"] + .map(key => [key, state.identity[key]])), checkpointEventPolicy: state.checkpointEventPolicy, checkedAtMs: Date.now() }; + } + expect(localJournal, "Missing actual owned local raw journal").toBeTruthy(); + } + return { ...proof, connectionClosedAtMs, localJournal }; +} + +export const processAbsent = (pid: number) => { + expect(Number.isSafeInteger(pid) && Math.abs(pid) > 1).toBe(true); + try { process.kill(pid, 0); return false; } + catch (error) { if ((error as NodeJS.ErrnoException).code === "ESRCH") return true; throw error; } +}; diff --git a/tests/e2e/fixtures/native-ingestion-blocked-stop-clock.ts b/tests/e2e/fixtures/native-ingestion-blocked-stop-clock.ts new file mode 100644 index 00000000000..918ecd8a5fa --- /dev/null +++ b/tests/e2e/fixtures/native-ingestion-blocked-stop-clock.ts @@ -0,0 +1,127 @@ +// Gate and two-frame observer copied from frozen H5c e431ec63, narrowed to one owned participant. +import { createServer, type ServerResponse } from "node:http"; +import { expect, type Page } from "@playwright/test"; +export type SeenMarker = { markerCode: string; index: number; visibleMs: number }; +export type Emission = { nonce: string; markerCode: string; index: number; intendedMs: number; emittedMs: number; receivedAtMs: number }; +type ClockStream = { nonce: string; markerCode: string; opened: boolean; hello: boolean; ended: boolean; aborted: boolean; + partialBytes: number; records: Map; error?: string; expectedLauncher: "host-local" | "staged-remote"; + launcher?: string; providerPid?: number; providerReadyAtMs?: number; clockHelloAtMs?: number; readinessObservedAtMs?: number }; +type ClockCase = { events: number; intervalMs: number; epochMs?: number; streams: Map; errors: string[] }; +export async function openClock(page: Page) { + const pending = new Map>(); + const clocks = new Map(); + let markerCounter = 0; + const nextMarkerCode = () => { + expect(markerCounter).toBeLessThan(36 ** 2); + return (markerCounter++).toString(36).padStart(2, "0"); + }; + const gate = createServer((req, res) => { + const url = new URL(req.url!, "http://127.0.0.1"); + const route = url.pathname.match(/^\/native-ingestion-gate\/([a-f0-9]{32})(\/clock)?$/); + const key = route?.[1], participant = url.searchParams.get("participant"); + const state = key ? clocks.get(key) : undefined; + const stream = participant ? state?.streams.get(participant) : undefined; + if (!key || !participant || !state || !stream) { res.writeHead(404).end(); return; } + if (!route?.[2] && req.method === "GET") { + if (pending.get(key)!.has(participant)) { res.writeHead(409).end(); return; } + stream.providerReadyAtMs = Date.now(); + pending.get(key)!.set(participant, res); + res.on("close", () => pending.get(key)?.delete(participant)); + return; + } + if (!route?.[2] || req.method !== "POST" || stream.opened) { res.writeHead(409).end(); return; } + stream.opened = true; + let buffered = ""; + req.setEncoding("utf8"); + const fail = (message: string) => { + stream.error = message; state.errors.push(`${participant}: ${message}`); + if (!res.writableEnded) res.writeHead(400).end(); + }; + const line = (text: string) => { + const value = JSON.parse(text); + if (value.nonce !== participant || value.markerCode !== stream.markerCode) throw new Error("Clock identity mismatch"); + if (!stream.hello) { + if (value.type !== "hello" || value.events !== state.events || value.intervalMs !== state.intervalMs + || value.launcher !== stream.expectedLauncher || !Number.isSafeInteger(value.providerPid) || value.providerPid <= 1 + || Object.keys(value).sort().join(",") !== "events,intervalMs,launcher,markerCode,nonce,providerPid,type") throw new Error("Clock hello mismatch"); + stream.launcher = value.launcher; stream.providerPid = value.providerPid; + stream.hello = true; stream.clockHelloAtMs = Date.now(); return; + } + if (value.type !== "emission" || !Number.isSafeInteger(value.index) || value.index < 0 || value.index >= state.events + || !Number.isSafeInteger(value.intendedMs) || !Number.isSafeInteger(value.emittedMs) || value.emittedMs <= 0 + || value.intendedMs !== state.epochMs! + value.index * state.intervalMs) throw new Error("Invalid original emission clock"); + if (stream.records.has(value.index)) throw new Error("Duplicate or conflicting original emission clock"); + stream.records.set(value.index, { nonce: participant, markerCode: stream.markerCode, index: value.index, + intendedMs: value.intendedMs, emittedMs: value.emittedMs, receivedAtMs: Date.now() }); + }; + req.on("data", (chunk: string) => { + if (stream.error) return; + try { + buffered += chunk; + for (;;) { + const end = buffered.indexOf("\n"); if (end < 0) break; + const text = buffered.slice(0, end); buffered = buffered.slice(end + 1); + if (text.length > 1024) throw new Error("Clock record exceeds bounded metadata size"); + if (text) line(text); + } + if (buffered.length > 1024) throw new Error("Clock record exceeds bounded metadata size"); + } catch (error) { fail(String(error)); } + }); + req.on("end", () => { + stream.ended = true; stream.partialBytes = Buffer.byteLength(buffered); + if (buffered.length) fail("Clock stream ended with a partial record"); + if (!stream.error && !res.writableEnded) res.writeHead(200).end(); + }); + req.on("aborted", () => { stream.aborted = true; stream.partialBytes = Buffer.byteLength(buffered); }); + req.on("error", error => { if (!stream.aborted) fail(String(error)); }); + }); + await new Promise(resolve => gate.listen(0, "127.0.0.1", resolve)); + const address = gate.address(); + if (!address || typeof address === "string") throw new Error("Local ingestion gate failed"); + const base = `http://127.0.0.1:${address.port}/native-ingestion-gate/`; + await page.addInitScript(() => { + const seen = new Set(), visible: SeenMarker[] = []; + Object.assign(window, { nativeIngestionVisible: visible }); + let scheduled = false, dirty = false; + const snapshot = () => { + const text = document.body?.innerText ?? ""; + const pattern = /\[([a-z0-9]{2})([a-z0-9]{1,3})\]/g; + const entries: Omit[] = []; + for (const match of text.matchAll(pattern)) { + const index = Number.parseInt(match[2], 36); + const key = `${match[1]}:${index}`; + if (seen.has(key)) continue; + entries.push({ markerCode: match[1], index }); + } + return entries; + }; + const schedule = () => { + dirty = true; + if (scheduled) return; + scheduled = true; + requestAnimationFrame(() => { + dirty = false; + const entries = snapshot(); + requestAnimationFrame(() => { + // Only stamp the previous frame's snapshot. A token inserted + // between frames has not had this snapshot's paint opportunity. + const visibleMs = Date.now(); + for (const entry of entries) { + const key = `${entry.markerCode}:${entry.index}`; + if (seen.has(key)) continue; + seen.add(key); visible.push({ ...entry, visibleMs }); + } + scheduled = false; + if (dirty) schedule(); + }); + }); + }; + const install = () => { + new MutationObserver(schedule).observe(document.body, { subtree: true, childList: true, characterData: true }); + schedule(); + }; + if (document.body) install(); else addEventListener("DOMContentLoaded", install, { once: true }); + }); + + return { pending, clocks, nextMarkerCode, base, gate }; +} diff --git a/tests/e2e/fixtures/native-ingestion-blocked-stop-db.ts b/tests/e2e/fixtures/native-ingestion-blocked-stop-db.ts new file mode 100644 index 00000000000..acb6a275c3b --- /dev/null +++ b/tests/e2e/fixtures/native-ingestion-blocked-stop-db.ts @@ -0,0 +1,43 @@ +import { createRequire } from "node:module"; +import { readFile, realpath } from "node:fs/promises"; +import path from "node:path"; +import { expect } from "@playwright/test"; + +export async function ownedDatabase(appRoot: string, name: string) { + const configuration = process.env.PAPERCLIP_E2E_SERVER_CONFIG; + if (!configuration) throw new Error("Missing disposable application configuration"); + const config = JSON.parse(await readFile(configuration, "utf8")); + expect(config.database.mode).toBe("embedded-postgres"); + const home = await realpath(path.resolve(path.dirname(configuration), "../..")); + const directory = await realpath(config.database.embeddedPostgresDataDir); + expect(directory.startsWith(`${home}${path.sep}`)).toBe(true); + const owner = (await readFile(path.join(directory, "postmaster.pid"), "utf8")).split(/\r?\n/); + expect(await realpath(owner[1])).toBe(directory); + const port = Number(owner[3]); + expect(Number.isSafeInteger(port) && port > 0 && port <= 65535).toBe(true); + const postgres = createRequire(path.join(appRoot, "packages/db/package.json"))("postgres"); + return postgres({ host: "127.0.0.1", port, database: "paperclip", username: "paperclip", password: "paperclip", + max: 1, connect_timeout: 10, connection: { application_name: name } }); +} +export async function cancellationSnapshot(sql: any, companyId: string, runId: string) { + const startedAtMs = Date.now(); + const rows = await sql`select id as run_id, company_id, native_issue_id as issue_id, status, + result_json->'nativeCancellation' as intent, + result_json->'executionCancellation' as shared_execution_cancellation, + result_json->'cancellation' as run_cancellation, + (select jsonb_build_object('id', d.id, 'companyId', d.company_id, 'runId', d.run_id, 'issueId', d.issue_id, + 'reasonCode', d.reason_code) from status_decisions d where d.company_id = r.company_id and d.run_id = r.id + and d.issue_id = r.native_issue_id and d.id::text = r.result_json->'nativeCancellation'->>'decisionId') as shared_decision, + (select coalesce(jsonb_agg(jsonb_build_object('id', a.id, 'action', a.action, 'details', a.details)), '[]'::jsonb) + from activity_log a where a.company_id = r.company_id and a.run_id = r.id + and a.action in ('native.cancellation_intent_recorded', 'native.cancellation_dispatch_acknowledged')) as audits + from heartbeat_runs r where r.company_id = ${companyId} and r.id = ${runId}`; + expect(rows).toHaveLength(1); + return { ...rows[0], queryStartedAtMs: startedAtMs, observedAtMs: Date.now() }; +} +export async function blockers(sql: any, holderPid: number) { + return sql`select pid, application_name, state, wait_event_type, wait_event, query, query_start, + pg_blocking_pids(pid) as blocking_pids from pg_stat_activity + where datname = current_database() and pid <> pg_backend_pid() + and ${holderPid} = any(pg_blocking_pids(pid))`; +} diff --git a/tests/e2e/fixtures/native-ingestion-blocked-stop-protocol.mjs b/tests/e2e/fixtures/native-ingestion-blocked-stop-protocol.mjs new file mode 100644 index 00000000000..cb4e126a7c0 --- /dev/null +++ b/tests/e2e/fixtures/native-ingestion-blocked-stop-protocol.mjs @@ -0,0 +1,303 @@ +import assert from 'node:assert/strict'; +import { createHash } from 'node:crypto'; +export const PROTOCOL = 'blocked-stop-history-v1'; +export const WORKLOAD = Object.freeze({ prelude: 1000, events: 256, intervalMs: 100, visibleBeforeLock: 3, + heldAfterActualRequestMs: 1000, pollIntervalMs: 50, readinessMs: 300000, streamMs: 120000, + terminalMs: 180000, acknowledgementAfterSettlementMs: 30000, leaseAfterSettlementMs: 30000, invocationMs: 900000 }); +export const WORKLOAD_SHA256 = createHash('sha256').update(JSON.stringify(WORKLOAD)).digest('hex'); +export const METRICS = ['actionToAckMs', 'releaseAckToAbsenceMs', 'releaseAckToAckMs']; +export function isAppendLockQuery(query) { + // Shared B0/B1/B2 projection prefix, unlike cancellation's full-row SELECT. + const sql = String(query).replace(/\s+/g, ' ').trim(); + return /^select "company_id", "agent_id"(?:,| from)/i.test(sql) + && / from "heartbeat_runs" where "heartbeat_runs"\."id" = \$\d+.* for update/i.test(sql); +} +export function assertHeldSample(sample, holderPid) { + assert.equal(sample.httpPending, true, 'Stop HTTP must remain pending while its durable barrier is locked'); + assert.equal(sample.intent, null, 'No committed intent before releasing the run lock'); + assert.deepEqual(sample.auditIds, [], 'No committed cancellation audit before releasing the run lock'); + assert.equal(sample.processesAlive, true, 'No physical provider Stop before the durable intent barrier'); + assert.equal(sample.clockOpen, true, 'The unchanged provider fixture closes its clock on turn/interrupt'); + assert(sample.blockers.some(row => row.blocking_pids.includes(holderPid)), 'Missing independent PostgreSQL blocker'); +} +export function assertIntentAudit(row) { + const intent = row.intent; + assert.equal(intent?.schema, 'paperclip.native-cancellation.v1'); + assert.equal(intent.companyId, row.company_id); assert.equal(intent.runId, row.run_id); + assert.equal(intent.issueId, row.issue_id); + assert.equal(intent.scope, 'run'); assert.equal(intent.reasonCode, 'cancellation_run_only'); + if (intent.decisionId != null) { + assert.equal(row.shared_decision?.id, intent.decisionId); + assert.equal(row.shared_decision.companyId, row.company_id); assert.equal(row.shared_decision.runId, row.run_id); + assert.equal(row.shared_decision.issueId, row.issue_id); + } + assert(row.audits.some(a => a.id === intent.intentAuditId && a.action === 'native.cancellation_intent_recorded' + && a.details?.intentId === intent.intentId), 'Intent must match its committed own audit'); + if (intent.dispatchState === 'acknowledged') { + assert.equal(intent.dispatched, true); + assert(row.audits.some(a => a.id === intent.acknowledgementAuditId + && a.action === 'native.cancellation_dispatch_acknowledged' && a.details?.intentId === intent.intentId + && a.details.intentAuditId === intent.intentAuditId && a.details.dispatched === true + && (a.details.decisionId ?? null) === (intent.decisionId ?? null)), 'ACK must match its own committed audit and decision'); + } +} +export function metrics(evidence) { + assert.equal(evidence.protocol, PROTOCOL); assert.equal(evidence.workloadSha256, WORKLOAD_SHA256); + assert.equal(evidence.complete, true); assert.equal(evidence.cleanup?.complete, true); + const t = evidence.timing; + for (const k of ['actionAtMs', 'requestAtMs', 'releaseRequestedAtMs', 'releaseCommitAckAtMs', 'absenceObservedAtMs', 'ackObservedAtMs']) + assert(Number.isSafeInteger(t[k]) && t[k] > 0, `Missing ${k}`); + assert(t.actionAtMs <= t.requestAtMs); assert(t.requestAtMs + WORKLOAD.heldAfterActualRequestMs <= t.releaseRequestedAtMs); + assert(t.releaseRequestedAtMs <= t.releaseCommitAckAtMs); assert(t.releaseCommitAckAtMs <= t.absenceObservedAtMs); + assert(t.releaseCommitAckAtMs <= t.ackObservedAtMs); + return { actionToAckMs: t.ackObservedAtMs - t.actionAtMs, + releaseAckToAbsenceMs: t.absenceObservedAtMs - t.releaseCommitAckAtMs, + releaseAckToAckMs: t.ackObservedAtMs - t.releaseCommitAckAtMs, + releaseRequestToAbsenceMs: t.absenceObservedAtMs - t.releaseRequestedAtMs, + releaseRequestToAckMs: t.ackObservedAtMs - t.releaseRequestedAtMs }; +} +export function registration() { + const stages = [['b0-aa','b0','b0'], ['b1-aa','b1','b1'], ['b0-b1','b0','b1'], ['b1-b2','b1','b2'], ['b0-b2','b0','b2']]; + return stages.flatMap(([stage,a,b]) => Array.from({ length: 5 }, (_, pair) => (pair % 2 ? [b,a] : [a,b]) + .map((variant, index) => ({ stage, pair: pair + 1, index, variant, role: pair % 2 ? (index ? 'baseline' : 'candidate') : (index ? 'candidate' : 'baseline') })))).flat(); +} +const median = values => { + assert(values.length > 0); + const sorted = [...values].sort((a,b) => a-b), middle = Math.floor(sorted.length / 2); + return sorted.length % 2 ? sorted[middle] : (sorted[middle - 1] + sorted[middle]) / 2; +}; +const spread = values => ({ minimumMs: Math.min(...values), maximumMs: Math.max(...values), + spreadMs: Math.max(...values) - Math.min(...values) }); +const hashReport = evidence => createHash('sha256').update(JSON.stringify(evidence)).digest('hex'); +export const PRECISION_RULE = Object.freeze({ schema: 'blocked-stop-precision-v1', + pollIntervalMs: WORKLOAD.pollIntervalMs, floorMs: 2 * WORKLOAD.pollIntervalMs, pooledMedianFraction: 0.10, + pairedRule: 'maximum absolute difference <= budget', driftRule: 'absolute first-five vs last-five chronological median drift <= budget', + bracketRule: 'each observer bracket <= corresponding metric budget', unsuitableRule: 'retain all ten arms; no noise freeze or replacement', + scope: 'Measurement repeatability; not a product Stop deadline' }); +export const PRECISION_RULE_SHA256 = createHash('sha256').update(JSON.stringify(PRECISION_RULE)).digest('hex'); +export function observerBrackets(evidence) { + const t = evidence.timing, origin = t.releaseCommitAckAtMs; + const samples = evidence.observerSamples; + assert(Array.isArray(samples), 'Missing raw observer query/check samples'); + const result = {}; + for (const [kind, timestamp, negativeTimestamp] of [ + ['ack-query', 'ackObservedAtMs', 'lastAckNotObservedAtMs'], + ['process-check', 'absenceObservedAtMs', 'lastProcessesPresentAtMs']]) { + const rows = samples.filter(row => row.kind === kind); + assert(rows.length > 0, `Missing ${kind} observer samples`); + let priorEnd = origin; + for (const [index,row] of rows.entries()) { + assert(Number.isSafeInteger(row.startedAtMs) && Number.isSafeInteger(row.completedAtMs)); + assert(row.startedAtMs >= priorEnd && row.completedAtMs >= row.startedAtMs); + assert.equal(row.durationMs, row.completedAtMs - row.startedAtMs); + assert.equal(row.value, index === rows.length - 1, 'Retain every negative sample through the first successful observation'); + priorEnd = row.completedAtMs; + } + const success = rows.at(-1), negative = rows.at(-2); + assert.equal(t[timestamp], success.completedAtMs); + if (negative) assert.equal(t[negativeTimestamp], negative.completedAtMs); + else assert(t[negativeTimestamp] === undefined, 'A negative timestamp needs its raw query/check sample'); + // A negative query can read its snapshot at any point during execution; + // use its start as the conservative lower boundary and retain both ends. + const lowerAtMs = negative?.startedAtMs ?? origin; + result[kind] = { lowerAtMs, upperAtMs: success.completedAtMs, + widthMs: success.completedAtMs - lowerAtMs, lowerKind: negative ? 'last-negative-query-or-check-start' : 'release-COMMIT-ACK', + successfulStartedAtMs: success.startedAtMs, successfulDurationMs: success.durationMs, + ...(negative ? { negativeStartedAtMs: negative.startedAtMs, negativeCompletedAtMs: negative.completedAtMs, + negativeDurationMs: negative.durationMs } : {}), sampleCount: rows.length }; + } + return { actionToAckMs: result['ack-query'], releaseAckToAckMs: result['ack-query'], + releaseAckToAbsenceMs: result['process-check'] }; +} +function validatedPairs(pairs) { + assert.equal(pairs.length, 5, 'Require five fresh pairs'); + const validated = pairs.map(pair => { + assert.equal(pair.length, 2); + return pair.map(evidence => ({ evidence, values: assertFunctional(evidence), brackets: observerBrackets(evidence) })); + }); + const chronological = validated.flat().sort((a,b) => a.evidence.startedAtMs - b.evidence.startedAtMs); + assert.equal(new Set(chronological.map(arm => arm.evidence.runId)).size, 10, 'Ten fresh owned runs are required'); + assert.equal(new Set(chronological.map(arm => arm.evidence.companyId)).size, 10, 'Ten fresh disposable invocations are required'); + for (const [index,arm] of chronological.entries()) { + assert(Number.isSafeInteger(arm.evidence.startedAtMs) && Number.isSafeInteger(arm.evidence.finishedAtMs)); + assert(arm.evidence.finishedAtMs >= arm.evidence.startedAtMs); + if (index) assert(chronological[index-1].evidence.finishedAtMs < arm.evidence.startedAtMs, + 'Arms must not overlap'); + } + validated.forEach((pair,index) => { + assert.equal(pair[0].evidence.startedAtMs, chronological[2*index + index%2].evidence.startedAtMs, + 'Pairs must alternate baseline-first then candidate-first in chronological order'); + assert.equal(pair[1].evidence.startedAtMs, chronological[2*index + (index%2 ? 0 : 1)].evidence.startedAtMs); + }); + return validated; +} +export function noiseEnvelope(pairs) { + const validated = validatedPairs(pairs), chronological = validated.flat().sort((a,b) => a.evidence.startedAtMs - b.evidence.startedAtMs); + for (const [index,arm] of chronological.entries()) { + assert(Number.isSafeInteger(arm.evidence.startedAtMs) && Number.isSafeInteger(arm.evidence.finishedAtMs)); + assert(arm.evidence.finishedAtMs >= arm.evidence.startedAtMs); + if (index) assert(chronological[index-1].evidence.finishedAtMs < arm.evidence.startedAtMs, + 'A/A requires ten distinct non-overlapping chronological arms'); + } + const variants = new Set(chronological.map(arm => String(arm.evidence.variant).toLowerCase())); + assert.equal(variants.size, 1, 'A/A calibration must use one baseline variant'); + const baselineVariant = [...variants][0]; assert(['b0','b1'].includes(baselineVariant)); + const result = { schema: 'paperclip.native-ingestion.blocked-stop-noise.v1', protocol: PROTOCOL, baselineVariant, + workloadSha256: WORKLOAD_SHA256, precisionRule: PRECISION_RULE, precisionRuleSha256: PRECISION_RULE_SHA256, + sampleCounts: { pairs: 5, arms: 10 }, suitable: false, noiseFrozen: false, metrics: {}, + arms: chronological.map(arm => ({ reportSha256: hashReport(arm.evidence), startedAtMs: arm.evidence.startedAtMs, + finishedAtMs: arm.evidence.finishedAtMs, values: arm.values, observerBrackets: arm.brackets })), + rawArms: chronological.map(arm => arm.evidence), pairBindings: validated.map(pair => pair.map(arm => hashReport(arm.evidence))), + sourceDigestScope: 'SHA256 of JSON.stringify(parsed raw arm); owning raw-byte seals are separately required' }; + for (const metric of METRICS) { + const values = chronological.map(arm => arm.values[metric]), pooledMedianMs = median(values); + const baselineValues = validated.map(pair => pair[0].values[metric]), candidateValues = validated.map(pair => pair[1].values[metric]); + const baselineMedianMs = median(baselineValues), candidateMedianMs = median(candidateValues); + const differencesMs = validated.map(pair => pair[1].values[metric] - pair[0].values[metric]); + const maximumAbsolutePairedDifferenceMs = Math.max(...differencesMs.map(Math.abs)); + const firstFiveMedianMs = median(values.slice(0,5)), lastFiveMedianMs = median(values.slice(5)); + const absoluteChronologicalMedianDriftMs = Math.abs(lastFiveMedianMs - firstFiveMedianMs); + const precisionBudgetMs = Math.max(PRECISION_RULE.floorMs, PRECISION_RULE.pooledMedianFraction * pooledMedianMs); + const observerBracketWidthsMs = chronological.map(arm => arm.brackets[metric].widthMs); + const floorDominated = PRECISION_RULE.pooledMedianFraction * pooledMedianMs <= PRECISION_RULE.floorMs; + result.metrics[metric] = { sampleCount: 10, pairCount: 5, pooledMedianMs, baselineMedianMs, candidateMedianMs, + absoluteMedianDifferenceMs: Math.abs(candidateMedianMs - baselineMedianMs), + medianDifferencePercent: floorDominated ? null : 100 * (candidateMedianMs - baselineMedianMs) / baselineMedianMs, + ...spread(values), pairedDifferencesMs: differencesMs, maximumAbsolutePairedDifferenceMs, + firstFiveMedianMs, lastFiveMedianMs, absoluteChronologicalMedianDriftMs, precisionBudgetMs, + observerBracketWidthsMs, maximumObserverBracketMs: Math.max(...observerBracketWidthsMs), + suitable: maximumAbsolutePairedDifferenceMs <= precisionBudgetMs && absoluteChronologicalMedianDriftMs <= precisionBudgetMs + && observerBracketWidthsMs.every(width => width <= precisionBudgetMs), + noiseMs: null, floorDominated, resolutionLabel: floorDominated ? '100 ms resolution floor' : 'Observer brackets and budget reported', + subfloorOrP99ClaimsAllowed: false, percentageSpeedClaimsAllowed: !floorDominated }; + } + result.suitable = METRICS.every(metric => result.metrics[metric].suitable); + result.noiseFrozen = result.suitable; + if (result.suitable) for (const metric of METRICS) result.metrics[metric].noiseMs = result.metrics[metric].maximumAbsolutePairedDifferenceMs; + return result; +} +export function assertNoiseEnvelope(noise) { + assert.equal(noise?.schema, 'paperclip.native-ingestion.blocked-stop-noise.v1'); + assert.equal(noise.protocol, PROTOCOL); assert.equal(noise.workloadSha256, WORKLOAD_SHA256); + assert.equal(noise.precisionRuleSha256, PRECISION_RULE_SHA256); assert.deepEqual(noise.precisionRule, PRECISION_RULE); + assert.deepEqual(noise.sampleCounts, { pairs: 5, arms: 10 }); + assert.equal(noise.suitable, true, 'Unsuitable A/A cannot freeze a noise envelope'); assert.equal(noise.noiseFrozen, true); + assert.equal(noise.rawArms?.length, 10); + // Recompute suitability and all numeric limits from retained raw arms. A + // caller cannot inflate a budget or flip a failed suitability flag. + const byDigest = new Map(noise.rawArms.map(arm => [hashReport(arm), arm])); + assert.equal(noise.pairBindings?.length, 5); + const logical = noise.pairBindings.map(pair => { + assert.equal(pair.length, 2); + return pair.map(digest => { const raw = byDigest.get(digest); assert(raw, 'Frozen arm digest mismatch'); return raw; }); + }); + const recomputed = noiseEnvelope(logical); + assert.equal(recomputed.suitable, true); + assert.deepEqual(noise.metrics, recomputed.metrics, 'Frozen noise or precision limit changed'); + assert.equal(noise.baselineVariant, recomputed.baselineVariant); + assert.deepEqual(noise.arms, recomputed.arms, 'Frozen chronological arms changed'); + return noise; +} +export function comparePairs(pairs, noise) { + const validated = validatedPairs(pairs); assertNoiseEnvelope(noise); + for (const [baseline] of validated) assert.equal(String(baseline.evidence.variant).toLowerCase(), noise.baselineVariant, + 'Comparisons must use their own baseline calibration'); + return Object.fromEntries(METRICS.map(metric => { + const calibrated = noise.metrics[metric], deltas = validated.map(([baseline,candidate]) => candidate.values[metric] - baseline.values[metric]); + const baselineValues = validated.map(pair => pair[0].values[metric]), candidateValues = validated.map(pair => pair[1].values[metric]); + const baselineMedianMs = median(baselineValues), candidateMedianMs = median(candidateValues); + const observerBracketWidthsMs = validated.map(pair => pair.map(arm => arm.brackets[metric].widthMs)); + const floorDominated = calibrated.floorDominated || PRECISION_RULE.pooledMedianFraction * Math.min(baselineMedianMs,candidateMedianMs) <= PRECISION_RULE.floorMs; + return [metric, { sampleCounts: { pairs: 5, arms: 10 }, baselineMedianMs, candidateMedianMs, deltaMedianMs: median(deltas), + medianDifferencePercent: floorDominated ? null : 100 * (candidateMedianMs - baselineMedianMs) / baselineMedianMs, + deltasMs: deltas, noiseMs: calibrated.noiseMs, precisionBudgetMs: calibrated.precisionBudgetMs, + baselineSpread: spread(baselineValues), candidateSpread: spread(candidateValues), ...spread(deltas), + observerBracketWidthsMs, floorDominated, resolutionLabel: floorDominated ? '100 ms resolution floor' : 'Observer brackets and budget reported', + subfloorOrP99ClaimsAllowed: false, percentageSpeedClaimsAllowed: !floorDominated, + pass: deltas.every(delta => delta <= calibrated.noiseMs) && median(deltas) <= calibrated.noiseMs + && observerBracketWidthsMs.flat().every(width => width <= calibrated.precisionBudgetMs) }]; + })); +} +function assertOptionalShared(row, companyId, runId) { + assert.equal(row.company_id, companyId); assert.equal(row.run_id, runId); + const timestamp = value => typeof value === 'string' && /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z$/.test(value) && Number.isFinite(Date.parse(value)); + const execution = row.shared_execution_cancellation; + if (execution != null) { + assert(execution && typeof execution === 'object' && !Array.isArray(execution)); + assert(['requested','acknowledged'].includes(execution.state)); + if (execution.state === 'requested') assert(timestamp(execution.requestedAt)); + if (execution.state === 'acknowledged') assert(timestamp(execution.acknowledgedAt)); + if (execution.requestedAt != null) assert(timestamp(execution.requestedAt)); + if (execution.acknowledgedAt != null) assert(timestamp(execution.acknowledgedAt)); + if (execution.requestedAt != null && execution.acknowledgedAt != null) + assert(Date.parse(execution.requestedAt) <= Date.parse(execution.acknowledgedAt)); + } + const cancellation = row.run_cancellation; + if (cancellation != null) { + assert.equal(cancellation.source, 'operator'); assert.equal(cancellation.expected, true); + assert.equal(cancellation.initiator?.type, 'user'); + assert(typeof cancellation.reason === 'string' && cancellation.reason.length > 0); + assert(timestamp(cancellation.recordedAt)); + } +} +export function assertFunctional(e) { + metrics(e); observerBrackets(e); + assert.equal(e.prelude?.individualDeltas, WORKLOAD.prelude); assert.equal(e.prelude?.uniqueIdentities, WORKLOAD.prelude); + assert(e.prelude.verifiedBeforeEpochAtMs < e.epochMs); + assert(e.visibleBeforeLock?.length >= WORKLOAD.visibleBeforeLock); + assert.equal(e.lock?.id, e.runId); assert.equal(e.lock.company_id, e.companyId); + assert(e.appendBlockedBeforeStop?.observedAtMs < e.timing.requestAtMs); + assert(e.appendBlockedBeforeStop.rows.some(row => isAppendLockQuery(row.query) && row.blocking_pids.includes(e.lock.holder_pid))); + assert(e.heldSamples?.length >= 2); + for (const sample of e.heldSamples) { + assertHeldSample(sample, e.lock.holder_pid); assertOptionalShared(sample, e.companyId, e.runId); + assert(sample.queryStartedAtMs >= e.timing.requestAtMs && sample.observedAtMs <= e.timing.releaseRequestedAtMs); + } + assert(e.heldSamples.at(-1).emittedCount > e.heldSamples[0].emittedCount, 'Provider must still emit through the stall'); + assertIntentAudit(e.ack); assertOptionalShared(e.ack, e.companyId, e.runId); + assert.equal(e.ack.intent.dispatchState, 'acknowledged'); + if (e.sharedSettlement != null) { + assert.equal(e.sharedSettlement.runId, e.runId); assert.equal(e.sharedSettlement.companyId, e.companyId); + assert.equal(e.sharedSettlement.status, 'cancelled'); + assertOptionalShared({ company_id: e.sharedSettlement.companyId, run_id: e.sharedSettlement.runId, + shared_execution_cancellation: e.sharedSettlement.executionCancellation, run_cancellation: e.sharedSettlement.cancellation }, e.companyId, e.runId); + if (e.sharedSettlement.nativeCancellation != null) assert.equal(e.sharedSettlement.nativeCancellation.intentId, e.ack.intent.intentId); + } + assert.equal(e.ack.company_id, e.companyId); assert.equal(e.ack.run_id, e.runId); assert.equal(e.ack.issue_id, e.issueId); + assert.equal(e.stopHttpStatus, 200); + assert.deepEqual([e.processesAfter?.runnerAbsent, e.processesAfter?.groupAbsent, e.processesAfter?.providerAbsent], [true,true,true]); + assert(e.timing.httpSettledAtMs >= e.timing.releaseRequestedAtMs); + const before = e.authorityBefore?.rows?.[0], after = e.authorityAfter?.rows?.[0]; + for (const row of [before, after]) { assert.equal(row?.company_id, e.companyId); assert.equal(row?.run_id, e.runId); assert.equal(row.runtime_mode, 'native'); } + assert.equal(after.status, 'cancelled'); assert.equal(after.cancellation_dispatch_state, 'acknowledged'); + assert(BigInt(after.process_stopped_seq) > BigInt(before.process_identity_seq)); + assert.equal(after.process_stopped_source_event_id, null); + assert.equal(after.process_stopped.processPid, e.processesBefore.runner.pid); + assert.equal(after.process_stopped.processGroupId, e.processesBefore.group); + for (const process of [e.processesBefore.runner, e.processesBefore.provider]) { + assert(Number.isSafeInteger(process.pid) && process.pid > 1 && process.birth); + assert.equal(process.group, e.processesBefore.group); + } + assert.equal(e.lease?.heartbeatRunId, e.runId); assert.equal(e.lease.companyId, e.companyId); + assert.equal(e.lease.provider, 'local'); assert.equal(e.lease.providerLeaseId, null); + assert(e.lease.releasedAt && ['released','expired','failed'].includes(e.lease.status)); assert.notEqual(e.lease.cleanupStatus, 'failed'); + const transcript = e.transcript; + assert.equal(transcript?.persistedAfterReload, true); assert.equal(transcript?.ordered, true); + assert.equal(transcript.resultAcceptedEvents, 0); assert.equal(transcript.preludeCount, WORKLOAD.prelude); + assert.equal(transcript.individualDeltas, transcript.deltas.length); + assert.equal(transcript.uniqueSourceIdentities, transcript.deltas.length); + const identities = new Set(); + for (const [index,event] of transcript.deltas.entries()) { + assert.equal(event.runId, e.runId); assert.equal(event.eventType, 'item.delta'); + assert(event.sourceInstanceId && event.sourceEventId && Number.isSafeInteger(event.sourceSeq) && event.sourceSeq > 0); + identities.add(`${event.sourceInstanceId}:${event.sourceEventId}:${event.sourceSeq}`); + assert.equal(event.payload.text, `[${e.clock.markerCode}${index.toString(36)}] `); + assert(e.clock.records.some(record => record.index === index && record.nonce === e.clock.nonce)); + } + assert.equal(identities.size, transcript.deltas.length); + for (const marker of transcript.seen) assert(marker.index >= 0 && marker.index < transcript.deltas.length); + assert(e.clock.ended || e.clock.aborted); assert(!e.clock.error); + assert.equal(e.measurement?.actionToAckIncludesForcedWait, true); + assert.equal(e.measurement?.recording?.trace, 'off'); assert.equal(e.measurement?.recording?.video, 'off'); + return metrics(e); +} diff --git a/tests/e2e/fixtures/native-ingestion-blocked-stop-protocol.test.mjs b/tests/e2e/fixtures/native-ingestion-blocked-stop-protocol.test.mjs new file mode 100644 index 00000000000..26e7c5c0991 --- /dev/null +++ b/tests/e2e/fixtures/native-ingestion-blocked-stop-protocol.test.mjs @@ -0,0 +1,206 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { PROTOCOL, WORKLOAD as W, WORKLOAD_SHA256, METRICS, isAppendLockQuery, assertHeldSample, + assertIntentAudit, metrics, registration, noiseEnvelope, comparePairs, assertFunctional, observerBrackets, assertNoiseEnvelope, PRECISION_RULE } from './native-ingestion-blocked-stop-protocol.mjs'; +const evidence = (shift = 0) => ({ protocol: PROTOCOL, workloadSha256: WORKLOAD_SHA256, complete: true, cleanup: { complete: true }, + timing: { actionAtMs: 1000, requestAtMs: 1005, releaseRequestedAtMs: 2005, releaseCommitAckAtMs: 2007, + absenceObservedAtMs: 2030 + shift, ackObservedAtMs: 2040 + shift } }); +const sample = () => ({ httpPending: true, intent: null, auditIds: [], processesAlive: true, clockOpen: true, blockers: [{ blocking_pids: [88] }] }); +const intentRow = () => ({ company_id: 'c', run_id: 'r', issue_id: 'i', intent: { + schema: 'paperclip.native-cancellation.v1', companyId: 'c', runId: 'r', issueId: 'i', + intentId: 'own', intentAuditId: 'a', dispatchState: 'acknowledged', acknowledgementAuditId: 'b', + scope: 'run', reasonCode: 'cancellation_run_only', dispatched: true, decisionId: null }, + audits: [{ id: 'a', action: 'native.cancellation_intent_recorded', details: { intentId: 'own' } }, + { id: 'b', action: 'native.cancellation_dispatch_acknowledged', details: { intentId: 'own', intentAuditId: 'a', dispatched: true, decisionId: null } }] }); + +test('geometry fits the unchanged 4,000 character bound, includes existing emitted history', () => { + const chars = '.'.repeat(W.prelude - 1) + ` PRELUDE_${'a'.repeat(32)}_READY ` + + Array.from({ length: W.events }, (_, index) => `[00${index.toString(36)}] `).join(''); + assert(chars.length < 4000); assert.equal(W.prelude, 1000); assert.equal(W.events, 256); assert.equal(W.intervalMs, 100); +}); +test('B0 and candidate append SQL match while cancellation full-row and unrelated SQL do not', () => { + for (const extra of ['', ', "next_event_seq", "event_cursor_projection_seq"']) assert(isAppendLockQuery( + `select "company_id", "agent_id"${extra} from "heartbeat_runs" where "heartbeat_runs"."id" = $1 limit $2 for update`)); + assert(!isAppendLockQuery('select "id", "company_id", "agent_id" from "heartbeat_runs" where "heartbeat_runs"."id" = $1 for update')); + assert(!isAppendLockQuery('select "company_id", "agent_id" from "heartbeat_runs" where "heartbeat_runs"."id" = $1')); +}); +test('held sample requires every barrier and physical liveness proof', () => { + assertHeldSample(sample(), 88); + for (const patch of [{ httpPending: false }, { intent: {} }, { auditIds: ['a'] }, { processesAlive: false }, { clockOpen: false }, { blockers: [] }]) + assert.throws(() => assertHeldSample({ ...sample(), ...patch }, 88)); + assert.throws(() => assertHeldSample(sample(), 89)); +}); +test('ACK correlates both committed audits to the same owned intent', () => { + assertIntentAudit(intentRow()); + for (const mutate of [row => { row.intent.runId = 'foreign'; }, row => { row.audits.pop(); }, + row => { row.audits[1].details.intentId = 'foreign'; }, row => { row.intent.intentAuditId = 'missing'; }]) { + const row = intentRow(); mutate(row); assert.throws(() => assertIntentAudit(row)); + } +}); +test('full action latency retains forced wait; release metrics use COMMIT ACK and also retain request-origin bounds', () => { + assert.deepEqual(metrics(evidence()), { actionToAckMs: 1040, releaseAckToAbsenceMs: 23, releaseAckToAckMs: 33, + releaseRequestToAbsenceMs: 25, releaseRequestToAckMs: 35 }); + for (const key of ['actionAtMs','requestAtMs','releaseRequestedAtMs','releaseCommitAckAtMs','absenceObservedAtMs','ackObservedAtMs']) { + const e = evidence(); delete e.timing[key]; assert.throws(() => metrics(e)); + } + const e = evidence(); e.timing.releaseRequestedAtMs--; assert.throws(() => metrics(e)); + e.timing.releaseRequestedAtMs++; e.complete = false; assert.throws(() => metrics(e)); +}); +test('registered companion is 50 separate alternating arms with fresh B0 and B1 A/A', () => { + const arms = registration(); assert.equal(arms.length, 50); + for (const stage of ['b0-aa','b1-aa','b0-b1','b1-b2','b0-b2']) { + const rows = arms.filter(a => a.stage === stage); assert.equal(rows.length, 10); + for (let pair = 1; pair <= 5; pair++) assert.equal(rows.filter(a => a.pair === pair).length, 2); + assert.deepEqual(rows.slice(0, 4).map(a => a.role), ['baseline','candidate','candidate','baseline']); + } +}); + +const functional = () => { + const e = evidence(), identity = { company_id: 'c', run_id: 'r', runtime_mode: 'native' }; + return { ...e, variant: 'b0', startedAtMs: 100, finishedAtMs: 3000, companyId: 'c', runId: 'r', issueId: 'i', epochMs: 900, + prelude: { individualDeltas: 1000, uniqueIdentities: 1000, verifiedBeforeEpochAtMs: 850 }, + visibleBeforeLock: [0,1,2], lock: { id: 'r', company_id: 'c', holder_pid: 88 }, + appendBlockedBeforeStop: { observedAtMs: 1004, rows: [{ query: 'select "company_id", "agent_id" from "heartbeat_runs" where "heartbeat_runs"."id" = $1 for update', blocking_pids: [88] }] }, + heldSamples: [{ ...sample(), company_id: 'c', run_id: 'r', queryStartedAtMs: 1010, observedAtMs: 1015, emittedCount: 3 }, + { ...sample(), company_id: 'c', run_id: 'r', queryStartedAtMs: 1995, observedAtMs: 2004, emittedCount: 13 }], ack: intentRow(), stopHttpStatus: 200, + timing: { ...e.timing, httpSettledAtMs: 2042, lastAckNotObservedAtMs: 2036, lastProcessesPresentAtMs: 2026 }, + observerSamples: [{ kind: 'ack-query', startedAtMs: 2035, completedAtMs: 2036, durationMs: 1, value: false }, + { kind: 'ack-query', startedAtMs: 2039, completedAtMs: 2040, durationMs: 1, value: true }, + { kind: 'process-check', startedAtMs: 2025, completedAtMs: 2026, durationMs: 1, value: false }, + { kind: 'process-check', startedAtMs: 2029, completedAtMs: 2030, durationMs: 1, value: true }], processesBefore: { runner: { pid: 10, group: 10, birth: 'birth' }, + provider: { pid: 11, group: 10, birth: 'birth' }, group: 10 }, processesAfter: { runnerAbsent: true, groupAbsent: true, providerAbsent: true }, + authorityBefore: { rows: [{ ...identity, process_identity_seq: '1' }] }, + authorityAfter: { rows: [{ ...identity, status: 'cancelled', cancellation_dispatch_state: 'acknowledged', process_stopped_seq: '20', + process_stopped_source_event_id: null, process_stopped: { processPid: 10, processGroupId: 10 } }] }, + lease: { heartbeatRunId: 'r', companyId: 'c', provider: 'local', providerLeaseId: null, releasedAt: 'time', status: 'released', cleanupStatus: 'success' }, + transcript: { individualDeltas: 3, uniqueSourceIdentities: 3, persistedAfterReload: true, ordered: true, resultAcceptedEvents: 0, + preludeCount: 1000, seen: [{ index: 0 },{ index: 1 },{ index: 2 }], deltas: [0,1,2].map(index => ({ runId: 'r', eventType: 'item.delta', + sourceInstanceId: 'source', sourceEventId: `id${index}`, sourceSeq: index+1, payload: { text: `[00${index}] ` } })) }, + clock: { markerCode: '00', nonce: 'nonce', ended: true, records: [0,1,2].map(index => ({ index, nonce: 'nonce' })) }, + measurement: { actionToAckIncludesForcedWait: true, recording: { trace: 'off', video: 'off' } } }; +}; +test('functional oracle accepts complete correlated proof and rejects plausible missing or wrong outcomes', () => { + assertFunctional(functional()); + const mutations = [e => { delete e.ack; }, e => { e.ack.intent.intentId = 'other'; }, + e => { e.appendBlockedBeforeStop.rows[0].blocking_pids = []; }, e => { e.prelude.individualDeltas = 999; }, + e => { e.authorityAfter.rows[0].process_stopped.processPid = 99; }, e => { e.processesAfter.providerAbsent = false; }, + e => { e.heldSamples[1].emittedCount = 3; }, e => { e.transcript.deltas[1].payload.text = '[000] '; }, + e => { e.transcript.deltas[1].sourceEventId = 'id0'; e.transcript.deltas[1].sourceSeq = 1; }, + e => { e.transcript.persistedAfterReload = false; }, e => { e.measurement.recording.trace = 'on'; }]; + for (const mutate of mutations) { const e = functional(); mutate(e); assert.throws(() => assertFunctional(e)); } +}); + +test('shared decision is correlated when present and absent run-only decisions stay valid', () => { + const row = intentRow(); assertIntentAudit(row); + row.intent.decisionId = 'decision'; row.audits[1].details.decisionId = 'decision'; + row.shared_decision = { id: 'decision', companyId: 'c', runId: 'r', issueId: 'i' }; assertIntentAudit(row); + row.shared_decision.runId = 'foreign'; assert.throws(() => assertIntentAudit(row)); +}); + +const arm = (index, shift = 0, variant = 'b0') => { + const offset = index * 10000; + const evidence = JSON.parse(JSON.stringify(functional()), (key,value) => { + if (value === 'r') return `r${index}`; if (value === 'c') return `c${index}`; if (value === 'i') return `i${index}`; + return typeof value === 'number' && (key.endsWith('AtMs') || key === 'epochMs') ? value + offset : value; + }); + evidence.variant = variant; + for (const key of ['ackObservedAtMs','absenceObservedAtMs','lastAckNotObservedAtMs','lastProcessesPresentAtMs','httpSettledAtMs']) + evidence.timing[key] += shift; + for (const sample of evidence.observerSamples) { sample.startedAtMs += shift; sample.completedAtMs += shift; } + return evidence; +}; +const pairs = (shift = () => 0) => Array.from({ length: 5 }, (_, pair) => [ + arm(2*pair + (pair % 2), shift(pair,0)), arm(2*pair + (pair % 2 ? 0 : 1), shift(pair,1))]); + +test('noise artifact freezes actual suitable A/A limits with counts, spread, chronological drift and resolution labels', () => { + const noise = noiseEnvelope(pairs((pair,side) => side ? pair+1 : 0)); + assert.equal(noise.suitable, true); assert.equal(noise.noiseFrozen, true); assertNoiseEnvelope(noise); + assert.deepEqual(noise.sampleCounts, { pairs: 5, arms: 10 }); assert.equal(noise.rawArms.length, 10); + for (const metric of METRICS) { + const row = noise.metrics[metric]; assert.equal(row.noiseMs, 5); assert.equal(row.sampleCount, 10); + assert.equal(row.precisionBudgetMs, Math.max(2*W.pollIntervalMs, 0.10*row.pooledMedianMs)); + assert(row.maximumObserverBracketMs <= row.precisionBudgetMs); + } + assert.equal(noise.metrics.releaseAckToAckMs.resolutionLabel, '100 ms resolution floor'); + assert.equal(noise.metrics.releaseAckToAckMs.medianDifferencePercent, null); + assert.equal(noise.metrics.releaseAckToAckMs.percentageSpeedClaimsAllowed, false); +}); +test('every calibration and comparison arm must pass the complete functional oracle', () => { + const noise = noiseEnvelope(pairs()); + const mutations = [e => { delete e.lock; }, e => { delete e.ack; }, e => { delete e.processesAfter; }, + e => { delete e.lease; }, e => { delete e.transcript; }, e => Object.assign(e, evidence()), + e => { e.transcript.persistedAfterReload = false; }]; + for (const mutate of mutations) { + const invalid = pairs(); mutate(invalid[4][1]); + assert.throws(() => noiseEnvelope(invalid)); assert.throws(() => comparePairs(invalid, noise)); + } + const forged = pairs(); forged[0][0] = evidence(); + assert.throws(() => noiseEnvelope(forged)); assert.throws(() => comparePairs(forged, noise)); +}); +test('paired jitter and chronological drift outside prospective budget retain all arms and cannot freeze noise', () => { + for (const data of [pairs((pair,side) => side ? 150 : 0), pairs(pair => pair * 60)]) { + const noise = noiseEnvelope(data); + assert.equal(noise.suitable, false); assert.equal(noise.noiseFrozen, false); assert.equal(noise.rawArms.length, 10); + assert(METRICS.every(metric => noise.metrics[metric].noiseMs === null)); + assert.throws(() => assertNoiseEnvelope(noise)); assert.throws(() => comparePairs(pairs(), noise)); + } + const drift = noiseEnvelope(pairs(pair => pair * 60)); + assert.equal(drift.metrics.releaseAckToAckMs.maximumAbsolutePairedDifferenceMs, 0); + assert(drift.metrics.releaseAckToAckMs.absoluteChronologicalMedianDriftMs > drift.metrics.releaseAckToAckMs.precisionBudgetMs); +}); +test('raw observer brackets retain actual query/check durations and reject missing or forged timing', () => { + const e = functional(), brackets = observerBrackets(e); + assert.equal(brackets.releaseAckToAckMs.widthMs, 5); + assert.equal(brackets.releaseAckToAckMs.negativeDurationMs, 1); + assert.equal(brackets.releaseAckToAbsenceMs.successfulDurationMs, 1); + e.observerSamples[0].durationMs = 0; assert.throws(() => observerBrackets(e)); + delete e.observerSamples; assert.throws(() => assertFunctional(e)); +}); +test('a wide observer bracket invalidates A/A suitability and comparison despite stable latency medians', () => { + const data = pairs(() => 120), e = data[0][0]; + e.observerSamples = e.observerSamples.filter(sample => sample.kind !== 'ack-query' || sample.value); + delete e.timing.lastAckNotObservedAtMs; + const unsuitable = noiseEnvelope(data); + assert.equal(unsuitable.metrics.releaseAckToAckMs.maximumAbsolutePairedDifferenceMs, 0); + assert.equal(unsuitable.suitable, false); assert.throws(() => assertNoiseEnvelope(unsuitable)); + const noise = noiseEnvelope(pairs(() => 120)); + const result = comparePairs(data, noise); + assert.equal(result.releaseAckToAckMs.pass, false); + assert.equal(result.releaseAckToAckMs.noiseMs, 0); +}); +test('comparison uses unchanged frozen noise and budget; one paired regression or inflated artifact fails', () => { + const calibration = pairs((pair,side) => side ? pair+1 : 0), noise = noiseEnvelope(calibration); + assert(Object.values(comparePairs(calibration, noise)).every(row => row.pass)); + const regression = pairs((pair,side) => side ? 6 : 0); + assert(Object.values(comparePairs(regression, noise)).every(row => !row.pass)); + const forged = structuredClone(noise); forged.metrics.releaseAckToAckMs.noiseMs = 999; + assert.throws(() => comparePairs(pairs(), forged)); + const inflated = structuredClone(noise); inflated.metrics.releaseAckToAckMs.precisionBudgetMs = 999; + assert.throws(() => comparePairs(pairs(), inflated)); + assert.throws(() => noiseEnvelope(pairs().slice(1))); +}); +test('optional shared cancellation accepts absent/requested B0 metadata and validates present state, time, attribution and run scope', () => { + const when = '2026-11-04T10:00:00.000Z'; + const e = functional(); assertFunctional(e); + e.ack.shared_execution_cancellation = { state: 'requested', requestedAt: when }; + e.ack.run_cancellation = { source: 'operator', expected: true, initiator: { type: 'user' }, reason: 'Operator Stop', recordedAt: when }; + assertFunctional(e); + e.ack.shared_execution_cancellation = { state: 'acknowledged', acknowledgedAt: when }; assertFunctional(e); + for (const patch of [{ state: 'requested' }, { state: 'acknowledged', acknowledgedAt: 'invalid' }, + { state: 'other', requestedAt: when }]) { + const bad = structuredClone(e); bad.ack.shared_execution_cancellation = patch; assert.throws(() => assertFunctional(bad)); + } + const wrong = structuredClone(e); wrong.ack.run_cancellation.source = 'provider'; assert.throws(() => assertFunctional(wrong)); + const foreign = structuredClone(e); foreign.sharedSettlement = { companyId: 'c', runId: 'foreign', status: 'cancelled' }; + assert.throws(() => assertFunctional(foreign)); +}); + +test('calibration and comparison reject reused runs, overlapping arms and nonalternating chronology', () => { + const noise = noiseEnvelope(pairs()); + for (const corrupt of [data => { data[4][1] = data[0][0]; }, + data => { data[0][1].startedAtMs = data[0][0].finishedAtMs; }, + data => { data[1].reverse(); }]) { + const invalid = pairs(); corrupt(invalid); + assert.throws(() => noiseEnvelope(invalid)); assert.throws(() => comparePairs(invalid, noise)); + } +}); diff --git a/tests/e2e/fixtures/native-ingestion-blocked-stop-validate.mjs b/tests/e2e/fixtures/native-ingestion-blocked-stop-validate.mjs new file mode 100644 index 00000000000..e6eb8383dd6 --- /dev/null +++ b/tests/e2e/fixtures/native-ingestion-blocked-stop-validate.mjs @@ -0,0 +1,22 @@ +// Read-only validation of an independently extracted raw attachment; no app/DB/process execution. +import { readFile, writeFile } from 'node:fs/promises'; +import { createHash } from 'node:crypto'; +import { pathToFileURL } from 'node:url'; +import { assertFunctional, PROTOCOL, WORKLOAD_SHA256, registration } from './native-ingestion-blocked-stop-protocol.mjs'; +export async function validateFile(file) { + const bytes = await readFile(file), evidence = JSON.parse(bytes.toString('utf8')); + const metrics = assertFunctional(evidence); + return { protocol: PROTOCOL, workloadSha256: WORKLOAD_SHA256, sourceFile: file, + sourceSha256: createHash('sha256').update(bytes).digest('hex'), sourceRevision: evidence.sourceRevision, + runId: evidence.runId, metrics, functionalPass: true, + qualification: 'Functional gate only; owning Playwright, frozen preflight, cleanup and paired noise/comparison campaign remain separately required' }; +} +if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) { + if (process.argv[2] === '--registration') process.stdout.write(`${JSON.stringify(registration(), null, 2)}\n`); + else { + if (!process.argv[2]) throw new Error('Pass extracted raw companion evidence JSON, or --registration'); + const result = await validateFile(process.argv[2]); + if (process.argv[3]) await writeFile(process.argv[3], `${JSON.stringify(result, null, 2)}\n`, { flag: 'wx' }); + else process.stdout.write(`${JSON.stringify(result, null, 2)}\n`); + } +} diff --git a/tests/e2e/fixtures/native-ingestion-blocked-stop.md b/tests/e2e/fixtures/native-ingestion-blocked-stop.md new file mode 100644 index 00000000000..d26fc8eb16e --- /dev/null +++ b/tests/e2e/fixtures/native-ingestion-blocked-stop.md @@ -0,0 +1,184 @@ +# Existing-history blocked Stop companion (prospective v1) + +Status: local source and pure oracle calibration only. No application, browser, +PostgreSQL, VM or quantitative arm has run. No runtime qualification or performance +claim follows from the pure tests. Execution needs lead release after independent +source review. This is separate from the immutable H5c 50-arm campaign. + +## Registration and ownership + +Explicit config: `tests/e2e/playwright-native-ingestion-blocked-stop.config.ts`. +Explicit source: `tests/e2e/native-event-ingestion-blocked-stop.journey.ts`. +Protocol: `blocked-stop-history-v1`; workload digest comes from the exact exported +`WORKLOAD` JSON in `native-ingestion-blocked-stop-protocol.mjs`. + +The config inherits the unchanged H5c production-built app bootstrap, local +Codex command fixture, normal local environment, and shipped UI. It selects only +the companion source and fixes a 900-second test budget. There is one owned +control agent and one task created and assigned through visible normal controls. +Provider execution is a deterministic existing Codex leaf fixture, not a live +model or a real Cloud provider. This companion grants no Cloud Stop qualification. +No resolver override, HTTP/API change, persisted outcome seeding or app/product +code change is included. Authority/gate/paint code is a bounded copy from H5c +`e431ec63d41fe90c6fd68bd2fc2bb969feeca917`; the original nine measured files stay +unchanged. The new files have their own source identity and review. + +## Fixed workload and functional gate + +All variants emit 1,000 actual prelude deltas, independently read back through +normal event APIs before the common gate releases. They then emit 256 compact +markers at 100 ms, fitting the unchanged 4,000 character Rust text bound together +with the prelude. Readiness is bounded at 300 seconds, measured streaming at +120 seconds, terminal settlement at 180 seconds; original later 30-second ACK +and lease observations remain after settlement. Setup stays inside the owning +900-second test budget. Server bootstrap retains its existing separate limit. + +After at least three visible markers and before Stop, a separately owned +PostgreSQL connection acquires `FOR UPDATE` on the exact company/run row. A +second connection reads `pg_stat_activity` and `pg_blocking_pids` and must see +an actual append query waiting on that holder. Its SELECT projection prefix is +checked against the B0/B1/B2 source signatures, distinguishing cancellation's +full-row SELECT. Raw SQL metadata is retained privately; no product hook is used. + +The ordinary visible Stop button must cause the actual matching public POST. +For at least 1,000 ms after that observed request, repeated independent snapshots +must show pending HTTP, no committed native cancellation intent or audit, the +same runner/provider PID births alive in their exact group, and an open fixture +clock. The fixture closes that clock on `turn/interrupt`. Provider emissions must +continue across the hold. The hold is not subtracted from action-to-ACK latency. +A late sample/query/timer can overrun the minimum; actual boundaries stay visible. + +Release records both COMMIT request and COMMIT response acknowledgement. Neither +is claimed to be the exact PostgreSQL commit instant. Independent observations +then require the cancellation intent and its own matching committed audit, +native acknowledged dispatch and its own matching acknowledgement audit, +normal cancelled settlement, successful Stop HTTP, server-authored +`native.local_process_stopped` receipt for the exact prior PID/group, independently +absent runner/group/provider, and the exact owned local lease released without +failed cleanup. Transcript checks require complete compact payloads, unique source +identities, ordered canonical prefix, original emission clocks, zero accepted +results, and persistence after reload. The raw transport recovery policy stays +unchanged: raw fixture emissions after the canonical prefix are not represented +as universally database-durable. + +The native receipt, shared `executionCancellation` metadata (including absence) +and public cancelled run are recorded separately. The native dispatch audit must +match the same intent and intent audit, `dispatched: true`, and nullable decision +ID. A non-null decision is independently correlated through the owned +`status_decisions` company/issue/run row. Run-only Stop may have no decision; a +missing optional decision or shared field is not silently invented. A shared ACK field is not synthesized from +row `recordedAt` or a provider clock. Source/runtime review must confirm which +shared receipt contract the serving revision exposes before releasing the probe. +No universal 30-second click-to-ACK gate is introduced. + +Every actual ACK query and process check through its first success retains its +start, completion, duration and boolean observation. An uncertainty bracket runs +from the last negative query/check start (conservative for snapshot-read timing) +or release-COMMIT-ACK when no negative was seen, through first successful +completion. Both negative and successful operation durations stay explicit. + +Timing includes action-to-observed-native-ACK, release-COMMIT-ACK-to-independently +observed absence, and release-COMMIT-ACK-to-observed-native-ACK. Both release +request-origin values also remain. Last unsuccessful and successful-query/check +bounds are retained with a 50-ms poll interval; query duration and scheduling add +uncertainty. Screenshots and reload happen after those timing observations. + +Any failure, timeout, censored observation, query/clock fault or cleanup error +makes the arm ineligible. Partial raw Playwright results and the final attachment +remain even after failure. Cleanup first releases/rolls back only the owned lock, +closes both connections, retires its clock, settles only its owned run and restores +owned experimental settings. The external operator must independently prove the +server, database, runner, provider and browser are gone before the next arm. + +## Separate quantitative registration (50 short arms) + +Generate the explicit registration without executing an app: + +```sh +node tests/e2e/fixtures/native-ingestion-blocked-stop-validate.mjs --registration +``` + +Order is B0 A/A five fresh alternating pairs, B1 A/A five fresh alternating pairs, +then B0→B1, B1→B2 and B0→B2 five alternating pairs each. Each pair is two fresh +owned invocations with matching data/load/payload/resource/cache conditions. +There are no repeats hidden inside an invocation. Keep the original main 50 arms +and their noise files untouched. + +Before any companion data, freeze this exact suitability rule independently per +metric. The precision budget is `max(2 * 50 ms, 0.10 * pooled ten-arm A/A metric +median)`. Both the maximum absolute paired difference and the absolute median +drift between the first five and last five chronological arms must fit that +budget. Every raw ACK/absence observer bracket must also fit the corresponding +metric budget; action-to-ACK uses the ACK observer bracket. This is a +measurement repeatability rule, not a product Stop deadline. + +Only suitable calibration freezes its observed maximum absolute paired +difference as noise. B0 comparisons use B0 A/A; B1→B2 uses B1 A/A. Every paired +candidate difference and their median must remain within that own frozen noise, +and all comparison observer brackets must fit the frozen baseline precision +budget. No unblocked Stop noise is reused. An unsuitable ten-arm calibration +retains every raw arm and produces no usable noise envelope: no silent +replacement, budget increase or inflated envelope follows from noisy data. + +`noiseEnvelope(pairs)` takes five logical `[baseline, candidate]` raw-report +pairs whose chronological order alternates baseline-first then candidate-first. +It validates the full functional oracle for every arm, requires ten distinct +owned runs/companies with nonoverlapping arm times, and returns the structured +`paperclip.native-ingestion.blocked-stop-noise.v1` artifact. This contains +sample counts, all raw arms and report digests, logical pair bindings, actual +absolute/percentage median differences, spread, chronological drift, budgets, +query/check uncertainty brackets, suitability, and nullable noise values. +Percentage fields are null when the 100-ms precision floor determines the +budget. Those metrics are labeled **100 ms resolution floor** and cannot support +subfloor, p99 or percentage speed claims. Ten observed arms do not establish p99. + +`assertNoiseEnvelope(artifact)` revalidates all retained functional proofs and +recomputes suitability and numeric limits; unsuitable or inflated artifacts +throw. `comparePairs(pairs, artifact)` validates every comparison arm's complete +functional proof and returns per-metric `.pass` results using only that +baseline's unchanged frozen noise and budget. It rejects reused runs, +nonalternating/overlapping arms, missing lock/ACK/process/lease/transcript proof, +and forged `complete` flags. Preserve all failures and censored arms outside +these successful artifacts as well. A single functional probe is not +quantitative nonregression evidence. + +## Reproduction and execution gates + +Credential-free local checks, with Node 24: + +```sh +node --test tests/e2e/fixtures/native-ingestion-blocked-stop-protocol.test.mjs +node --check tests/e2e/native-event-ingestion-blocked-stop.journey.ts +node --check tests/e2e/playwright-native-ingestion-blocked-stop.config.ts +node tests/e2e/fixtures/native-ingestion-blocked-stop-validate.mjs --registration +``` + +Before runtime, the lead must review/freeze the companion source, workload digest, +registration, validator and companion-only release descriptor. Keep the reviewed +H5c private runner `6b6376b01de6cdc15ca05343a0bdd5c00424f1d6c546b9b4d61cd76716b253c4` +unchanged. A companion orchestrator should import its checked exported +`validateVariantBuild`, `validateActualInventory`, `assertNoPreferredUiShadow`, +`physicalQuietGuard`, `sealArmArtifacts` and `verifyRawArtifactSeals`; it must not +invoke or alter the original stage plan. Bind each arm to exact source/build/UI, +qualified receipt, full inventory manifest, Node/dependency/native pins, unchanged +frozen fixture hashes, companion hashes, recording-off flags and a clean source. +Use original B0 `bac53ec3`, B1 `07061a60`, unoptimized B2 `ca7cba17` only where +explicitly released; a future optimized B2 needs its own qualified descriptor. +The existing build receipt `a1259ba7…`, full inventory `d3766fc4…` and native binary +`abb65bdc…` are provenance, not substitutes for current artifact validation. + +Only the performance owner may stage/execute on the sole benchmark VM, after +lead release. First do discovery, narrow static typecheck, and one smallest +functional qualification with raw Playwright, server/banner, PostgreSQL blocker, +process/clock and external cleanup evidence. Validate its extracted attachment: + +```sh +node tests/e2e/fixtures/native-ingestion-blocked-stop-validate.mjs RAW.json RECEIPT.json +``` + +Do not start the 50-arm registration until the source and runtime probe qualify. +Capture quiet before/after-arm receipts, owning Playwright exit/status/duration, +raw API/SQL/process evidence, immutable report seals, and all cleanup outcomes. +The source validator is an additional functional oracle; it cannot establish the +owning Playwright pass, infrastructure cleanup, build authenticity or performance +on its own. Freeze separate B0/B1 A/A noise before nonregression comparisons. diff --git a/tests/e2e/native-event-ingestion-blocked-stop.journey.ts b/tests/e2e/native-event-ingestion-blocked-stop.journey.ts new file mode 100644 index 00000000000..7bcf955c90b --- /dev/null +++ b/tests/e2e/native-event-ingestion-blocked-stop.journey.ts @@ -0,0 +1,312 @@ +import { randomUUID, createHash } from "node:crypto"; +import { execFileSync } from "node:child_process"; +import { readFile } from "node:fs/promises"; +import path from "node:path"; +import { expect, test, type Request } from "@playwright/test"; +import { json } from "./agent-chat.shared"; +import { runtimeAuthority, processAbsent } from "./fixtures/native-ingestion-blocked-stop-authority"; +import { openClock } from "./fixtures/native-ingestion-blocked-stop-clock"; +import { ownedDatabase, cancellationSnapshot, blockers } from "./fixtures/native-ingestion-blocked-stop-db"; +import { PROTOCOL, WORKLOAD as W, WORKLOAD_SHA256, assertHeldSample, assertIntentAudit, isAppendLockQuery, metrics, assertFunctional } + from "./fixtures/native-ingestion-blocked-stop-protocol.mjs"; +import { observeRunSettlement } from "./fixtures/native-ingestion-h5.mjs"; + +const delay = (ms: number) => new Promise(resolve => setTimeout(resolve, ms)); +const birth = (pid: number) => { + const value = execFileSync("ps", ["-p", String(pid), "-o", "pid=,pgid=,lstart="], { encoding: "utf8" }).trim(); + const match = value.match(/^(\d+)\s+(\d+)\s+(.+)$/); + if (!match || Number(match[1]) !== pid) throw new Error("Missing exact owned process birth identity"); + return { pid, group: Number(match[2]), birth: match[3] }; +}; + +test("existing-history Stop preserves its durable barrier behind a real blocked append", async ({ page, request }, info) => { + const appRoot = path.resolve(process.env.PAPERCLIP_INGESTION_APP_ROOT ?? ""); + expect(process.env.PAPERCLIP_INGESTION_SERVER_ARTIFACT).toBe("compiled"); + expect(process.env.PAPERCLIP_INGESTION_COMPANION_PROTOCOL).toBe(PROTOCOL); + const report: any = { protocol: PROTOCOL, workload: W, workloadSha256: WORKLOAD_SHA256, complete: false, + variant: process.env.PAPERCLIP_INGESTION_VARIANT, startedAtMs: Date.now(), timing: {}, heldSamples: [], observerSamples: [], + observedClock: "Independent observer wall clock; PostgreSQL row recordedAt is not a commit timestamp", + measurement: { pollIntervalMs: W.pollIntervalMs, releaseOrigin: "Owned COMMIT response acknowledgement", + releaseCommitBracket: "releaseRequestedAtMs..releaseCommitAckAtMs; exact server commit instant is unknown", + absenceAndAckUncertainty: "Last unsuccessful observation through successful query/process observation; polling plus query duration", + actionToAckIncludesForcedWait: true, noUniversalClickAckDeadline: true, + recording: { trace: info.project.use.trace, video: info.project.use.video }, quantitativeAcceptance: "Requires separately frozen companion A/A and five paired arms per comparison" } }; + const clock = await openClock(page); + const company = await json(await request.post("/api/companies", { data: { name: `Blocked Stop ${Date.now()}` } })); + report.companyId = company.id; + const original = await json(await request.get("/api/instance/settings/experimental")); + let holder: any, observer: any, held = false, companyRunId: string | undefined; + let stopRequest: Request | undefined, stopSettled = false; + const nonce = randomUUID().replaceAll("-", ""), key = randomUUID().replaceAll("-", ""), code = clock.nextMarkerCode(); + const stopFinished = (req: Request) => { + if (req === stopRequest) { stopSettled = true; report.timing.httpSettledAtMs = Date.now(); } + }; + page.on("requestfinished", stopFinished); + page.on("requestfailed", stopFinished); + try { + const files = ["native-event-ingestion-blocked-stop.journey.ts", "playwright-native-ingestion-blocked-stop.config.ts", + "fixtures/native-ingestion-blocked-stop-authority.ts", "fixtures/native-ingestion-blocked-stop-clock.ts", + "fixtures/native-ingestion-blocked-stop-db.ts", "fixtures/native-ingestion-blocked-stop-protocol.mjs", + "native-event-ingestion.journey.ts", "playwright-native-ingestion.config.ts", "fixtures/in-feed-codex.mjs", + "fixtures/native-ingestion-bin/codex", "fixtures/composer-stop-codex-loader.mjs", "fixtures/native-ingestion-bootstrap.mjs", + "fixtures/native-ingestion-remote-server.mjs", "fixtures/native-ingestion-protocol-qualification.mjs", "fixtures/native-ingestion-h5.mjs"]; + report.sourceHashes = Object.fromEntries(await Promise.all(files.map(async file => [file, + createHash("sha256").update(await readFile(path.join(import.meta.dirname, file))).digest("hex")]))); + const harnessRoot = path.resolve(import.meta.dirname, "../.."); + report.harnessProvenance = { + sourceRevision: execFileSync("git", ["rev-parse", "HEAD"], { cwd: harnessRoot, encoding: "utf8" }).trim(), + sourceStatus: execFileSync("git", ["status", "--porcelain", "--", "tests/e2e"], { cwd: harnessRoot, encoding: "utf8" }).trim(), + driverLockSha256: createHash("sha256").update(await readFile(path.join(harnessRoot, "pnpm-lock.yaml"))).digest("hex"), + inheritedH5cRevision: "e431ec63d41fe90c6fd68bd2fc2bb969feeca917", + }; + report.sourceRevision = execFileSync("git", ["rev-parse", "HEAD"], { cwd: appRoot, encoding: "utf8" }).trim(); + report.buildInfo = JSON.parse(await readFile(path.join(appRoot, "server/dist/build-info.json"), "utf8")); + expect(report.buildInfo.commit).toBe(report.sourceRevision); + report.serverEntrySha256 = createHash("sha256").update(await readFile(path.join(appRoot, "server/dist/index.js"))).digest("hex"); + report.health = await json(await request.get("/api/health")); + await json(await request.patch("/api/instance/settings/experimental", { data: { enableNativeRunner: true, enableWorktreeRunExecution: true } })); + await json(await request.post(`/api/companies/${company.id}/user-secret-definitions`, { data: { key: "OPENAI_API_KEY", name: "Fixture credential" } })); + await json(await request.post(`/api/companies/${company.id}/me/user-secrets`, { data: { definitionKey: "OPENAI_API_KEY", value: "not-a-real-model-key" } })); + const environments = await json(await request.get(`/api/companies/${company.id}/environments?driver=local`)); + expect(environments).toHaveLength(1); + const environment = environments[0]; expect(environment.driver).toBe("local"); + const agent = await json(await request.post(`/api/companies/${company.id}/agents`, { timeout: 120_000, data: { + name: "Blocked Stop local control", role: "engineer", adapterType: "paperclip_runner", defaultEnvironmentId: environment.id, + adapterConfig: { provider: "codex", model: "gpt-5.1-codex-mini", env: { + OPENAI_API_KEY: { type: "user_secret_ref", key: "OPENAI_API_KEY", version: "latest", required: true } } }, + runtimeConfig: { heartbeat: { enabled: false, wakeOnDemand: true, maxConcurrentRuns: 50 } }, + } })); + await expect.poll(async () => (await json(await request.get(`/api/agents/${agent.id}`))).lifecycleState, + { timeout: 60_000 }).toBe("ready"); + report.setup = { environmentId: environment.id, agentId: agent.id, + selection: "Normal local environment and control agent; existing deterministic Codex leaf command; no resolver override" }; + const text = ".".repeat(W.prelude - 1) + ` PRELUDE_${nonce}_READY ` + + Array.from({ length: W.events }, (_, index) => `[${code}${index.toString(36)}] `).join(""); + expect([...text].length).toBeLessThan(4000); + const state = { events: W.events, intervalMs: W.intervalMs, streams: new Map(), errors: [] } as any; + state.streams.set(nonce, { nonce, markerCode: code, opened: false, hello: false, ended: false, aborted: false, + partialBytes: 0, records: new Map(), expectedLauncher: "host-local" }); + clock.clocks.set(key, state); clock.pending.set(key, new Map()); + const prompt = `Measure native streaming for ${nonce.slice(0, 8)}.\n\nINGESTION_PERF nonce=${nonce} events=${W.events} intervalMs=${W.intervalMs} prelude=${W.prelude} markerCode=${code} gate=${clock.base}${key}`; + const before = new Set((await json(await request.get(`/api/companies/${company.id}/issues`))).map((row: any) => row.id)); + await page.goto(`/${company.issuePrefix}/agents/${agent.id}/overview`); + await page.getByRole("link", { name: "Tasks", exact: true }).click(); + await page.getByRole("button", { name: "New Task", exact: true }).last().click(); + const dialog = page.getByRole("dialog"); + await dialog.getByRole("textbox", { name: "editable markdown" }).fill(prompt); + await dialog.getByRole("button", { name: "Select assignee", exact: true }).click(); + await page.getByRole("listbox", { name: "Assignees", exact: true }).getByRole("option") + .filter({ has: page.getByText(agent.name, { exact: true }) }).click(); + await expect(dialog.getByRole("button", { name: "Select assignee" })).toContainText(agent.name); + report.setup.taskCreateActionAtMs = Date.now(); + await dialog.getByRole("button", { name: "Create task", exact: true }).click(); + let issue: any; + await expect.poll(async () => { + issue = (await json(await request.get(`/api/companies/${company.id}/issues`))) + .find((row: any) => !before.has(row.id) && row.description?.includes(nonce)); return Boolean(issue); + }).toBe(true); + report.issueId = issue.id; + await page.goto(`/${company.issuePrefix}/issues/${issue.identifier}`); + await expect.poll(() => clock.pending.get(key)!.has(nonce) && state.streams.get(nonce).hello, + { timeout: W.readinessMs, intervals: [W.pollIntervalMs] }).toBe(true); + await expect(page.getByText(`PRELUDE_${nonce}_READY`, { exact: false }).first()).toBeVisible({ timeout: W.streamMs }); + let run: any; + await expect.poll(async () => { + run = (await json(await request.get(`/api/companies/${company.id}/heartbeat-runs?limit=1000`))) + .find((row: any) => (row.issueId ?? row.nativeIssueId ?? row.contextSnapshot?.issueId ?? row.contextSnapshot?.taskId) === issue.id); + return Boolean(run); + }).toBe(true); + companyRunId = run.id; report.runId = run.id; + const readEvents = async () => { + const result: any[] = []; let afterSeq = 0; + for (;;) { + const rows = await json(await request.get(`/api/heartbeat-runs/${run.id}/events?afterSeq=${afterSeq}&limit=1000`)); + if (!rows.length) return result; + result.push(...rows); + const next = Math.max(...rows.map((row: any) => row.seq)); expect(next).toBeGreaterThan(afterSeq); afterSeq = next; + } + }; + const prelude = (await readEvents()).map(row => row.payload?.prpEvent) + .filter(event => event?.eventType === "item.delta" && String(event.itemId).startsWith("prelude-")); + expect(prelude).toHaveLength(W.prelude); + prelude.forEach((event, index) => { + expect(event.payload.text).toBe(index === W.prelude - 1 ? ` PRELUDE_${nonce}_READY ` : "."); + expect(event.sourceInstanceId).toBeTruthy(); expect(event.sourceEventId).toBeTruthy(); + expect(Number.isSafeInteger(event.sourceSeq) && event.sourceSeq > 0).toBe(true); + }); + expect(new Set(prelude.map(event => `${event.sourceInstanceId}:${event.sourceEventId}:${event.sourceSeq}`)).size).toBe(W.prelude); + report.prelude = { individualDeltas: prelude.length, uniqueIdentities: prelude.length, verifiedBeforeEpochAtMs: Date.now(), + method: "Normal event API read of actual emitted and committed deltas; no outcome prepopulation" }; + const authority = await runtimeAuthority({ appRoot, companyId: company.id, environmentId: environment.id, + issueIds: [issue.id], transport: "local" }); + const row = authority.rows[0], stream = state.streams.get(nonce); + expect(row.run_id).toBe(run.id); + expect(row.process_identity.processPid).toBe(row.process_pid); + expect(row.process_identity.processGroupId).toBe(row.process_group_id); + const processes = { runner: birth(row.process_pid), provider: birth(stream.providerPid), group: row.process_group_id }; + expect(processes.runner.group).toBe(processes.group); expect(processes.provider.group).toBe(processes.group); + const alive = () => !processAbsent(processes.runner.pid) && !processAbsent(-processes.group) && !processAbsent(processes.provider.pid); + expect(alive()).toBe(true); report.authorityBefore = authority; report.processesBefore = processes; + state.epochMs = Date.now() + 1000; report.epochMs = state.epochMs; + clock.pending.get(key)!.get(nonce)!.writeHead(200, { "Content-Type": "application/json" }).end(JSON.stringify({ epochMs: state.epochMs })); + const visible = () => page.evaluate(markerCode => (window as any).nativeIngestionVisible.filter((item: any) => item.markerCode === markerCode), code); + await expect.poll(async () => (await visible()).length, { timeout: W.streamMs }).toBeGreaterThanOrEqual(W.visibleBeforeLock); + report.visibleBeforeLock = await visible(); + holder = await ownedDatabase(appRoot, "blocked_stop_owned_lock"); + observer = await ownedDatabase(appRoot, "blocked_stop_independent_observer"); + await holder.unsafe("BEGIN"); held = true; + const owned = await holder`select id, company_id, pg_backend_pid() as holder_pid from heartbeat_runs + where company_id = ${company.id} and id = ${run.id} for update`; + expect(owned).toHaveLength(1); const holderPid = owned[0].holder_pid; + report.lock = { ...owned[0], acquiredAtMs: Date.now(), scope: "One owned company/run row; no persisted mutation" }; + await expect.poll(async () => { + const rows = await blockers(observer, holderPid); + report.appendBlockedBeforeStop = { observedAtMs: Date.now(), rows }; + return rows.some((candidate: any) => isAppendLockQuery(candidate.query)); + }, { timeout: 10_000, intervals: [W.pollIntervalMs] }).toBe(true); + const button = page.getByRole("button", { name: "Stop", exact: true }); await expect(button).toBeVisible(); + const stop = page.waitForRequest(req => { + const matched = req.method() === "POST" && new URL(req.url()).pathname === `/api/heartbeat-runs/${run.id}/cancel`; + if (matched) report.timing.requestAtMs = Date.now(); + return matched; + }); + report.timing.actionAtMs = Date.now(); + const click = button.click(); + stopRequest = await stop; await click; + const emissionsAtRequest = stream.records.size; + do { + const snapshot = await cancellationSnapshot(observer, company.id, run.id); + const sample = { ...snapshot, httpPending: !stopSettled, auditIds: snapshot.audits.map((audit: any) => audit.id), + processesAlive: alive(), clockOpen: !stream.ended && !stream.aborted, + emittedCount: stream.records.size, blockers: await blockers(observer, holderPid) }; + expect(birth(processes.runner.pid)).toEqual(processes.runner); + expect(birth(processes.provider.pid)).toEqual(processes.provider); + report.heldSamples.push(sample); assertHeldSample(sample, holderPid); + if (Date.now() >= report.timing.requestAtMs + W.heldAfterActualRequestMs) break; + await delay(Math.min(W.pollIntervalMs, report.timing.requestAtMs + W.heldAfterActualRequestMs - Date.now())); + } while (true); + expect(stream.records.size, "Real provider continues emitting while transcript writes and Stop await the run lock").toBeGreaterThan(emissionsAtRequest); + report.timing.releaseRequestedAtMs = Date.now(); + await holder.unsafe("COMMIT"); held = false; report.timing.releaseCommitAckAtMs = Date.now(); + let acknowledged = false, absent = false; + const observationStartedAtMs = Date.now(); + const observe = async () => { + const snapshot = await cancellationSnapshot(observer, company.id, run.id); + if (snapshot.intent) { assertIntentAudit(snapshot); report.intentObserved ??= snapshot; } + if (!acknowledged) { + const value = snapshot.intent?.dispatchState === "acknowledged"; + report.observerSamples.push({ kind: "ack-query", startedAtMs: snapshot.queryStartedAtMs, + completedAtMs: snapshot.observedAtMs, durationMs: snapshot.observedAtMs - snapshot.queryStartedAtMs, value }); + if (value) { + assertIntentAudit(snapshot); acknowledged = true; report.timing.ackObservedAtMs = snapshot.observedAtMs; + report.ack = snapshot; report.timing.ackSuccessfulQueryStartedAtMs = snapshot.queryStartedAtMs; + } else report.timing.lastAckNotObservedAtMs = snapshot.observedAtMs; + } + if (!absent) { + const checkStartedAtMs = Date.now(); + const value = processAbsent(processes.runner.pid) && processAbsent(-processes.group) && processAbsent(processes.provider.pid); + const completedAtMs = Date.now(); + report.observerSamples.push({ kind: "process-check", startedAtMs: checkStartedAtMs, + completedAtMs, durationMs: completedAtMs - checkStartedAtMs, value }); + if (value) { + absent = true; report.timing.absenceObservedAtMs = completedAtMs; + report.timing.absenceSuccessfulCheckStartedAtMs = checkStartedAtMs; + report.processesAfter = { runnerAbsent: true, groupAbsent: true, providerAbsent: true, checkedAtMs: completedAtMs }; + } else report.timing.lastProcessesPresentAtMs = completedAtMs; + } + }; + await observe(); + const settlement = await observeRunSettlement({ expectedRuns: 1, + readRuns: async () => { await observe(); return [await json(await request.get(`/api/heartbeat-runs/${run.id}`))]; }, + readRun: async () => json(await request.get(`/api/heartbeat-runs/${run.id}`)), + poll: (read: () => Promise, timeout: number) => expect.poll(read, { timeout, intervals: [W.pollIntervalMs] }).toBe(true), + onOutcome: (outcome: unknown) => { report.settlement = outcome; } }); + expect(settlement.runs[0].status).toBe("cancelled"); report.settledAtMs = Date.now(); + const sharedDetail = await json(await request.get(`/api/heartbeat-runs/${run.id}`)); + report.sharedSettlement = { runId: sharedDetail.id, companyId: sharedDetail.companyId, status: sharedDetail.status, + cancellation: sharedDetail.resultJson?.cancellation ?? null, + executionCancellation: sharedDetail.resultJson?.executionCancellation ?? null, + nativeCancellation: sharedDetail.resultJson?.nativeCancellation ?? null, observedAtMs: Date.now() }; + await expect.poll(async () => { await observe(); return acknowledged; }, + { timeout: W.acknowledgementAfterSettlementMs, intervals: [W.pollIntervalMs] }).toBe(true); + await expect.poll(async () => { await observe(); return absent; }, + { timeout: W.acknowledgementAfterSettlementMs, intervals: [W.pollIntervalMs] }).toBe(true); + report.timing.postReleaseObservationStartedAtMs = observationStartedAtMs; + await expect.poll(() => stopSettled, { timeout: W.acknowledgementAfterSettlementMs, + intervals: [W.pollIntervalMs] }).toBe(true); + const response = await stopRequest.response(); expect(response?.ok()).toBe(true); report.stopHttpStatus = response!.status(); + report.authorityAfter = await runtimeAuthority({ appRoot, companyId: company.id, environmentId: environment.id, + issueIds: [issue.id], transport: "local", afterStop: true }); + const stopped = report.authorityAfter.rows[0]; + expect(stopped.status).toBe("cancelled"); expect(stopped.cancellation_dispatch_state).toBe("acknowledged"); + expect(BigInt(stopped.process_stopped_seq)).toBeGreaterThan(BigInt(row.process_identity_seq)); + expect(stopped.process_stopped_source_event_id).toBeNull(); + expect(stopped.process_stopped.processPid).toBe(processes.runner.pid); + expect(stopped.process_stopped.processGroupId).toBe(processes.group); + let lease: any; + await expect.poll(async () => { + lease = (await json(await request.get(`/api/environments/${environment.id}/leases`))) + .find((candidate: any) => candidate.heartbeatRunId === run.id); + return Boolean(lease?.releasedAt && ["released", "expired", "failed"].includes(lease.status) && lease.cleanupStatus !== "failed"); + }, { timeout: W.leaseAfterSettlementMs }).toBe(true); + expect(lease.provider).toBe("local"); expect(lease.providerLeaseId).toBeNull(); + report.lease = lease; + const seen = await visible(), events = await readEvents(); + const deltas = events.map(event => event.payload?.prpEvent).filter(event => event?.eventType === "item.delta" + && String(event.payload?.text).startsWith(`[${code}`)); + expect(deltas.length).toBeGreaterThanOrEqual(W.visibleBeforeLock); + deltas.forEach((event, index) => { + expect(event.payload.text).toBe(`[${code}${index.toString(36)}] `); expect(event.runId).toBe(run.id); + expect(event.sourceInstanceId).toBeTruthy(); expect(event.sourceEventId).toBeTruthy(); + expect(Number.isSafeInteger(event.sourceSeq) && event.sourceSeq > 0).toBe(true); + expect(stream.records.has(index)).toBe(true); + }); + expect(new Set(deltas.map(event => `${event.sourceInstanceId}:${event.sourceEventId}:${event.sourceSeq}`)).size).toBe(deltas.length); + seen.forEach((marker: any) => expect(marker.index).toBeLessThan(deltas.length)); + expect(events.filter(event => event.eventType === "run.result.accepted" || event.payload?.prpEvent?.eventType === "run.result.accepted")).toHaveLength(0); + await page.screenshot({ path: info.outputPath("blocked-stop-existing-history.png"), fullPage: true }); + await page.reload(); + await expect(page.getByRole("button", { name: "Stop", exact: true })).toHaveCount(0); + await expect(page.getByText(`[${code}${seen.at(-1).index.toString(36)}]`, { exact: false }).first()).toBeVisible(); + const transcriptRows = (rows: any[]) => rows.filter(row => row.payload?.prpEvent?.eventType === "item.delta") + .map(row => ({ id: row.id, seq: row.seq, event: row.payload.prpEvent })); + expect(transcriptRows(await readEvents())).toEqual(transcriptRows(events)); + await expect.poll(() => stream.ended || stream.aborted, { timeout: 30_000 }).toBe(true); + expect(state.errors).toEqual([]); + report.transcript = { individualDeltas: deltas.length, uniqueSourceIdentities: deltas.length, ordered: true, + persistedAfterReload: true, resultAcceptedEvents: 0, seen, deltas, preludeCount: prelude.length }; + report.clock = { ...stream, records: [...stream.records.values()] }; + report.complete = true; + } catch (error) { + report.failure = { message: String(error), stack: error instanceof Error ? error.stack : undefined, atMs: Date.now() }; + throw error; + } finally { + const cleanupErrors: unknown[] = []; + const cleanup = async (step: string, action: () => Promise) => { + try { await action(); } catch (error) { cleanupErrors.push({ step, error: String(error) }); } + }; + // Always release our exact owned lock before invoking any public cancellation cleanup. + if (held) await cleanup("rollback-owned-lock", async () => { await holder.unsafe("ROLLBACK"); held = false; }); + if (holder) await cleanup("close-holder", () => holder.end({ timeout: 5 })); + if (observer) await cleanup("close-observer", () => observer.end({ timeout: 5 })); + for (const pending of clock.pending.values()) for (const response of pending.values()) if (!response.writableEnded) response.writeHead(503).end(); + clock.gate.closeAllConnections(); await new Promise(resolve => clock.gate.close(() => resolve())); + if (companyRunId) await cleanup("settle-owned-run", async () => { + const run = await json(await request.get(`/api/heartbeat-runs/${companyRunId}`)); + if (["running", "queued"].includes(run.status)) await json(await request.post(`/api/heartbeat-runs/${companyRunId}/cancel`)); + }); + await cleanup("restore-experimental-settings", async () => json(await request.patch("/api/instance/settings/experimental", { + data: { enableNativeRunner: original.enableNativeRunner, enableWorktreeRunExecution: original.enableWorktreeRunExecution } }))); + report.cleanup = { complete: cleanupErrors.length === 0, errors: cleanupErrors }; + report.complete &&= report.cleanup.complete; report.finishedAtMs = Date.now(); + let validationError: unknown; + if (report.complete) { + try { report.metrics = metrics(report); assertFunctional(report); } + catch (error) { validationError = error; report.complete = false; report.validationFailure = String(error); } + } + await info.attach("native-ingestion-blocked-stop-evidence", { body: JSON.stringify(report, null, 2), contentType: "application/json" }); + if (validationError) throw validationError; + if (!report.failure) expect(cleanupErrors).toEqual([]); + page.off("requestfinished", stopFinished); page.off("requestfailed", stopFinished); + } +}); diff --git a/tests/e2e/playwright-native-ingestion-blocked-stop.config.ts b/tests/e2e/playwright-native-ingestion-blocked-stop.config.ts new file mode 100644 index 00000000000..3446dbe0af1 --- /dev/null +++ b/tests/e2e/playwright-native-ingestion-blocked-stop.config.ts @@ -0,0 +1,5 @@ +import { defineConfig } from "@playwright/test"; +import base from "./playwright-native-ingestion.config"; +process.env.PAPERCLIP_INGESTION_COMPANION_PROTOCOL = "blocked-stop-history-v1"; +export default defineConfig({ ...base, testMatch: "native-event-ingestion-blocked-stop.journey.ts", timeout: 900_000, + retries: 0, workers: 1, use: { ...base.use, trace: "off", video: "off" } });