mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
Keep original emission clocks outside displayed ingestion text
This commit is contained in:
Dotta committed
1 parent
cbc9bbce20
commit
c80a467737
7 files changed
+670
-68
No files matched your search
@@ -267,3 +267,166 @@ the combined change still requires the loaded improvement. Comparison rejects
|
||||
smoke, source-server evidence, mismatched artifacts/protocols, missing background
|
||||
delta proof and incomplete runs. Live-provider walkthroughs remain a separate
|
||||
integration gate.
|
||||
|
||||
### Browser protocol v2, registered before fresh A/A
|
||||
|
||||
The first full v1 baseline qualification failed in the paced 1×128 case. Its
|
||||
paint observer accumulated 127 markers and missed index 127. Independent,
|
||||
read-only PostgreSQL inspection found all 128 deltas with unique source
|
||||
identities and no gaps. The run succeeded. The normalized completed item and
|
||||
final comment were each exactly 4,000 characters with a truncation marker and
|
||||
only complete marker indices 0–55. Raw Playwright DOM snapshots confirmed the
|
||||
same 4,000-character terminal text, after streamed DOM text reached 9,033
|
||||
characters. This is not an accessibility-snapshot truncation assumption.
|
||||
|
||||
The existing Rust provider normalizer bounds completed text at 4,000 Unicode
|
||||
characters. UI projection treats a completed item as authoritative and replaces
|
||||
its streamed deltas. Long in-band diagnostic clocks made the fixture's completed
|
||||
answer exceed that preexisting bound. No application or normalizer policy is
|
||||
changed to make the experiment pass. V1 failed artifacts, trace, decoded paint
|
||||
observations and SELECT-only database receipts remain evidence of an unqualified
|
||||
experiment; none establishes a speedup.
|
||||
|
||||
V2 changes the browser measurement protocol for every arm before any fresh A/A:
|
||||
|
||||
- A visible chunk is `[rrII] `: a unique two-base36-character run code followed by
|
||||
its base36 index. A full 512-chunk answer is 3,548 ASCII characters and bytes.
|
||||
The existing-history fixture keeps its 1,000 actual single-character prelude
|
||||
deltas and sentinel; that prelude plus 128 compact chunks totals 1,907 characters.
|
||||
Explicit checks keep these completed payloads below the unchanged 4,000 bound.
|
||||
Before releasing measured emission, an independent normal event-API read verifies
|
||||
all 1,000 prelude deltas and source identities; the report records that pre-gate proof.
|
||||
- Full nonce, index, original intended emission time and original provider-process
|
||||
emission time travel on one streamed HTTP clock channel per provider. The
|
||||
channel opens and sends its hello before the common gate releases. There is no
|
||||
per-token HTTP request or wait for clock drain/response before writing text.
|
||||
- The browser still captures a DOM snapshot in one animation frame and assigns
|
||||
that snapshot's paint-opportunity timestamp in the next. It records that time
|
||||
immediately, even when the clock channel has not arrived. It joins original
|
||||
clocks afterward, preserving the original browser timestamp.
|
||||
- Every visible marker and every independently read durable delta must match its
|
||||
original full nonce/code/index clock. Missing, duplicated or conflicting clocks
|
||||
fail. Stop may terminate the clock stream with a partial unacknowledged raw
|
||||
tail, but no observed or database-acknowledged prefix receives a synthetic clock.
|
||||
- The provider completes its assistant item before invoking semantic progress
|
||||
and Finish tools. Tool/result/terminal output does not wait for the clock upload
|
||||
response. Standalone qualification deliberately withholds that response through
|
||||
terminal settlement and Stop acknowledgement.
|
||||
- All 16/128/512 event counts, 1/10/100 run counts, emission schedules, history
|
||||
counts, payload-identity checks, drainage checks, timeouts and performance gates
|
||||
remain fixed. Removing diagnostic clocks makes the browser text payload smaller;
|
||||
this is explicit and matched across all v2 arms. The owner-path workload retains
|
||||
its original 128-byte text payload and unchanged source/gate hashes.
|
||||
- Timed A/A and comparison runs disable Playwright trace, video and screencast.
|
||||
Their metadata records the actual configured capture modes; the comparator
|
||||
rejects recording, legacy v1, smoke, profile and incomplete evidence. The two
|
||||
animation-frame observer remains declared measurement overhead. Manual
|
||||
screenshots occur after timed terminal drainage. Separate diagnostics may
|
||||
capture browser traces and app-only Node CPU profiles and stay ineligible.
|
||||
- A failure attachment retains the active case, requested and ready participants,
|
||||
observed marker timestamps, raw rendered text, DOM marker inventory, received
|
||||
source clocks, stream errors and the latest independent durable read.
|
||||
|
||||
The separate browser harness H is based on application ref
|
||||
`7922802f069ab3b39e9f79368c46615ba2201ba9` with only the v2 fixture, observer,
|
||||
recording metadata, comparator, tests and this record changed. H uses explicitly
|
||||
identified driver-only tooling dependencies. Each measured application retains
|
||||
its own independent locked dependencies, clean source checkout and matching
|
||||
compiled build stamp: B0 `83b5f7d486e70e484acd9a41bc957054f4e10175`, B1
|
||||
`7922802f069ab3b39e9f79368c46615ba2201ba9`, B2
|
||||
`6afeba4edafd47120e8117f736641d5424b27d4a`. Reports retain H's own revision,
|
||||
source status, individual protocol hashes and lockfile hash. No application
|
||||
build stamp is rewritten for a test-only change. Exact H commit and combined
|
||||
protocol hash are frozen with the qualification receipts before new A/A starts.
|
||||
|
||||
Qualification must pass the source-clock protocol smoke and the full B0 browser
|
||||
matrix, including actual 100-participant readiness and independent background
|
||||
persistence. Fresh A/A still precedes any candidate comparison. The owner-path
|
||||
20% maximum paired variance / 10% median drift limits and the requirement that
|
||||
four individual candidate pairs exceed observed noise are unchanged. Failed v1
|
||||
samples are never reused as v2 evidence.
|
||||
|
||||
Registered v2 browser protocol SHA256: `5498ed8dfc7139285aeda582c540eb572f07a1ad3e73de6e0f12c5af4c45e0ba`.
|
||||
Comparator SHA256: `d8cf6f55339ad811ca1743f4bc73b74afa3a321b93e17ef913fc0339b47b4d7d`. Individual source hashes and
|
||||
H's exact committed revision are retained with each qualification/execution receipt.
|
||||
|
||||
### V2 admission qualification correction, before fresh browser A/A
|
||||
|
||||
H `96b1a5f1a4497d1389c21b0cc52d55b86641690d` passed the compiled
|
||||
single-run smoke and completed paced 128, burst 16, and ten runs of 128 events.
|
||||
Its first full baseline qualification then failed during setup for the 100-run
|
||||
case, before the common emission epoch was released. All 100 issues existed,
|
||||
but only one provider/clock hello arrived and no measured deltas were emitted.
|
||||
Ten concurrent normal task-activation requests timed out; cleanup also timed
|
||||
out and masked the original exception. The complete failed report and an
|
||||
independent SELECT-only database inspection remain excluded from acceptance.
|
||||
The active issues were 50 todo, 49 backlog, and 1 in_progress; 31 runs existed
|
||||
(3 running, 28 queued), with no measured item.delta rows.
|
||||
|
||||
Static inspection found a separate potential pool-starvation path: each native
|
||||
wakeup transaction can await a responsible-user lookup on the outer database
|
||||
client, whose default pool is 10; a group of ten activation wakeups can consume
|
||||
that pool. This is an observed pre-epoch startup-contention failure with a
|
||||
matching static candidate, not a runtime-proven deadlock diagnosis: no live
|
||||
pg_stat_activity capture exists. The product admission hazard remains separate
|
||||
from this ingestion change; increasing pool size or request timeouts is not
|
||||
part of the experiment.
|
||||
|
||||
Before any fresh browser A/A, H now schedules each background using the same normal
|
||||
PATCH, then requires that exact participant's provider GET and original-clock
|
||||
hello before scheduling the next. All 100 providers remain live behind the
|
||||
closed common emission gate. A **shared 180-second readiness budget starts
|
||||
before activation**; this is stricter than starting the same readiness budget
|
||||
after activation. The 600-second overall test limit, 20-second request limits,
|
||||
event counts, payloads, rates, 100-run measured concurrency and every performance
|
||||
gate are unchanged. Source owner-path protocol and gates are byte-identical.
|
||||
The report identifies this admission protocol explicitly, and its comparator
|
||||
rejects reports without it. Original exceptions, exact activation progress and
|
||||
separate cleanup errors are retained without accepting a cleanup failure.
|
||||
|
||||
Registered corrected browser protocol SHA256:
|
||||
`5430314c028c9e05ddf4e27e6e22a0191ee0d4a61ffc61649627fd93777acebc`.
|
||||
Comparator SHA256:
|
||||
`c1b7249022a95ff2186e72784247ca83125550c02910d8ad2bf52e1389cf9afc`.
|
||||
The corrected H commit and source hashes are recorded before its full baseline
|
||||
qualification and new browser A/A. Earlier H96b samples never enter that series.
|
||||
|
||||
The original sparse-stream acceptance also requires no regression in gaps between
|
||||
rendered chunks. After terminal drainage, H derives consecutive gaps between
|
||||
distinct observed paint timestamps from the existing retained `visibleMs` values.
|
||||
Reports include paint count, gap count, median gap and maximum observed gap;
|
||||
no tail quantile is inferred from a small gap sample. The comparator requires
|
||||
observed paced-stream gaps and gates the maximum observed gap against the same
|
||||
five-pair A/A noise envelope. This adds the previously missing acceptance check
|
||||
before browser A/A; it changes neither the streaming observer nor emitted work.
|
||||
The independent source-owner A/A may run during local H work because its workload,
|
||||
measurement code and gates remain unchanged.
|
||||
|
||||
### Setup readiness observation correction, before browser A/A
|
||||
|
||||
The full H `e27b338315034c062b18606c0f3ff7d883e3dfed` qualification
|
||||
reached activation index 95, with 95 provider GETs and clock hellos, then hit the
|
||||
unchanged shared 180-second deadline. No common epoch was released and no
|
||||
measured emission clocks existed. The retained report records the primary
|
||||
984-millisecond remaining-budget timeout and a separate cleanup 409 response;
|
||||
it is ineligible for acceptance. After teardown, an independent process check
|
||||
verified no owned Node, PostgreSQL or Rust Runner processes remained before the
|
||||
separate source-owner cursor comparison began.
|
||||
|
||||
The default readiness poll backoff is a suspected contributor to the approximately
|
||||
1.85-second setup cadence. That failed run did not record clock-hello receipt or
|
||||
readiness-observation times, so it does not prove poll quantization. Before fresh
|
||||
browser A/A, H changes only the two setup predicates to a fixed 50-millisecond
|
||||
observation interval. Both predicates read local provider/clock maps; they issue
|
||||
no database or application request. The shared 180-second hard deadline, 600-second
|
||||
overall limit, exact-participant checks, 100 live providers, common epoch, measured
|
||||
event schedules, counts, payloads and performance gates remain unchanged.
|
||||
|
||||
Future qualification records pre-epoch provider GET receipt, clock-hello receipt
|
||||
and background readiness-observation timestamps to expose any observation lag.
|
||||
These are setup diagnostics, distinct from the unchanged original provider
|
||||
emission clocks and browser paint observer. The comparator requires the frozen
|
||||
50-millisecond setup interval. Earlier failed samples remain excluded.
|
||||
|
||||
Registered browser protocol after setup observation correction: `902c6a208793f0bcde1b326a1fee46d85df3274117754e3bfbbf77a65dc730ee`.
|
||||
Comparator SHA256: `e50d958532906d037413db6f66bdc02e60c93ddcacfe2e18e5b731d302f6a785`.
|
||||
@@ -22,9 +22,17 @@ const median = (values: number[]) => [...values].sort((a, b) => a - b)[Math.floo
|
||||
const first = pairs[0];
|
||||
for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const) {
|
||||
const report = pair[arm];
|
||||
assert.equal(report.protocolVersion, "v2-compact-markers-original-clock-stream", "Fresh browser acceptance requires the qualified v2 clock protocol");
|
||||
assert.equal(report.admissionProtocol, "sequential-provider-and-clock-readiness-before-common-epoch", "Use the qualified pre-measurement admission protocol");
|
||||
assert.equal(report.setupReadinessPollingMs, 50, "Keep the qualified setup readiness observation interval");
|
||||
assert.equal(report.complete, true); assert.equal(report.qualificationOnly, false, "Smoke cannot qualify performance");
|
||||
assert.equal(report.cleanup?.complete, true, "Acceptance requires completed fixture cleanup");
|
||||
assert.deepEqual(report.cleanup?.errors, [], "Acceptance rejects fixture cleanup errors");
|
||||
assert.notEqual(report.diagnosticOnly, true, "Diagnostic profiles cannot qualify unprofiled performance");
|
||||
assert.notEqual(report.performanceAcceptanceEligible, false, "Profile-only evidence cannot qualify performance");
|
||||
assert.equal(report.browserRecording?.trace, "off", "Acceptance requires browser trace capture off");
|
||||
assert.equal(report.browserRecording?.video, "off", "Acceptance requires browser video capture off");
|
||||
assert.equal(report.browserRecording?.screencast, false, "Acceptance requires browser screencast capture off");
|
||||
assert.equal(report.serverArtifact, "compiled", "Performance acceptance requires the compiled server and shipped UI");
|
||||
assert.ok(report.serverEntrySha256 && report.measurementSha256);
|
||||
assert.equal(report.compiledBuildRevision, report.sourceRevision, "Compiled build stamp does not match the measured revision");
|
||||
@@ -38,9 +46,21 @@ for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const)
|
||||
const result = report.cases.find((item: any) => item.name === name);
|
||||
assert.ok(result, `Missing fixed workload ${name}`);
|
||||
assert.equal(result.persistedAfterReload, true);
|
||||
assert.equal(result.matchedVisibleClocks, result.visible?.length, "Every observed marker needs its original provider clock");
|
||||
assert.ok(result.matchedVisibleClocks >= (result.stop ? 3 : result.events), "Missing foreground render/clock observations");
|
||||
if (name === "paced") {
|
||||
assert.ok(result.interChunkGaps?.count >= 1, "Paced streaming needs observed inter-chunk paint gaps");
|
||||
assert.equal(result.interChunkGaps.paintCount, result.interChunkGaps.count + 1);
|
||||
}
|
||||
if (result.prelude) {
|
||||
assert.equal(result.preludeProof?.individualDeltas, result.prelude, "Existing history requires independently stored prelude deltas");
|
||||
assert.equal(result.preludeProof?.uniqueSourceIdentities, result.prelude);
|
||||
assert.ok(result.preludeProof?.verifiedAtMs < result.epochMs, "Verify retained prelude before releasing measured emission");
|
||||
}
|
||||
assert.equal(result.durable.length, result.runs, "Every background run needs independent delta persistence proof");
|
||||
for (const proof of result.durable) {
|
||||
assert.equal(proof.ordered, true); assert.equal(proof.uniqueSourceIdentities, proof.individualDeltas);
|
||||
assert.equal(proof.matchedSourceClocks, proof.individualDeltas, "Every durable measured delta needs an original provider clock");
|
||||
if (!result.stop) assert.equal(proof.individualDeltas, result.events);
|
||||
}
|
||||
const reference = first.baseline.cases.find((item: any) => item.name === name);
|
||||
@@ -51,6 +71,7 @@ const metric = (item: any, key: string): number => key.split(".").reduce((value,
|
||||
const measurements = names.flatMap(name => {
|
||||
const observed = first.baseline.cases.find((item: any) => item.name === name);
|
||||
const keys = ["firstVisibleAfterEmissionMs", "emissionToVisible.medianMs", "drainThroughTerminalMs",
|
||||
...(name === "paced" ? ["interChunkGaps.maxMs"] : []),
|
||||
...(observed.emissionToVisible.p95Ms === undefined ? [] : ["emissionToVisible.p95Ms", "intendedToVisible.p95Ms"]),
|
||||
...(observed.stop ? ["stopThroughTerminalMs", "stopThroughAcknowledgementMs"] : [])];
|
||||
return keys.map(key => {
|
||||
|
||||
@@ -75,15 +75,18 @@ const workloads = [
|
||||
{ name: "operator-stop", runs: 1, events: 512, intervalMs: 100, prelude: 0, stop: true },
|
||||
];
|
||||
function browserReport(revision, latency) {
|
||||
return { complete: true, qualificationOnly: false, diagnosticOnly: false, performanceAcceptanceEligible: true,
|
||||
return { protocolVersion: "v2-compact-markers-original-clock-stream", admissionProtocol: "sequential-provider-and-clock-readiness-before-common-epoch", setupReadinessPollingMs: 50, complete: true, cleanup: { complete: true, errors: [] }, qualificationOnly: false, diagnosticOnly: false, performanceAcceptanceEligible: true, browserRecording: { trace: "off", video: "off", screencast: false },
|
||||
sourceRevision: revision, compiledBuildRevision: revision, serverArtifact: "compiled", serverEntrySha256: `${revision}-compiled`,
|
||||
measurementSha256: "synthetic-fixed-fixture", compiledHotPathHashes: { append: "append", port: "port", runtime: "runtime" },
|
||||
cases: workloads.map(workload => ({ ...workload, stop: Boolean(workload.stop), persistedAfterReload: true,
|
||||
epochMs: 1000, preludeProof: workload.prelude ? { individualDeltas: workload.prelude, uniqueSourceIdentities: workload.prelude, verifiedAtMs: 999 } : undefined,
|
||||
visible: Array(workload.stop ? 3 : workload.events).fill({}), matchedVisibleClocks: workload.stop ? 3 : workload.events,
|
||||
firstVisibleAfterEmissionMs: latency, emissionToVisible: { medianMs: latency, p95Ms: latency },
|
||||
interChunkGaps: { paintCount: workload.events, count: workload.events - 1, medianMs: latency, maxMs: latency },
|
||||
intendedToVisible: { p95Ms: latency }, drainThroughTerminalMs: latency,
|
||||
stopThroughTerminalMs: latency, stopThroughAcknowledgementMs: latency,
|
||||
durable: Array.from({ length: workload.runs }, (_, index) => ({ token: String(index), ordered: true,
|
||||
uniqueSourceIdentities: workload.events, individualDeltas: workload.events })),
|
||||
uniqueSourceIdentities: workload.events, individualDeltas: workload.events, matchedSourceClocks: workload.events })),
|
||||
})) };
|
||||
}
|
||||
|
||||
@@ -119,6 +122,18 @@ test("browser CLI rejects the three-pairs-above-noise counterexample and ineligi
|
||||
["source", report => { report.serverArtifact = "source-qualification"; }, /compiled server/],
|
||||
["profile", report => { report.diagnosticOnly = true; }, /Diagnostic profiles/],
|
||||
["profile-ineligible", report => { report.performanceAcceptanceEligible = false; }, /Profile-only evidence/],
|
||||
["missing-retained-prelude", report => { report.cases.find(row => row.prelude).preludeProof.individualDeltas--; }, /stored prelude deltas/],
|
||||
["legacy-protocol", report => { delete report.protocolVersion; }, /v2 clock protocol/],
|
||||
["unqualified-admission", report => { delete report.admissionProtocol; }, /admission protocol/],
|
||||
["unqualified-readiness-observation", report => { delete report.setupReadinessPollingMs; }, /readiness observation interval/],
|
||||
["unfinished-cleanup", report => { report.cleanup.complete = false; }, /completed fixture cleanup/],
|
||||
["cleanup-error", report => { report.cleanup.errors.push({ step: "restore", message: "timeout" }); }, /cleanup errors/],
|
||||
["missing-paced-gaps", report => { delete report.cases[0].interChunkGaps; }, /inter-chunk paint gaps/],
|
||||
["unmatched-visible-clock", report => { report.cases[0].matchedVisibleClocks--; }, /original provider clock/],
|
||||
["unmatched-durable-clock", report => { report.cases[0].durable[0].matchedSourceClocks--; }, /original provider clock/],
|
||||
["browser-trace", report => { report.browserRecording.trace = "on"; }, /trace capture off/],
|
||||
["browser-video", report => { report.browserRecording.video = "on"; }, /video capture off/],
|
||||
["browser-screencast", report => { report.browserRecording.screencast = true; }, /screencast capture off/],
|
||||
]) {
|
||||
const rejected = run([`--pairs=${await manifest(name, Array(5).fill(0.30), false, change)}`, `--noise-report=${noise}`, "--require-acceptance"]);
|
||||
assert.notEqual(rejected.status, 0); assert.match(rejected.stderr, message);
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { once } from 'node:events';
|
||||
import { readFileSync } from 'node:fs';
|
||||
import { request as httpRequest } from 'node:http';
|
||||
import { join, resolve } from 'node:path';
|
||||
import { createInterface } from 'node:readline';
|
||||
if (process.argv.includes('--version')) { console.log('codex-cli 0.115.0 (in-feed fixture)'); process.exit(0); }
|
||||
@@ -14,6 +15,7 @@ const recoveryFixture = process.env.PAPERCLIP_RECOVERY_FIXTURE === "1";
|
||||
const ingestionFixture = process.argv.includes('--ingestion-stream-fixture');
|
||||
let ingestionPlan = null;
|
||||
let ingestionGeneration = 0;
|
||||
let ingestionClock = null;
|
||||
let currentObjective = "";
|
||||
let emitCeoLineage = false;
|
||||
let completionContract = { revision: "1", criterionIds: ["objective"] };
|
||||
@@ -42,15 +44,56 @@ function ingestionFixturePlan(value) {
|
||||
// The real Markdown composer escapes punctuation, and the native brief
|
||||
// also carries a shortened title. Accept a complete fixture plan only.
|
||||
for (let pass = 0; pass < 4; pass++) text = text.replace(/\\([\\_=:./-])/g, '$1');
|
||||
const match = text.match(/INGESTION_PERF nonce=([a-z0-9]+) events=(\d+) intervalMs=(\d+) prelude=(\d+) gate=(http:\/\/(?:127\.0\.0\.1|localhost):\d+\/native-ingestion-gate\/[a-f0-9]{32})/);
|
||||
const match = text.match(/INGESTION_PERF nonce=([a-z0-9]+) events=(\d+) intervalMs=(\d+) prelude=(\d+) markerCode=([a-z0-9]{2}) gate=(http:\/\/(?:127\.0\.0\.1|localhost):\d+\/native-ingestion-gate\/[a-f0-9]{32})/);
|
||||
if (!match) continue;
|
||||
const plan = { nonce: match[1], events: Number(match[2]), intervalMs: Number(match[3]), prelude: Number(match[4]), gate: match[5] };
|
||||
const plan = { nonce: match[1], events: Number(match[2]), intervalMs: Number(match[3]), prelude: Number(match[4]), markerCode: match[5], gate: match[6] };
|
||||
if (plan.events < 1 || plan.events > 4096 || plan.intervalMs > 1000 || plan.prelude > 10_000) throw new Error('Ingestion fixture exceeds its bounded workload');
|
||||
return plan;
|
||||
}
|
||||
throw new Error('Missing complete INGESTION_PERF fixture plan');
|
||||
}
|
||||
|
||||
function openIngestionClock(plan, gate) {
|
||||
const endpoint = new URL(gate);
|
||||
endpoint.pathname += '/clock';
|
||||
endpoint.searchParams.set('participant', plan.nonce);
|
||||
let closed = false, failure = null, settle;
|
||||
const done = new Promise(resolve => { settle = resolve; });
|
||||
const fail = (error) => {
|
||||
if (failure) return;
|
||||
failure = new Error(`Ingestion fixture clock ${plan.nonce}: ${error.message}`);
|
||||
process.stderr.write(`${failure.message}\n`);
|
||||
process.exitCode = 1;
|
||||
settle(false);
|
||||
};
|
||||
const request = httpRequest(endpoint, { method: 'POST', agent: false,
|
||||
headers: { 'Content-Type': 'application/x-ndjson' } }, response => {
|
||||
response.resume();
|
||||
response.on('error', fail);
|
||||
response.on('aborted', () => fail(new Error('Response aborted')));
|
||||
response.on('end', () => {
|
||||
if (response.statusCode < 200 || response.statusCode >= 300) fail(new Error(`HTTP ${response.statusCode}`));
|
||||
else settle(true);
|
||||
});
|
||||
});
|
||||
request.on('error', fail);
|
||||
request.setTimeout(300_000, () => request.destroy(new Error('Upload timed out')));
|
||||
const write = record => {
|
||||
if (failure) throw failure;
|
||||
if (closed) throw new Error('Ingestion fixture clock is closed');
|
||||
// One buffered stream, with no per-token request or clock-drain wait.
|
||||
request.write(`${JSON.stringify(record)}\n`);
|
||||
};
|
||||
const close = () => {
|
||||
if (closed) return;
|
||||
closed = true;
|
||||
try { request.end(); } catch (error) { fail(error); }
|
||||
};
|
||||
write({ type: 'hello', nonce: plan.nonce, markerCode: plan.markerCode,
|
||||
events: plan.events, intervalMs: plan.intervalMs });
|
||||
return { write, close, done };
|
||||
}
|
||||
|
||||
async function streamIngestion() {
|
||||
if (!ingestionPlan) throw new Error('Missing INGESTION_PERF fixture plan');
|
||||
const plan = ingestionPlan, generation = ++ingestionGeneration;
|
||||
@@ -71,32 +114,47 @@ async function streamIngestion() {
|
||||
item: { id: preludeId, type: 'agentMessage', phase: 'commentary',
|
||||
text: `${'.'.repeat(Math.max(0, plan.prelude - 1))} PRELUDE_${plan.nonce}_READY ` } } });
|
||||
}
|
||||
gate.searchParams.set('participant', plan.nonce);
|
||||
const response = await fetch(gate, { signal: AbortSignal.timeout(300_000) });
|
||||
if (!response.ok) throw new Error(`Ingestion gate returned ${response.status}`);
|
||||
const { epochMs } = await response.json();
|
||||
if (!Number.isSafeInteger(epochMs)) throw new Error('Invalid ingestion schedule');
|
||||
const itemId = `ingestion-${turnId}`;
|
||||
await sendIngestion({ method: 'item/started', params: { threadId, turnId,
|
||||
item: { id: itemId, type: 'agentMessage', phase: 'final_answer', text: '' } } });
|
||||
let text = '';
|
||||
for (let index = 0; index < plan.events; index++) {
|
||||
const intended = epochMs + index * plan.intervalMs;
|
||||
const wait = intended - Date.now();
|
||||
if (wait > 0) await new Promise(resolve => setTimeout(resolve, wait));
|
||||
const clock = openIngestionClock(plan, gate);
|
||||
ingestionClock = clock;
|
||||
try {
|
||||
gate.searchParams.set('participant', plan.nonce);
|
||||
const response = await fetch(gate, { signal: AbortSignal.timeout(300_000) });
|
||||
if (!response.ok) throw new Error(`Ingestion gate returned ${response.status}`);
|
||||
const { epochMs } = await response.json();
|
||||
if (!Number.isSafeInteger(epochMs)) throw new Error('Invalid ingestion schedule');
|
||||
if (generation !== ingestionGeneration) return;
|
||||
const delta = `INGEST_${plan.nonce}_${index}_${intended}_${Date.now()} `;
|
||||
text += delta;
|
||||
await sendIngestion({ method: 'item/agentMessage/delta', params: { threadId, turnId, itemId, delta } });
|
||||
const itemId = `ingestion-${turnId}`;
|
||||
await sendIngestion({ method: 'item/started', params: { threadId, turnId,
|
||||
item: { id: itemId, type: 'agentMessage', phase: 'final_answer', text: '' } } });
|
||||
let text = '';
|
||||
for (let index = 0; index < plan.events; index++) {
|
||||
const intendedMs = epochMs + index * plan.intervalMs;
|
||||
const wait = intendedMs - Date.now();
|
||||
if (wait > 0) await new Promise(resolve => setTimeout(resolve, wait));
|
||||
if (generation !== ingestionGeneration) return;
|
||||
const emittedMs = Date.now();
|
||||
clock.write({ type: 'emission', nonce: plan.nonce, markerCode: plan.markerCode, index, intendedMs, emittedMs });
|
||||
const delta = `[${plan.markerCode}${index.toString(36)}] `;
|
||||
text += delta;
|
||||
await sendIngestion({ method: 'item/agentMessage/delta', params: { threadId, turnId, itemId, delta } });
|
||||
}
|
||||
clock.close();
|
||||
if (generation !== ingestionGeneration) return;
|
||||
await sendIngestion({ method: 'item/completed', params: { threadId, turnId,
|
||||
item: { id: itemId, type: 'agentMessage', phase: 'final_answer', text } } });
|
||||
await call('report_progress', { idempotencyKey: `ingestion-${plan.nonce}`, body: `Ingestion fixture emitted ${plan.events} measured chunks.` });
|
||||
if (generation !== ingestionGeneration) return;
|
||||
await finish(`Ingestion fixture ${plan.nonce} completed.`, `fixture:ingestion-${plan.nonce}`);
|
||||
if (generation !== ingestionGeneration) return;
|
||||
await sendIngestion({ method: 'turn/completed', params: { threadId, turn: { id: turnId, status: 'completed' } } });
|
||||
// Upload completion cannot delay output, tools, result or protocol terminal.
|
||||
// Errors are already explicit on stderr and set a nonzero process exit;
|
||||
// do not produce a second terminal notification after a completed turn.
|
||||
await clock.done;
|
||||
} finally {
|
||||
clock.close();
|
||||
if (ingestionClock === clock) ingestionClock = null;
|
||||
}
|
||||
if (generation !== ingestionGeneration) return;
|
||||
await call('report_progress', { idempotencyKey: `ingestion-${plan.nonce}`, body: `Ingestion fixture emitted ${plan.events} measured chunks.` });
|
||||
if (generation !== ingestionGeneration) return;
|
||||
await finish(`Ingestion fixture ${plan.nonce} completed.`, `fixture:ingestion-${plan.nonce}`);
|
||||
if (generation !== ingestionGeneration) return;
|
||||
await sendIngestion({ method: 'item/completed', params: { threadId, turnId,
|
||||
item: { id: itemId, type: 'agentMessage', phase: 'final_answer', text } } });
|
||||
await sendIngestion({ method: 'turn/completed', params: { threadId, turn: { id: turnId, status: 'completed' } } });
|
||||
}
|
||||
function unwrap(result) {
|
||||
const texts = result?.contentItems ?? result?.content ?? [];
|
||||
@@ -268,6 +326,7 @@ createInterface({ input: process.stdin }).on('line', (line) => {
|
||||
}), 50);
|
||||
} else if (ingestionFixture && method === 'turn/interrupt') {
|
||||
ingestionGeneration += 1;
|
||||
ingestionClock?.close();
|
||||
send({ id, result: {} });
|
||||
send({ method: 'turn/completed', params: { threadId, turn: { id: turnId, status: 'interrupted' } } });
|
||||
} else if (id != null) send({ id, result: {} });
|
||||
|
||||
@@ -0,0 +1,144 @@
|
||||
import assert from 'node:assert/strict';
|
||||
import { spawn, spawnSync } from 'node:child_process';
|
||||
import { randomUUID, createHash } from 'node:crypto';
|
||||
import { readFile, writeFile } from 'node:fs/promises';
|
||||
import { createServer } from 'node:http';
|
||||
import { createInterface } from 'node:readline';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
|
||||
const fixture = fileURLToPath(new URL('./in-feed-codex.mjs', import.meta.url));
|
||||
const output = process.argv[2];
|
||||
assert.ok(output, 'Pass an owned output JSON path');
|
||||
const reports = [];
|
||||
assert.equal(spawnSync(process.execPath, [fixture, '--version'], { encoding: 'utf8' }).stdout.trim(), 'codex-cli 0.115.0 (in-feed fixture)');
|
||||
|
||||
async function run({ name, count, intervalMs = 0, stopAfter, clockStatus = 200, prelude = 0 }) {
|
||||
const nonce = randomUUID().replaceAll('-', ''), key = randomUUID().replaceAll('-', ''), markerCode = 'q7';
|
||||
const messages = [], emissions = [], clockRecords = [], deltas = [];
|
||||
let clockPosts = 0, gateGets = 0, hello, pendingGate, clockResponse, clockEnded = false;
|
||||
let terminal, interruptRequestedAt, interruptAcknowledgedAt, uploadResponseAt, completedItem;
|
||||
let epochMs, stderr = '', failure, child, finished = false;
|
||||
const fail = error => { failure ??= error; child?.kill('SIGTERM'); };
|
||||
const maybeGate = () => {
|
||||
if (!hello || !pendingGate) return;
|
||||
epochMs = Date.now() + 25;
|
||||
pendingGate.writeHead(200, { 'Content-Type': 'application/json' }).end(JSON.stringify({ epochMs }));
|
||||
pendingGate = null;
|
||||
};
|
||||
const maybeFinishUpload = () => {
|
||||
if (!clockEnded || !terminal || !clockResponse || (stopAfter && !interruptAcknowledgedAt)) return;
|
||||
uploadResponseAt = Date.now();
|
||||
clockResponse.writeHead(clockStatus, { 'Content-Type': 'application/json' }).end(JSON.stringify({ received: emissions.length }));
|
||||
clockResponse = null;
|
||||
child.stdin.end();
|
||||
};
|
||||
const server = createServer((req, res) => {
|
||||
try {
|
||||
const url = new URL(req.url, 'http://127.0.0.1');
|
||||
assert.equal(url.searchParams.get('participant'), nonce);
|
||||
if (req.method === 'POST' && url.pathname === `/native-ingestion-gate/${key}/clock`) {
|
||||
clockPosts++;
|
||||
clockResponse = res;
|
||||
let buffer = '';
|
||||
req.setEncoding('utf8');
|
||||
req.on('data', chunk => {
|
||||
try {
|
||||
buffer += chunk;
|
||||
while (buffer.includes('\n')) {
|
||||
const cut = buffer.indexOf('\n'), record = JSON.parse(buffer.slice(0, cut));
|
||||
buffer = buffer.slice(cut + 1); clockRecords.push(record);
|
||||
if (record.type === 'hello') { assert.equal(hello, undefined); hello = record; maybeGate(); }
|
||||
else { assert.equal(record.type, 'emission'); emissions.push(record); }
|
||||
}
|
||||
} catch (error) { fail(error); }
|
||||
});
|
||||
req.on('end', () => { assert.equal(buffer, ''); clockEnded = true; maybeFinishUpload(); });
|
||||
} else {
|
||||
assert.equal(req.method, 'GET');
|
||||
assert.equal(url.pathname, `/native-ingestion-gate/${key}`);
|
||||
gateGets++; pendingGate = res; maybeGate();
|
||||
}
|
||||
} catch (error) { res.writeHead(500).end(); fail(error); }
|
||||
});
|
||||
await new Promise(resolve => server.listen(0, '127.0.0.1', resolve));
|
||||
const gate = `http://127.0.0.1:${server.address().port}/native-ingestion-gate/${key}`;
|
||||
const send = value => child.stdin.write(JSON.stringify(value) + '\n');
|
||||
child = spawn(process.execPath, [fixture, '--ingestion-stream-fixture'], { stdio: ['pipe', 'pipe', 'pipe'] });
|
||||
child.stderr.on('data', value => { stderr += value; });
|
||||
const watchdog = setTimeout(() => fail(new Error(`${name}: qualification watchdog`)), 10_000);
|
||||
createInterface({ input: child.stdout }).on('line', line => {
|
||||
try {
|
||||
const message = JSON.parse(line); messages.push({ at: Date.now(), message });
|
||||
if (message.id === 'initialize') send({ id: 'thread', method: 'thread/start', params: { model: 'fixture' } });
|
||||
else if (message.id === 'thread') {
|
||||
// Match the normal nested brief plus Markdown punctuation escaping.
|
||||
const prompt = `INGESTION_PERF nonce=${nonce} events=${count} intervalMs=${intervalMs} prelude=${prelude} markerCode=${markerCode} gate=${gate}`;
|
||||
const escaped = prompt.replaceAll('_', '\\_').replaceAll('=', '\\=');
|
||||
send({ id: 'turn', method: 'turn/start', params: { input: [{ type: 'text', text: JSON.stringify({ message: JSON.stringify({ objective: escaped }) }) }] } });
|
||||
} else if (message.method === 'item/agentMessage/delta' && message.params.itemId.startsWith('ingestion-')) {
|
||||
deltas.push({ delta: message.params.delta, at: Date.now() });
|
||||
if (stopAfter && deltas.length === stopAfter) {
|
||||
interruptRequestedAt = Date.now();
|
||||
send({ id: 'stop', method: 'turn/interrupt' });
|
||||
}
|
||||
} else if (message.method === 'item/completed' && message.params.item.id.startsWith('ingestion-')) completedItem = message.params.item;
|
||||
else if (message.method === 'item/tool/call') {
|
||||
assert.ok(completedItem, 'Assistant completion must precede semantic tools');
|
||||
assert.equal(uploadResponseAt, undefined, 'Tools must run before clock upload completes');
|
||||
send({ id: message.id, result: {} });
|
||||
} else if (message.id === 'stop') {
|
||||
interruptAcknowledgedAt = Date.now();
|
||||
assert.equal(uploadResponseAt, undefined, 'Stop acknowledgement cannot wait for clock upload');
|
||||
maybeFinishUpload();
|
||||
} else if (message.method === 'turn/completed') {
|
||||
terminal = message.params.turn;
|
||||
assert.equal(uploadResponseAt, undefined, 'Protocol terminal cannot wait for clock upload');
|
||||
maybeFinishUpload();
|
||||
}
|
||||
} catch (error) { fail(error); }
|
||||
});
|
||||
send({ id: 'initialize', method: 'initialize' });
|
||||
const exit = await new Promise(resolve => child.once('close', (code, signal) => resolve({ code, signal })));
|
||||
clearTimeout(watchdog); finished = true;
|
||||
server.closeAllConnections(); await new Promise(resolve => server.close(resolve));
|
||||
if (failure) throw failure;
|
||||
assert.equal(exit.code, clockStatus === 200 ? 0 : 1, stderr);
|
||||
assert.equal(clockPosts, 1); assert.equal(gateGets, 1);
|
||||
assert.deepEqual(hello, { type: 'hello', nonce, markerCode, events: count, intervalMs });
|
||||
assert.equal(emissions.length, deltas.length);
|
||||
assert.ok(deltas.length > 0);
|
||||
for (let index = 0; index < emissions.length; index++) {
|
||||
const event = emissions[index];
|
||||
assert.deepEqual({ ...event, emittedMs: 0 }, { type: 'emission', nonce, markerCode, index, intendedMs: epochMs + index * intervalMs, emittedMs: 0 });
|
||||
assert.ok(event.emittedMs <= deltas[index].at);
|
||||
assert.equal(deltas[index].delta, `[${markerCode}${index.toString(36)}] `);
|
||||
}
|
||||
if (stopAfter) {
|
||||
assert.equal(terminal.status, 'interrupted'); assert.ok(interruptAcknowledgedAt);
|
||||
assert.ok(deltas.length >= stopAfter && deltas.length < count);
|
||||
assert.ok(uploadResponseAt >= interruptAcknowledgedAt);
|
||||
assert.equal(completedItem, undefined);
|
||||
} else {
|
||||
assert.equal(terminal.status, 'completed'); assert.equal(deltas.length, count);
|
||||
assert.equal(completedItem.text, deltas.map(item => item.delta).join(''));
|
||||
if (count === 512) assert.equal(completedItem.text.length, 3548);
|
||||
const tools = messages.filter(row => row.message.method === 'item/tool/call').map(row => row.message.params.tool);
|
||||
assert.deepEqual(tools, ['report_progress', 'paperclip_finish']);
|
||||
}
|
||||
if (clockStatus !== 200) assert.match(stderr, /Ingestion fixture clock .*HTTP 503/);
|
||||
reports.push({ name, exit, count, emitted: deltas.length, clockPosts, clockRecords: clockRecords.length,
|
||||
completedTextCharacters: completedItem?.text.length, terminal: terminal.status,
|
||||
uploadHeldUntilAfterTerminal: true, stopAcknowledgedBeforeUploadResponse: stopAfter ? true : undefined,
|
||||
explicitClockError: clockStatus !== 200 ? stderr.trim() : undefined });
|
||||
}
|
||||
try {
|
||||
await run({ name: 'full-512-with-prelude', count: 512, prelude: 8 });
|
||||
await run({ name: 'interrupt-with-held-clock-response', count: 512, intervalMs: 10, stopAfter: 3 });
|
||||
await run({ name: 'explicit-upload-error', count: 16, clockStatus: 503 });
|
||||
const proof = { fixture, fixtureSha256: createHash('sha256').update(await readFile(fixture)).digest('hex'), reports, passed: true };
|
||||
await writeFile(output, JSON.stringify(proof, null, 2) + '\n');
|
||||
console.log(JSON.stringify(proof));
|
||||
} catch (error) {
|
||||
await writeFile(output, JSON.stringify({ fixture, reports, passed: false, error: error.stack }, null, 2) + '\n');
|
||||
throw error;
|
||||
}
|
||||
@@ -6,7 +6,13 @@ import path from "node:path";
|
||||
import { expect, test } from "@playwright/test";
|
||||
import { json } from "./agent-chat.shared";
|
||||
|
||||
type Visible = { nonce: string; index: number; intendedMs: number; emittedMs: number; visibleMs: number };
|
||||
type SeenMarker = { markerCode: string; index: number; visibleMs: number };
|
||||
type Emission = { nonce: string; markerCode: string; index: number; intendedMs: number; emittedMs: number; receivedAtMs: number };
|
||||
type Visible = SeenMarker & Emission;
|
||||
type ClockStream = { nonce: string; markerCode: string; opened: boolean; hello: boolean; ended: boolean; aborted: boolean;
|
||||
partialBytes: number; records: Map<number, Emission>; error?: string;
|
||||
providerReadyAtMs?: number; clockHelloAtMs?: number; readinessObservedAtMs?: number };
|
||||
type ClockCase = { events: number; intervalMs: number; epochMs?: number; streams: Map<string, ClockStream>; errors: string[] };
|
||||
type Case = { name: string; runs: number; events: number; intervalMs: number; prelude: number };
|
||||
const smoke = process.env.PAPERCLIP_INGESTION_SMOKE === "1";
|
||||
const cases: Case[] = smoke ? [{ name: "qualification", runs: 1, events: 16, intervalMs: 25, prelude: 0 }] : [
|
||||
@@ -24,32 +30,86 @@ const stats = (values: number[]) => {
|
||||
|
||||
test("native ingestion reaches the shipped UI promptly and survives reload", async ({ page, request }, info) => {
|
||||
const pending = new Map<string, Map<string, ServerResponse>>();
|
||||
const clocks = new Map<string, ClockCase>();
|
||||
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 key = url.pathname.replace("/native-ingestion-gate/", "");
|
||||
const participant = url.searchParams.get("participant");
|
||||
if (req.method !== "GET" || !pending.has(key) || !participant || !/^[a-z0-9]+$/.test(participant)) {
|
||||
res.writeHead(404).end(); return;
|
||||
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;
|
||||
}
|
||||
pending.get(key)!.set(participant, res);
|
||||
res.on("close", () => pending.get(key)?.delete(participant));
|
||||
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) throw new Error("Clock hello mismatch");
|
||||
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<void>(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<string>(), visible: Visible[] = [];
|
||||
const seen = new Set<string>(), visible: SeenMarker[] = [];
|
||||
Object.assign(window, { nativeIngestionVisible: visible });
|
||||
let scheduled = false, dirty = false;
|
||||
const snapshot = () => {
|
||||
const text = document.body?.innerText ?? "";
|
||||
const pattern = /INGEST_([a-z0-9]+)_(\d+)_(\d+)_(\d+)/g;
|
||||
const entries: Omit<Visible, "visibleMs">[] = [];
|
||||
const pattern = /\[([a-z0-9]{2})([a-z0-9]{1,3})\]/g;
|
||||
const entries: Omit<SeenMarker, "visibleMs">[] = [];
|
||||
for (const match of text.matchAll(pattern)) {
|
||||
const key = `${match[1]}:${match[2]}`;
|
||||
const index = Number.parseInt(match[2], 36);
|
||||
const key = `${match[1]}:${index}`;
|
||||
if (seen.has(key)) continue;
|
||||
entries.push({ nonce: match[1], index: Number(match[2]), intendedMs: Number(match[3]), emittedMs: Number(match[4]) });
|
||||
entries.push({ markerCode: match[1], index });
|
||||
}
|
||||
return entries;
|
||||
};
|
||||
@@ -65,7 +125,7 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
|
||||
// between frames has not had this snapshot's paint opportunity.
|
||||
const visibleMs = Date.now();
|
||||
for (const entry of entries) {
|
||||
const key = `${entry.nonce}:${entry.index}`;
|
||||
const key = `${entry.markerCode}:${entry.index}`;
|
||||
if (seen.has(key)) continue;
|
||||
seen.add(key); visible.push({ ...entry, visibleMs });
|
||||
}
|
||||
@@ -82,13 +142,21 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
|
||||
});
|
||||
const company = await json(await request.post("/api/companies", { data: { name: `Ingestion ${Date.now()}` } }));
|
||||
const original = await json(await request.get("/api/instance/settings/experimental"));
|
||||
const captureMode = (value: unknown) => typeof value === "string" ? value
|
||||
: value && typeof value === "object" && "mode" in value ? String(value.mode) : "off";
|
||||
const browserRecording = { trace: captureMode(info.project.use.trace), video: captureMode(info.project.use.video),
|
||||
screencast: captureMode(info.project.use.trace) !== "off" || captureMode(info.project.use.video) !== "off" };
|
||||
const report: Record<string, unknown> = {
|
||||
protocolVersion: "v2-compact-markers-original-clock-stream",
|
||||
admissionProtocol: "sequential-provider-and-clock-readiness-before-common-epoch",
|
||||
setupReadinessPollingMs: 50,
|
||||
dependency: "Deterministic Codex app-server fixture; real application/runtime/port/PostgreSQL/live WebSocket/shipped UI",
|
||||
variant: process.env.PAPERCLIP_INGESTION_VARIANT ?? "unlabelled", qualificationOnly: smoke,
|
||||
diagnosticOnly: Boolean(process.env.PAPERCLIP_NATIVE_INGESTION_DIAGNOSTIC_DIR),
|
||||
performanceAcceptanceEligible: !process.env.PAPERCLIP_NATIVE_INGESTION_DIAGNOSTIC_DIR,
|
||||
performanceAcceptanceEligible: !process.env.PAPERCLIP_NATIVE_INGESTION_DIAGNOSTIC_DIR && !browserRecording.screencast,
|
||||
browserRecording,
|
||||
startedAtMs: Date.now(),
|
||||
emittedClock: "Provider process epoch; original schedule retained even if stdout backpressures",
|
||||
emittedClock: "Original provider-process clocks on one pre-opened streamed HTTP channel per provider; no per-token request or flush wait",
|
||||
observedClock: "Browser epoch after a DOM update and one paint opportunity",
|
||||
cases: [], companyId: company.id,
|
||||
health: await json(await request.get("/api/health")),
|
||||
@@ -110,7 +178,17 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
|
||||
report.fixtureSha256 = createHash("sha256").update(await readFile(path.resolve(import.meta.dirname, "fixtures/in-feed-codex.mjs"))).digest("hex");
|
||||
const protocolFiles = ["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"];
|
||||
report.measurementSha256 = createHash("sha256").update(Buffer.concat(await Promise.all(protocolFiles.map(file => readFile(path.resolve(import.meta.dirname, file)))))).digest("hex");
|
||||
const protocolSources = await Promise.all(protocolFiles.map(async file => ({ file, bytes: await readFile(path.resolve(import.meta.dirname, file)) })));
|
||||
report.measurementSha256 = createHash("sha256").update(Buffer.concat(protocolSources.map(source => source.bytes))).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(),
|
||||
sourceHashes: Object.fromEntries(protocolSources.map(source => [source.file, createHash("sha256").update(source.bytes).digest("hex")])),
|
||||
driverDependencySource: process.env.PAPERCLIP_INGESTION_DRIVER_DEPENDENCIES ?? "Checkout tooling dependencies",
|
||||
driverLockSha256: createHash("sha256").update(await readFile(path.join(harnessRoot, "pnpm-lock.yaml"))).digest("hex"),
|
||||
scope: "Separate browser harness and driver-only tooling; every measured app uses its own frozen clean source and matching compiled stamp",
|
||||
};
|
||||
try {
|
||||
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" } }));
|
||||
@@ -131,14 +209,26 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
|
||||
|
||||
async function perform(workload: Case, stop = false) {
|
||||
const key = randomUUID().replaceAll("-", ""), nonce = randomUUID().replaceAll("-", "");
|
||||
const markerCode = nextMarkerCode();
|
||||
pending.set(key, new Map());
|
||||
const prompt = (token: string, prelude: number) => `Measure native streaming for ${token.slice(0, 8)}.\n\nINGESTION_PERF nonce=${token} events=${workload.events} intervalMs=${workload.intervalMs} prelude=${prelude} gate=${base}${key}`;
|
||||
const state: ClockCase = { events: workload.events, intervalMs: workload.intervalMs, streams: new Map(), errors: [] };
|
||||
clocks.set(key, state);
|
||||
const register = (token: string, code: string) => state.streams.set(token, { nonce: token, markerCode: code,
|
||||
opened: false, hello: false, ended: false, aborted: false, partialBytes: 0, records: new Map() });
|
||||
register(nonce, markerCode);
|
||||
const compactOutput = Array.from({ length: workload.events }, (_, index) => `[${markerCode}${index.toString(36)}] `).join("");
|
||||
const preludeOutput = workload.prelude ? `${".".repeat(Math.max(0, workload.prelude - 1))} PRELUDE_${nonce}_READY ` : "";
|
||||
expect([...compactOutput].length, "Full measured completion must fit the unchanged Rust text bound").toBeLessThan(4000);
|
||||
expect([...preludeOutput + compactOutput].length, "Prelude and measured text must remain below that bound").toBeLessThan(4000);
|
||||
const active: Record<string, unknown> = { ...workload, stop, nonce, markerCode, phase: "setup", key };
|
||||
report.activeCase = active;
|
||||
const prompt = (token: string, code: string, prelude: number) => `Measure native streaming for ${token.slice(0, 8)}.\n\nINGESTION_PERF nonce=${token} events=${workload.events} intervalMs=${workload.intervalMs} prelude=${prelude} markerCode=${code} gate=${base}${key}`;
|
||||
const before = new Set((await json(await request.get(`/api/companies/${company.id}/issues`))).map((issue: { id: string }) => issue.id));
|
||||
await page.goto(`/${company.issuePrefix}/agents/${agents[0].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(nonce, workload.prelude));
|
||||
await dialog.getByRole("textbox", { name: "editable markdown" }).fill(prompt(nonce, markerCode, workload.prelude));
|
||||
await expect(dialog.getByRole("button", { name: "Select assignee" })).toContainText(agents[0].name);
|
||||
await dialog.getByRole("button", { name: "Create task", exact: true }).click();
|
||||
let issue: any;
|
||||
@@ -147,27 +237,68 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
|
||||
return Boolean(issue);
|
||||
}).toBe(true);
|
||||
const issueIds = [issue.id];
|
||||
active.issueIds = issueIds;
|
||||
const issuedTokens = new Map<string, string>([[issue.id, nonce]]);
|
||||
for (let index = 1; index < workload.runs; index++) {
|
||||
const backgroundToken = randomUUID().replaceAll("-", "");
|
||||
const backgroundToken = randomUUID().replaceAll("-", ""), backgroundCode = nextMarkerCode();
|
||||
register(backgroundToken, backgroundCode);
|
||||
const background = await json(await request.post(`/api/companies/${company.id}/issues`, { data: {
|
||||
title: `Ingestion background ${key} ${index}`, description: prompt(backgroundToken, 0),
|
||||
title: `Ingestion background ${key} ${index}`, description: prompt(backgroundToken, backgroundCode, 0),
|
||||
status: "backlog", assigneeAgentId: agents[Math.floor(index / 50)].id,
|
||||
} }));
|
||||
issueIds.push(background.id);
|
||||
issuedTokens.set(background.id, backgroundToken);
|
||||
}
|
||||
for (let offset = 1; offset < issueIds.length; offset += 10) {
|
||||
await Promise.all(issueIds.slice(offset, offset + 10).map(async id => json(await request.patch(`/api/issues/${id}`, { data: { status: "todo" } }))));
|
||||
const readinessDeadlineMs = Date.now() + 180_000;
|
||||
active.phase = "activation-and-provider-readiness";
|
||||
for (let index = 1; index < issueIds.length; index++) {
|
||||
const id = issueIds[index], token = issuedTokens.get(id)!;
|
||||
active.activation = { index, issueId: id, token, startedAtMs: Date.now(), readinessDeadlineMs };
|
||||
await json(await request.patch(`/api/issues/${id}`, { data: { status: "todo" } }));
|
||||
// PATCH schedules wakeup asynchronously. Wait for this actual provider,
|
||||
// while the common emission gate remains closed, before scheduling more.
|
||||
await expect.poll(() => pending.get(key)!.has(token) && state.streams.get(token)!.hello,
|
||||
{ timeout: Math.max(1, readinessDeadlineMs - Date.now()), intervals: [50] }).toBe(true);
|
||||
state.streams.get(token)!.readinessObservedAtMs = Date.now();
|
||||
}
|
||||
const route = `/${company.issuePrefix}/issues/${issue.identifier}`;
|
||||
await page.goto(route);
|
||||
await expect.poll(() => pending.get(key)!.size, { timeout: 180_000 }).toBe(workload.runs);
|
||||
if (workload.prelude) await expect(page.getByText(`PRELUDE_${nonce}_READY`, { exact: false }).first()).toBeVisible({ timeout: 120_000 });
|
||||
active.phase = "provider-and-clock-readiness";
|
||||
await expect.poll(() => ({ providers: pending.get(key)!.size,
|
||||
clocks: [...state.streams.values()].filter(stream => stream.hello).length }),
|
||||
{ timeout: Math.max(1, readinessDeadlineMs - Date.now()), intervals: [50] }).toEqual({ providers: workload.runs, clocks: workload.runs });
|
||||
expect(Date.now(), "All participants must be ready within the shared activation/readiness budget").toBeLessThanOrEqual(readinessDeadlineMs);
|
||||
let preludeProof: Record<string, unknown> | undefined;
|
||||
if (workload.prelude) {
|
||||
await expect(page.getByText(`PRELUDE_${nonce}_READY`, { exact: false }).first()).toBeVisible({ timeout: 120_000 });
|
||||
active.phase = "independent-prelude-verification";
|
||||
const listed = await json(await request.get(`/api/companies/${company.id}/heartbeat-runs?limit=1000`));
|
||||
const foreground = listed.find((run: any) => (run.issueId ?? run.nativeIssueId ?? run.contextSnapshot?.issueId ?? run.contextSnapshot?.taskId) === issue.id);
|
||||
expect(foreground).toBeTruthy();
|
||||
const identities = new Set<string>(); let count = 0, afterSeq = 0;
|
||||
while (true) {
|
||||
const rows = await json(await request.get(`/api/heartbeat-runs/${foreground.id}/events?afterSeq=${afterSeq}&limit=1000`));
|
||||
if (!rows.length) break;
|
||||
for (const row of rows) {
|
||||
const event = row.payload?.prpEvent;
|
||||
if (event?.eventType !== "item.delta" || !String(event.itemId ?? "").startsWith("prelude-")) continue;
|
||||
expect(event.payload?.text).toBe(count === workload.prelude - 1 ? ` PRELUDE_${nonce}_READY ` : ".");
|
||||
count++; identities.add(`${event.sourceInstanceId}:${event.sourceEventId}:${event.sourceSeq}`);
|
||||
}
|
||||
const next = Math.max(...rows.map((row: { seq: number }) => row.seq));
|
||||
expect(next).toBeGreaterThan(afterSeq); afterSeq = next;
|
||||
}
|
||||
preludeProof = { runId: foreground.id, individualDeltas: count, uniqueSourceIdentities: identities.size,
|
||||
retainedGlobalHead: afterSeq, verifiedAtMs: Date.now(), method: "Independent normal event API read before common emission gate release" };
|
||||
active.preludeProof = preludeProof;
|
||||
expect(count).toBe(workload.prelude); expect(identities.size).toBe(count);
|
||||
}
|
||||
const epochMs = Date.now() + 1000;
|
||||
state.epochMs = epochMs; active.epochMs = epochMs; active.phase = "streaming";
|
||||
const participants = [...pending.get(key)!.keys()];
|
||||
active.participants = participants;
|
||||
for (const response of pending.get(key)!.values()) response.writeHead(200, { "Content-Type": "application/json" }).end(JSON.stringify({ epochMs }));
|
||||
const observed = () => page.evaluate(token => (window as unknown as { nativeIngestionVisible: Visible[] }).nativeIngestionVisible.filter(item => item.nonce === token), nonce);
|
||||
const observed = () => page.evaluate(code => (window as unknown as { nativeIngestionVisible: SeenMarker[] }).nativeIngestionVisible.filter(item => item.markerCode === code), markerCode);
|
||||
await expect.poll(async () => (await observed()).length, { timeout: 120_000 }).toBeGreaterThanOrEqual(stop ? 3 : workload.events);
|
||||
let stopRequestedAt: number | undefined;
|
||||
if (stop) {
|
||||
@@ -176,12 +307,14 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
|
||||
stopRequestedAt = Date.now();
|
||||
await button.click();
|
||||
}
|
||||
active.phase = "terminal-drainage";
|
||||
let runs: any[] = [];
|
||||
await expect.poll(async () => {
|
||||
runs = (await json(await request.get(`/api/companies/${company.id}/heartbeat-runs?limit=1000`))).filter((run: { issueId?: string; nativeIssueId?: string; contextSnapshot?: { issueId?: string; taskId?: string } }) => issueIds.includes(run.issueId ?? run.nativeIssueId ?? run.contextSnapshot?.issueId ?? run.contextSnapshot?.taskId));
|
||||
return runs.length === issueIds.length && runs.every(run => ["succeeded", "cancelled", "failed"].includes(run.status));
|
||||
}, { timeout: 180_000 }).toBe(true);
|
||||
const settledAt = Date.now(), visible = await observed();
|
||||
const settledAt = Date.now(), seenMarkers = await observed();
|
||||
active.settledAt = settledAt; active.seenMarkers = seenMarkers;
|
||||
const details = [];
|
||||
for (const run of runs) details.push(await json(await request.get(`/api/heartbeat-runs/${run.id}`)));
|
||||
runs = details;
|
||||
@@ -191,8 +324,24 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
|
||||
(await json(await request.get(`/api/heartbeat-runs/${runs[0].id}`))).resultJson?.nativeCancellation?.dispatchState,
|
||||
{ timeout: 30_000 }).toBe("acknowledged");
|
||||
const acknowledgedAt = stop ? Date.now() : undefined;
|
||||
active.phase = "clock-reconciliation";
|
||||
await expect.poll(() => [...state.streams.values()].every(stream => !stream.error && (stop
|
||||
? seenMarkers.every(marker => state.streams.get(nonce)!.records.has(marker.index))
|
||||
: stream.records.size === workload.events && stream.ended)), { timeout: 30_000 }).toBe(true);
|
||||
expect(state.errors).toEqual([]);
|
||||
const visible: Visible[] = seenMarkers.map(marker => {
|
||||
const original = state.streams.get(nonce)!.records.get(marker.index);
|
||||
expect(original, `Missing original provider clock for visible ${marker.markerCode}:${marker.index}`).toBeTruthy();
|
||||
return { ...original!, ...marker }; // Keep the original paint timestamp; never restamp after clock arrival.
|
||||
});
|
||||
if (!stop) expect(new Set(visible.map(item => item.index)).size).toBe(workload.events);
|
||||
const result = { ...workload, stop, nonce, runIds: runs.map(run => run.id), participants: participants.length, visible,
|
||||
// Terminal drainage has finished. Derive paint gaps from the retained
|
||||
// observer timestamps without adding work to the streaming observer.
|
||||
const paintTimes = [...new Set(visible.map(item => item.visibleMs))].sort((a, b) => a - b);
|
||||
const { count, medianMs, maxMs } = stats(paintTimes.slice(1).map((at, index) => at - paintTimes[index]));
|
||||
const interChunkGaps = { paintCount: paintTimes.length, count, medianMs, maxMs };
|
||||
const clockEvidence: Array<Record<string, unknown>> = [];
|
||||
const result = { ...workload, stop, nonce, markerCode, epochMs, preludeProof, clockEvidence, interChunkGaps, runIds: runs.map(run => run.id), participants: participants.length, visible, matchedVisibleClocks: visible.length,
|
||||
firstVisibleAfterEmissionMs: visible[0].visibleMs - visible[0].emittedMs,
|
||||
emissionToVisible: stats(visible.map(item => item.visibleMs - item.emittedMs)),
|
||||
intendedToVisible: stats(visible.map(item => item.visibleMs - item.intendedMs)),
|
||||
@@ -202,18 +351,20 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
|
||||
stopThroughAcknowledgementMs: acknowledgedAt! - stopRequestedAt }) };
|
||||
await page.screenshot({ path: info.outputPath(`${workload.name}${stop ? "-stop" : ""}.png`), fullPage: true });
|
||||
await page.reload();
|
||||
if (!stop) await expect(page.getByText(`INGEST_${nonce}_${workload.events - 1}_`, { exact: false }).first()).toBeVisible();
|
||||
if (!stop) await expect(page.getByText(`[${markerCode}${(workload.events - 1).toString(36)}]`, { exact: false }).first()).toBeVisible();
|
||||
else await expect(page.getByRole("button", { name: "Stop", exact: true })).toHaveCount(0);
|
||||
const comments = await json(await request.get(`/api/issues/${issue.id}/comments`));
|
||||
if (!stop) expect(JSON.stringify(comments)).toContain(`INGEST_${nonce}_${workload.events - 1}_`);
|
||||
if (!stop) expect(JSON.stringify(comments)).toContain(`[${markerCode}${(workload.events - 1).toString(36)}]`);
|
||||
else {
|
||||
const lastVisible = visible.at(-1)!;
|
||||
await expect(page.getByText(`INGEST_${nonce}_${lastVisible.index}_`, { exact: false }).first()).toBeVisible();
|
||||
await expect(page.getByText(`[${markerCode}${lastVisible.index.toString(36)}]`, { exact: false }).first()).toBeVisible();
|
||||
}
|
||||
active.phase = "independent-durable-verification";
|
||||
const durable = [];
|
||||
for (const run of runs) {
|
||||
const taskId = run.issueId ?? run.nativeIssueId ?? run.contextSnapshot?.issueId ?? run.contextSnapshot?.taskId;
|
||||
const token = issuedTokens.get(taskId)!;
|
||||
const code = state.streams.get(token)!.markerCode;
|
||||
expect(token).toBeTruthy();
|
||||
const deltas: Array<{ index: number; identity: string }> = [];
|
||||
let afterSeq = 0;
|
||||
@@ -223,33 +374,80 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
|
||||
for (const row of rows) {
|
||||
const event = row.payload?.prpEvent;
|
||||
if (event?.eventType !== "item.delta") continue;
|
||||
for (const match of String(event.payload?.text ?? "").matchAll(new RegExp(`INGEST_${token}_(\\d+)_(\\d+)_(\\d+)`, "g"))) {
|
||||
deltas.push({ index: Number(match[1]), identity: `${event.sourceInstanceId}:${event.sourceEventId}:${event.sourceSeq}` });
|
||||
for (const match of String(event.payload?.text ?? "").matchAll(new RegExp(`\\[${code}([a-z0-9]{1,3})\\]`, "g"))) {
|
||||
const index = Number.parseInt(match[1], 36);
|
||||
expect(event.payload.text, "Every durable normalized delta retains its complete compact payload").toBe(`[${code}${index.toString(36)}] `);
|
||||
deltas.push({ index, identity: `${event.sourceInstanceId}:${event.sourceEventId}:${event.sourceSeq}` });
|
||||
}
|
||||
}
|
||||
const next = Math.max(...rows.map((row: { seq: number }) => row.seq));
|
||||
expect(next).toBeGreaterThan(afterSeq); afterSeq = next;
|
||||
}
|
||||
active.latestDurableRead = { runId: run.id, nonce: token, markerCode: code, deltas };
|
||||
await expect.poll(() => deltas.every(delta => state.streams.get(token)!.records.has(delta.index)), { timeout: 30_000 }).toBe(true);
|
||||
for (const delta of deltas) {
|
||||
const original = state.streams.get(token)!.records.get(delta.index)!;
|
||||
expect({ nonce: original.nonce, markerCode: original.markerCode, index: original.index })
|
||||
.toEqual({ nonce: token, markerCode: code, index: delta.index });
|
||||
}
|
||||
expect(new Set(deltas.map(item => item.identity)).size).toBe(deltas.length);
|
||||
expect(deltas.map(item => item.index)).toEqual(Array.from({ length: stop ? deltas.length : workload.events }, (_, index) => index));
|
||||
if (stop) for (const item of visible) expect(deltas.some(delta => delta.index === item.index)).toBe(true);
|
||||
durable.push({ runId: run.id, token, individualDeltas: deltas.length, uniqueSourceIdentities: deltas.length, ordered: true });
|
||||
durable.push({ runId: run.id, token, markerCode: code, individualDeltas: deltas.length, uniqueSourceIdentities: deltas.length, matchedSourceClocks: deltas.length, ordered: true });
|
||||
}
|
||||
await expect.poll(() => [...state.streams.values()].every(stream => stream.ended || stream.aborted), { timeout: 30_000 }).toBe(true);
|
||||
expect(state.errors).toEqual([]);
|
||||
expect([...state.streams.values()].every(stream => !stream.error)).toBe(true);
|
||||
clockEvidence.push(...[...state.streams.values()].map(stream => ({ nonce: stream.nonce, markerCode: stream.markerCode,
|
||||
hello: stream.hello, ended: stream.ended, aborted: stream.aborted, error: stream.error, partialBytes: stream.partialBytes,
|
||||
providerReadyAtMs: stream.providerReadyAtMs, clockHelloAtMs: stream.clockHelloAtMs, readinessObservedAtMs: stream.readinessObservedAtMs,
|
||||
emissions: [...stream.records.values()].sort((a, b) => a.index - b.index) })));
|
||||
(report.cases as unknown[]).push({ ...result, durable, persistedAfterReload: true });
|
||||
await info.attach(`${workload.name}${stop ? "-stop" : ""}-metrics`, { body: JSON.stringify(result, null, 2), contentType: "application/json" });
|
||||
pending.delete(key);
|
||||
pending.delete(key); clocks.delete(key); delete report.activeCase;
|
||||
}
|
||||
for (const workload of cases) await perform(workload);
|
||||
await perform({ name: "operator-stop", runs: 1, events: 512, intervalMs: 100, prelude: 0 }, true);
|
||||
report.complete = true;
|
||||
} catch (error) {
|
||||
report.primaryFailure = { name: error instanceof Error ? error.name : "UnknownError",
|
||||
message: error instanceof Error ? error.message : String(error), stack: error instanceof Error ? error.stack : undefined };
|
||||
throw error;
|
||||
} finally {
|
||||
await info.attach("native-ingestion-evidence", { body: JSON.stringify(report, null, 2), contentType: "application/json" });
|
||||
if (report.activeCase) {
|
||||
const active = report.activeCase as Record<string, unknown>;
|
||||
try {
|
||||
active.failureBrowser = await page.evaluate(code => ({ capturedAtMs: Date.now(),
|
||||
observed: (window as unknown as { nativeIngestionVisible?: SeenMarker[] }).nativeIngestionVisible?.filter(item => item.markerCode === code),
|
||||
domTextLength: (document.body?.innerText ?? "").length,
|
||||
renderedTexts: [...document.querySelectorAll("[data-conversation-text]")].map(node => ({ kind: node.getAttribute("data-conversation-text"), text: node.textContent })),
|
||||
domMarkers: [...(document.body?.innerText ?? "").matchAll(/\[([a-z0-9]{2})([a-z0-9]{1,3})\]/g)]
|
||||
.filter(match => match[1] === code).map(match => ({ markerCode: match[1], index: Number.parseInt(match[2], 36) })),
|
||||
}), active.markerCode as string);
|
||||
} catch (error) { active.captureError = String(error); }
|
||||
active.failurePendingProviders = [...(pending.get(active.key as string)?.keys() ?? [])];
|
||||
const state = clocks.get(active.key as string);
|
||||
active.failureClockEvidence = state && { errors: state.errors, streams: [...state.streams.values()].map(stream => ({
|
||||
...stream, records: [...stream.records.values()],
|
||||
})) };
|
||||
}
|
||||
// Retain primary failure evidence even if API teardown cannot finish. This
|
||||
// attachment is never eligible for the standalone acceptance comparator.
|
||||
await info.attach("native-ingestion-precleanup-evidence", { body: JSON.stringify(report, null, 2), contentType: "application/json" });
|
||||
for (const requests of pending.values()) for (const response of requests.values()) response.writeHead(503).end();
|
||||
gate.closeAllConnections();
|
||||
await new Promise<void>(resolve => gate.close(() => resolve()));
|
||||
const live = await json(await request.get(`/api/companies/${company.id}/live-runs`)).catch(() => []);
|
||||
for (const run of live) await request.post(`/api/heartbeat-runs/${run.id}/cancel`);
|
||||
await request.patch("/api/instance/settings/experimental", { data: { enableNativeRunner: original.enableNativeRunner,
|
||||
enableWorktreeRunExecution: original.enableWorktreeRunExecution } });
|
||||
const cleanupErrors: Array<{ step: string; message: string }> = [];
|
||||
const attemptCleanup = async (step: string, action: () => Promise<unknown>) => {
|
||||
try { return await action(); } catch (error) { cleanupErrors.push({ step, message: String(error) }); }
|
||||
};
|
||||
const live = await attemptCleanup("list-live-runs", async () => json(await request.get(`/api/companies/${company.id}/live-runs`))) as any[] | undefined;
|
||||
for (const run of live ?? []) await attemptCleanup(`cancel-${run.id}`, async () => json(await request.post(`/api/heartbeat-runs/${run.id}/cancel`)));
|
||||
await attemptCleanup("restore-experimental-settings", async () => json(await request.patch("/api/instance/settings/experimental", {
|
||||
data: { enableNativeRunner: original.enableNativeRunner, enableWorktreeRunExecution: original.enableWorktreeRunExecution } })));
|
||||
report.cleanup = { errors: cleanupErrors, complete: cleanupErrors.length === 0 };
|
||||
if (cleanupErrors.length) { report.complete = false; report.performanceAcceptanceEligible = false; }
|
||||
await info.attach("native-ingestion-evidence", { body: JSON.stringify(report, null, 2), contentType: "application/json" });
|
||||
if (!report.primaryFailure) expect(cleanupErrors, "API cleanup must succeed before accepting an invocation").toEqual([]);
|
||||
}
|
||||
});
|
||||
@@ -6,6 +6,8 @@ import base from "./playwright.config";
|
||||
|
||||
const server = base.webServer as Exclude<typeof base.webServer, unknown[]>;
|
||||
const evidence = process.env.PAPERCLIP_INGESTION_EVIDENCE_DIR;
|
||||
const browserRecording = Boolean(process.env.PAPERCLIP_NATIVE_INGESTION_DIAGNOSTIC_DIR) || process.env.PAPERCLIP_INGESTION_BROWSER_CAPTURE === "1";
|
||||
process.env.PAPERCLIP_INGESTION_BROWSER_RECORDING = browserRecording ? "on" : "off";
|
||||
const appRoot = path.resolve(process.env.PAPERCLIP_INGESTION_APP_ROOT ?? path.resolve(import.meta.dirname, "../.."));
|
||||
const quote = (value: string) => `'${value.replaceAll("'", "'\\''")}'`;
|
||||
const bootstrap = path.resolve(import.meta.dirname, "fixtures/native-ingestion-bootstrap.mjs");
|
||||
@@ -27,7 +29,7 @@ export default defineConfig({
|
||||
timeout: 600_000,
|
||||
...(evidence ? { outputDir: path.join(evidence, "test-results"),
|
||||
reporter: [["list"], ["json", { outputFile: path.join(evidence, "playwright.json") }]] as ReporterDescription[] } : {}),
|
||||
use: { ...base.use, trace: "retain-on-failure", actionTimeout: 20_000 },
|
||||
use: { ...base.use, trace: browserRecording ? "on" : "off", video: "off", actionTimeout: 20_000 },
|
||||
webServer: {
|
||||
...server,
|
||||
cwd: appRoot,
|
||||
|
||||
Reference in new issue
Block a user