test: observe native execution callback verification cost

tests/fixtures/public-safety-access/execution-browser.tsmodified+65-12

Changes to tests/fixtures/public-safety-access/execution-browser.ts

		⋯4 linesShow the 4 lines above this hunk@@ -5,6 +5,7 @@ import { mkdir, writeFile } from 'node:fs/promises'
5	5	import { join, resolve } from 'node:path'
6	6	import { randomUUID } from 'node:crypto'
7	7	import { setupHandoverFixture } from './handover-setup'
	8	+import { createExecutionCallbackObserver } from '../../helpers/execution-worker-observation'
8	9
9	10	// Real native setup owns the guarded disposable SQLite file before any application boot.
10	11	const fixture = await setupHandoverFixture()
		⋯12 linesShow the 12 lines above this hunk@@ -23,6 +24,15 @@ let delayStart = true, releaseStart: (() => void) | undefined
23	24	let streamStarts = 0, startRequests = 0, admissionRequests = 0, closeStream: (() => Promise<void>) | undefined
24	25	let closeWorkspaceStream: (() => Promise<void>) | undefined, releaseWorkspace: (() => void) | undefined
25	26	let holdWorkspace = false, workspaceHeld = false
	27	+let preAdmissionRenewalArmed = false, preAdmissionRenewed = false, preAdmissionHoldExpired = false
	28	+const preAdmissionReads = new Set<() => void>()
	29	+let preAdmissionTimeout: ReturnType<typeof setTimeout> | undefined
	30	+function releasePreAdmissionReads() {
	31	+ preAdmissionRenewalArmed = false
	32	+ clearTimeout(preAdmissionTimeout)
	33	+ for (const release of preAdmissionReads) release()
	34	+ preAdmissionReads.clear()
	35	+}
26	36
27	37	// Diagnostic state belongs only to this disposable fixture, never the execution controller.
28	38	type RouteClass = 'execution-start' | 'execution-close' | 'execution-admission' | 'execution-events' | 'execution-status' | 'workspace-read'
		⋯5 linesShow the 5 lines above this hunk@@ -34,6 +44,14 @@ let nextOrdinal = 0
34	44	const scalar = (value: unknown) => typeof value === 'number' && Number.isFinite(value) && Math.abs(value) <= Number.MAX_SAFE_INTEGER ? value : null
35	45	const field = (value: unknown, name: string): unknown => value && typeof value === 'object' ? Object.getOwnPropertyDescriptor(value, name)?.value : undefined
36	46	const choice = <T extends string>(value: unknown, values: readonly T[]): T | null => values.find(item => item === value) ?? null
	47	+const callbackObserver = createExecutionCallbackObserver(value => console.log(JSON.stringify(value)))
	48	+function observeAcceptedWorker(acceptedOrdinal: number) {
	49	+ try {
	50	+ const worker = field(executionOwner().slot, 'worker')
	51	+ if (worker instanceof Worker) callbackObserver.observe(worker, acceptedOrdinal)
	52	+ }
	53	+ catch { /* Private observation must not change an accepted native response. */ }
	54	+}
37	55	const ordinal = (slot: object) => {
38	56	const existing = ordinals.get(slot)
39	57	if (existing !== undefined) return existing
		⋯12 linesShow the 12 lines above this hunk@@ -52,8 +70,8 @@ function diagnosticSnapshot() {
52	70	return typeof value === 'number' ? scalar(value - now) : null
53	71	}
54	72	return {
55		- available: true, pendingChecks: scalar(owner.pendingChecks),
56		- fixture: { startRequests, admissionRequests, streamStarts, holdArmed, held, holdUsed, holdWorkspace, workspaceHeld }, lastStart, lastAdmission, responses: responses.slice(),
	73	+ available: true, pendingChecks: scalar(owner.pendingChecks), workerCallbacks: callbackObserver.snapshot(),
	74	+ fixture: { startRequests, admissionRequests, streamStarts, holdArmed, held, holdUsed, holdWorkspace, workspaceHeld, preAdmissionRenewalArmed, preAdmissionRenewed, preAdmissionHeldReads: preAdmissionReads.size, preAdmissionHoldExpired }, lastStart, lastAdmission, responses: responses.slice(),
57	75	slot: slot === null ? null : {
58	76	ordinal: ordinal(slot),
59	77	phase: choice(field(slot, 'phase'), ['waiting-for-consumer', 'running', 'stopping', 'completed', 'canceled', 'failed', 'discarded'] satisfies ExecutionView['phase'][]),
		⋯42 linesShow the 42 lines above this hunk@@ -102,10 +120,19 @@ assert(!html.includes('<IncidentWorkspace'))
102	120	assert(!html.includes(fixture.senderToken))
103	121	const secret = randomUUID()
104	122	const sources = () => fixture.database.db.primary.selectFrom('public_safety_sources').selectAll().execute()
	123	+async function holdWorkspaceRenewal() {
	124	+ holdWorkspace = true
	125	+ assert(closeWorkspaceStream, 'The real workspace stream must exist before renewal')
	126	+ await closeWorkspaceStream()
	127	+ const until = Date.now() + 5000
	128	+ while (!workspaceHeld && Date.now() < until) await Bun.sleep(5)
	129	+ assert(workspaceHeld, 'Native workspace renewal must request a fresh authorized snapshot')
	130	+}
105	131	const control = async (request: Request) => {
106	132	if (request.headers.get('x-fixture-control') !== secret) return new Response(null, { status: 404 })
107	133	const action = new URL(request.url).pathname.slice('/__fixture/'.length)
108		- if (action === 'status') return Response.json({ held, streamStarts, startRequests, startResponseReady: !!releaseStart })
	134	+ if (action === 'status') return Response.json({ held, streamStarts, startRequests, startResponseReady: !!releaseStart, workspaceHeld, preAdmissionRenewed, preAdmissionHeldReads: preAdmissionReads.size, preAdmissionHoldExpired, admissionRequests })
	135	+ if (action === 'worker-callback-timings') return Response.json(callbackObserver.snapshot())
109	136	if (action === 'diagnostics') {
110	137	try {
111	138	const text = `${JSON.stringify(diagnosticSnapshot())}\n`
		⋯22 linesShow the 22 lines above this hunk@@ -134,21 +161,34 @@ const control = async (request: Request) => {
134	161	}
135	162	if (action === 'release-and-renew') { releaseCredit?.(); await closeStream?.(); return Response.json({ ok: true }) }
136	163	if (action === 'before-admission') { assert.equal((await sources()).length, 1, 'Completion does not write a durable source'); return Response.json({ ok: true }) }
	164	+ if (action === 'hold-completion-read-renewal') {
	165	+ assert.equal(field(executionOwner().slot, 'phase'), 'completed')
	166	+ assert.equal(admissionRequests, 0)
	167	+ await holdWorkspaceRenewal()
	168	+ return Response.json({ ok: true })
	169	+ }
	170	+ if (action === 'arm-pre-admission-renewal') {
	171	+ assert(!preAdmissionRenewalArmed && !preAdmissionRenewed && admissionRequests === 0 && closeWorkspaceStream, 'Arm one renewal before any admission POST')
	172	+ preAdmissionRenewalArmed = true
	173	+ return Response.json({ ok: true })
	174	+ }
	175	+ if (action === 'renew-pre-admission-held-read') {
	176	+ assert(preAdmissionRenewalArmed && !preAdmissionRenewed && !preAdmissionHoldExpired && preAdmissionReads.size > 0 && admissionRequests === 0, 'Renew only while an explicit decision is waiting on held native reads')
	177	+ assert(closeWorkspaceStream, 'The real workspace stream must exist before admission')
	178	+ preAdmissionRenewed = true
	179	+ await closeWorkspaceStream()
	180	+ return Response.json({ ok: true })
	181	+ }
137	182	if (action === 'verify-admitted') {
138	183	const rows = await sources()
139	184	assert.equal(rows.length, 2)
140	185	assert(rows.some(row => row.digest !== seeded.source.digest))
141	186	const committed = await fixture.snapshot(seeded.key)
142	187	assert.equal(committed.audits.filter(audit => audit.action === 'source-registered').length, 2)
143		- holdWorkspace = true
144		- assert(closeWorkspaceStream, 'The real workspace stream must exist before renewal')
145		- await closeWorkspaceStream()
146		- const until = Date.now() + 5000
147		- while (!workspaceHeld && Date.now() < until) await Bun.sleep(5)
148		- assert(workspaceHeld, 'Native workspace renewal must request a fresh authorized snapshot')
	188	+ await holdWorkspaceRenewal()
149	189	return Response.json({ ok: true })
150	190	}
151		- if (action === 'release-workspace') { assert(workspaceHeld, 'Release only the held native renewal'); releaseWorkspace?.(); return Response.json({ ok: true }) }
	191	+ if (action === 'release-workspace') { assert(workspaceHeld || preAdmissionReads.size > 0, 'Release only held native reads'); releaseWorkspace?.(); releasePreAdmissionReads(); return Response.json({ ok: true }) }
152	192	if (action === 'read-only') return Response.json({ token: await fixture.credential(fixture.sender.id, ['public-safety:read']) })
153	193	if (action === 'revoke') { await fixture.tokens.revokeToken(fixture.senderToken); return Response.json({ ok: true }) }
154	194	return new Response(null, { status: 404 })
		⋯7 linesShow the 7 lines above this hunk@@ -162,6 +202,16 @@ route.use(async (request: Request, next: () => Promise<Response | null> | Respon
162	202	if (admissionOrdinal !== null) lastAdmission = { ordinal: admissionOrdinal, settled: false, status: null, elapsedMs: null }
163	203	let status: number | null = null
164	204	try {
	205	+ if (kind === 'workspace-read' && preAdmissionRenewalArmed) {
	206	+ // Ordinary polls may arrive first. Keep authorization visible until the driver
	207	+ // observes a real pending admission decision, then explicitly renew the stream.
	208	+ assert(preAdmissionReads.size < 8, 'The bounded native read barrier must not overflow')
	209	+ await new Promise<void>(resolve => {
	210	+ const release = () => { preAdmissionReads.delete(release); resolve() }
	211	+ preAdmissionReads.add(release)
	212	+ if (!preAdmissionTimeout) preAdmissionTimeout = setTimeout(() => { preAdmissionHoldExpired = true; releasePreAdmissionReads() }, 5000)
	213	+ })
	214	+ }
165	215	if (kind === 'workspace-read' && holdWorkspace) {
166	216	holdWorkspace = false; workspaceHeld = true
167	217	await new Promise<void>(resolve => {
		⋯4 linesShow the 4 lines above this hunk@@ -172,6 +222,9 @@ route.use(async (request: Request, next: () => Promise<Response | null> | Respon
172	222	}
173	223	const response = await next()
174	224	status = response?.status ?? null
	225	+ // The genuine core Worker is already constructed. No constructor or frame is replaced.
	226	+ // Observation precedes the accepted response and the later native consumer attachment.
	227	+ if (startOrdinal !== null && status === 202) observeAcceptedWorker(startOrdinal)
175	228	if (delayStart && request.method === 'POST' && pathname.endsWith('/workspace/execution') && response?.ok) { delayStart = false; await new Promise<void>(resolve => { releaseStart = resolve }) }
176	229	const executionEvents = /\/workspace\/execution\/[a-f0-9-]+\/events$/.test(pathname), workspaceEvents = pathname.endsWith('/workspace/events')
177	230	if (request.method !== 'GET' || !(executionEvents || workspaceEvents) || !response?.ok || !response.body) return response
		⋯29 linesShow the 29 lines above this hunk@@ -207,7 +260,7 @@ route.use(async (request: Request, next: () => Promise<Response | null> | Respon
207	260	})
208	261	const page = () => new Response(html, { headers: { 'content-type': 'text/html', 'cache-control': 'no-store' } })
209	262	route.get('/incidents', page)
210		-for (const action of ['status', 'diagnostics', 'release-start', 'arm-before-return', 'before-return', 'release-and-renew', 'before-admission', 'verify-admitted', 'release-workspace', 'read-only', 'revoke']) route.get(`/__fixture/${action}`, control)
	263	+for (const action of ['status', 'diagnostics', 'worker-callback-timings', 'release-start', 'arm-before-return', 'before-return', 'release-and-renew', 'before-admission', 'hold-completion-read-renewal', 'arm-pre-admission-renewal', 'renew-pre-admission-held-read', 'verify-admitted', 'release-workspace', 'read-only', 'revoke']) route.get(`/__fixture/${action}`, control)
211	264	const server = await route.serve({ hostname: '127.0.0.1', port: 0, nativeRoutes: false })
212	265	if (process.env.EXECUTION_BROWSER_EVIDENCE) { await mkdir(process.env.EXECUTION_BROWSER_EVIDENCE, { recursive: true }); await writeFile(join(process.env.EXECUTION_BROWSER_EVIDENCE, 'compiled.html'), html) }
213	266	await writeFile(join(root, 'execution-credentials.json'), JSON.stringify({ url: `http://127.0.0.1:${server.port}`, key: seeded.key, sourceDigest: seeded.source.digest, token: fixture.senderToken, control: secret }), { mode: 0o600 })
		⋯2 linesShow the 2 lines above this hunk@@ -216,7 +269,7 @@ try {
216	269	assert.equal(await child.exited, 0, 'Isolated real execution browser driver must pass')
217	270	}
218	271	finally {
219		- releaseStart?.(); releaseCredit?.(); releaseWorkspace?.()
	272	+ releaseStart?.(); releaseCredit?.(); releaseWorkspace?.(); releasePreAdmissionReads()
220	273	await closeStream?.(); await closeWorkspaceStream?.(); await server.stop(true)
221	274	const slot = executionOwner().slot as { execution: string, binding: { scope: string }, producer: string, worker: Worker | null } | null
222	275	if (slot?.producer === 'pending') {