ReviewOS
Reload

also looking at this

stacks/drone

feat: add explicit VTOL crosswind heading plans

#90
Merged chris wants to merge codex/vtol-crosswind-capture into main
21 files +815 -36

Review threads live on the whole diff, not on one commit, so none are shown here - a thread's line means something in the branch's final form, and painting it into an intermediate step would put it on code it is not about.

.github/workflows/ci.ymlmodified+1-0
@@ -513,6 +513,7 @@ jobs:
513513 ${{ runner.temp }}/execution-browser-evidence/admission-renewal.json
514514 ${{ runner.temp }}/execution-browser-evidence/pre-admission-renewal.json
515515 ${{ runner.temp }}/execution-browser-evidence/completion-read-renewal.json
516 ${{ runner.temp }}/execution-browser-evidence/worker-callback-timings.json
516517 ${{ runner.temp }}/execution-browser-evidence/chrome-startup-stderr.log
517518 purge-namespace:
518519 runs-on: ubuntu-latest
tests/fixtures/public-safety-access/execution-browser.tsmodified+15-2
@@ -5,6 +5,7 @@ import { mkdir, writeFile } from 'node:fs/promises'
55import { join, resolve } from 'node:path'
66import { randomUUID } from 'node:crypto'
77import { setupHandoverFixture } from './handover-setup'
8import { createExecutionCallbackObserver } from '../../helpers/execution-worker-observation'
89
910// Real native setup owns the guarded disposable SQLite file before any application boot.
1011const fixture = await setupHandoverFixture()
@@ -43,6 +44,14 @@ let nextOrdinal = 0
4344const scalar = (value: unknown) => typeof value === 'number' && Number.isFinite(value) && Math.abs(value) <= Number.MAX_SAFE_INTEGER ? value : null
4445const field = (value: unknown, name: string): unknown => value && typeof value === 'object' ? Object.getOwnPropertyDescriptor(value, name)?.value : undefined
4546const choice = <T extends string>(value: unknown, values: readonly T[]): T | null => values.find(item => item === value) ?? null
47const callbackObserver = createExecutionCallbackObserver(value => console.log(JSON.stringify(value)))
48function 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}
4655const ordinal = (slot: object) => {
4756 const existing = ordinals.get(slot)
4857 if (existing !== undefined) return existing
@@ -61,7 +70,7 @@ function diagnosticSnapshot() {
6170 return typeof value === 'number' ? scalar(value - now) : null
6271 }
6372 return {
64 available: true, pendingChecks: scalar(owner.pendingChecks),
73 available: true, pendingChecks: scalar(owner.pendingChecks), workerCallbacks: callbackObserver.snapshot(),
6574 fixture: { startRequests, admissionRequests, streamStarts, holdArmed, held, holdUsed, holdWorkspace, workspaceHeld, preAdmissionRenewalArmed, preAdmissionRenewed, preAdmissionHeldReads: preAdmissionReads.size, preAdmissionHoldExpired }, lastStart, lastAdmission, responses: responses.slice(),
6675 slot: slot === null ? null : {
6776 ordinal: ordinal(slot),
@@ -123,6 +132,7 @@ const control = async (request: Request) => {
123132 if (request.headers.get('x-fixture-control') !== secret) return new Response(null, { status: 404 })
124133 const action = new URL(request.url).pathname.slice('/__fixture/'.length)
125134 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())
126136 if (action === 'diagnostics') {
127137 try {
128138 const text = `${JSON.stringify(diagnosticSnapshot())}\n`
@@ -212,6 +222,9 @@ route.use(async (request: Request, next: () => Promise<Response | null> | Respon
212222 }
213223 const response = await next()
214224 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)
215228 if (delayStart && request.method === 'POST' && pathname.endsWith('/workspace/execution') && response?.ok) { delayStart = false; await new Promise<void>(resolve => { releaseStart = resolve }) }
216229 const executionEvents = /\/workspace\/execution\/[a-f0-9-]+\/events$/.test(pathname), workspaceEvents = pathname.endsWith('/workspace/events')
217230 if (request.method !== 'GET' || !(executionEvents || workspaceEvents) || !response?.ok || !response.body) return response
@@ -247,7 +260,7 @@ route.use(async (request: Request, next: () => Promise<Response | null> | Respon
247260})
248261const page = () => new Response(html, { headers: { 'content-type': 'text/html', 'cache-control': 'no-store' } })
249262route.get('/incidents', page)
250for (const action of ['status', 'diagnostics', '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)
263for (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)
251264const server = await route.serve({ hostname: '127.0.0.1', port: 0, nativeRoutes: false })
252265if (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) }
253266await 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 })
tests/fixtures/public-safety-access/execution-observer-worker.tsadded+5-0
@@ -0,0 +1,5 @@
1self.onmessage = ({ data }) => {
2 if (data !== 'emit') return
3 self.onmessage = null
4 for (const kind of ['event', 'result', 'finished', 'unrecognized']) self.postMessage({ kind, payload: 'private-callback-payload', bytes: new Uint8Array([1, 2, 3]) })
5}
tests/helpers/execution-browser-driver.tsmodified+22-0
@@ -214,6 +214,28 @@ try {
214214 await control('revoke')
215215 await first.wait("document.querySelector('[data-workspace-status]')?.textContent.trim() === 'refused'", 'Revocation must clear private execution content', 16000)
216216 assert.equal(await first.evaluate("!!document.querySelector('[data-execution-id]') || document.body.textContent.includes(" + JSON.stringify(digest) + ")"), false)
217 const callbackTimings = await control('worker-callback-timings')
218 assert.equal(callbackTimings.overflow, false)
219 assert.equal(callbackTimings.unavailable, false)
220 assert.equal(callbackTimings.observations.length, 2, 'Both genuine core Workers must have bounded callback observations')
221 const nominalCallbacks = callbackTimings.observations[1]
222 assert.equal(nominalCallbacks.ordinal, 2)
223 assert.equal(nominalCallbacks.inCallback, false)
224 assert.equal(nominalCallbacks.capped, false)
225 assert(nominalCallbacks.costs.event.calls > 0)
226 assert.equal(nominalCallbacks.costs.result.calls, 1)
227 assert.equal(nominalCallbacks.costs.finished.calls, 1)
228 for (const observation of callbackTimings.observations) {
229 assert.equal(observation.inCallback, false)
230 assert.equal(observation.capped, false)
231 for (const kind of ['event', 'result', 'finished', 'other']) {
232 const bucket = observation.costs[kind]
233 assert(Number.isSafeInteger(bucket.calls) && bucket.calls >= 0 && bucket.calls <= 1000000)
234 assert(Number.isFinite(bucket.totalMs) && bucket.totalMs >= 0 && bucket.totalMs <= Number.MAX_SAFE_INTEGER)
235 assert(Number.isFinite(bucket.maxMs) && bucket.maxMs >= 0 && bucket.maxMs <= bucket.totalMs)
236 }
237 }
238 await writeFile(join(evidence, 'worker-callback-timings.json'), `${JSON.stringify(callbackTimings)}\n`, { mode: 0o600, flag: 'wx' })
217239 console.log('PASS actual execution browser: native live-before-return/GET renewal/stop/completion/explicit admission/read-only/revocation')
218240}
219241catch (error) {
tests/helpers/execution-worker-observation.tsadded+63-0
@@ -0,0 +1,63 @@
1type CallbackKind = 'event' | 'result' | 'finished' | 'other'
2interface CallbackCost { calls: number, totalMs: number, maxMs: number }
3interface CallbackObservation { ordinal: number, inCallback: boolean, capped: boolean, costs: Record<CallbackKind, CallbackCost> }
4type CallbackMarker = { evidence: 'native-worker-callback', ordinal: number, kind: 'result' | 'finished', phase: 'entered', sinceObservationMs: number | null }
5 | { evidence: 'native-worker-callback', ordinal: number, kind: 'result' | 'finished', phase: 'settled', callbackMs: number | null, sinceObservationMs: number | null }
6const scalar = (value: number) => Number.isFinite(value) && Math.abs(value) <= Number.MAX_SAFE_INTEGER ? value : null
7const field = (value: unknown, name: string): unknown => value && typeof value === 'object' ? Object.getOwnPropertyDescriptor(value, name)?.value : undefined
8const choice = <T extends string>(value: unknown, values: readonly T[]): T | null => values.find(item => item === value) ?? null
9
10/** Private fixture observation only. Measures synchronous callback work, never async completion. */
11export function createExecutionCallbackObserver(marker: (value: CallbackMarker) => void, now: () => number = () => performance.now()) {
12 const callbackObservations: CallbackObservation[] = [], observedWorkers = new WeakSet<Worker>()
13 let callbackObservationOverflow = false, callbackObservationUnavailable = false
14 const snapshot = () => ({ overflow: callbackObservationOverflow, unavailable: callbackObservationUnavailable, observations: structuredClone(callbackObservations) })
15 function observe(worker: Worker, acceptedOrdinal: number) {
16 try {
17 if (observedWorkers.has(worker)) return
18 if (callbackObservations.length >= 2) { callbackObservationOverflow = true; return }
19 const callback = worker.onmessage
20 if (typeof callback !== 'function') { callbackObservationUnavailable = true; return }
21 const cost = (): CallbackCost => ({ calls: 0, totalMs: 0, maxMs: 0 })
22 const observation: CallbackObservation = { ordinal: acceptedOrdinal, inCallback: false, capped: false, costs: { event: cost(), result: cost(), finished: cost(), other: cost() } }
23 const installedAt = now()
24 let markerCount = 0
25 const emitMarker = (value: CallbackMarker) => {
26 if (markerCount >= 4) { observation.capped = true; return }
27 markerCount++
28 marker(value)
29 }
30 worker.onmessage = function (event) {
31 let began: number | null = null, kind: CallbackKind = 'other'
32 try {
33 kind = choice(field(event.data, 'kind'), ['event', 'result', 'finished']) ?? 'other'
34 if (kind === 'result' || kind === 'finished') emitMarker({ evidence: 'native-worker-callback', ordinal: acceptedOrdinal, kind, phase: 'entered', sinceObservationMs: scalar(now() - installedAt) })
35 }
36 catch { callbackObservationUnavailable = true }
37 try { observation.inCallback = true; began = now() }
38 catch { callbackObservationUnavailable = true }
39 try { return callback.call(this, event) }
40 finally {
41 // Timing is observational: preserve the exact native callback outcome,
42 // including an exception, even if collection or its closed console marker fails.
43 try {
44 const ended = now()
45 observation.inCallback = false
46 const elapsed = began === null ? null : scalar(ended - began)
47 if (elapsed === null || elapsed < 0) callbackObservationUnavailable = true
48 else {
49 const bucket = observation.costs[kind]
50 if (bucket.calls >= 1000000 || bucket.totalMs + elapsed > Number.MAX_SAFE_INTEGER) observation.capped = true
51 else { bucket.calls++; bucket.totalMs += elapsed; bucket.maxMs = Math.max(bucket.maxMs, elapsed) }
52 }
53 if (kind === 'result' || kind === 'finished') emitMarker({ evidence: 'native-worker-callback', ordinal: acceptedOrdinal, kind, phase: 'settled', callbackMs: elapsed, sinceObservationMs: scalar(now() - installedAt) })
54 }
55 catch { observation.inCallback = false; callbackObservationUnavailable = true }
56 }
57 }
58 observedWorkers.add(worker); callbackObservations.push(observation)
59 }
60 catch { callbackObservationUnavailable = true }
61 }
62 return { observe, snapshot }
63}
tests/unit/public-safety-execution-callbacks.test.tsadded+127-0
@@ -0,0 +1,127 @@
1import { expect, test } from 'bun:test'
2import { createExecutionCallbackObserver } from '../helpers/execution-worker-observation'
3
4const worker = () => new Worker(new URL('../fixtures/public-safety-access/execution-observer-worker.ts', import.meta.url), { env: {}, preload: [] })
5
6test('real Worker frames and receiver survive private callback observation without persisting payloads', async () => {
7 const source = worker(), markers: unknown[] = []
8 const observer = createExecutionCallbackObserver(value => markers.push(value))
9 const received: string[] = []
10 let timer: ReturnType<typeof setTimeout> | undefined
11 try {
12 const delivered = new Promise<void>((resolve, reject) => {
13 timer = setTimeout(() => reject(new Error('Owned callback fixture did not deliver its frames')), 2000)
14 source.onmessage = function (event) {
15 try {
16 expect(this).toBe(source)
17 const frame = event.data as { kind: string, payload: string, bytes: Uint8Array }
18 expect(frame.payload).toBe('private-callback-payload')
19 expect(frame.bytes).toEqual(new Uint8Array([1, 2, 3]))
20 received.push(frame.kind)
21 if (received.length === 4) resolve()
22 }
23 catch (error) { reject(error) }
24 }
25 source.onerror = () => reject(new Error('Owned callback fixture Worker failed'))
26 })
27 observer.observe(source, 1)
28 source.postMessage('emit')
29 await delivered
30 expect(received).toEqual(['event', 'result', 'finished', 'unrecognized'])
31 const snapshot = observer.snapshot()
32 expect(snapshot.overflow).toBe(false)
33 expect(snapshot.unavailable).toBe(false)
34 expect(snapshot.observations).toHaveLength(1)
35 expect(snapshot.observations[0]!.inCallback).toBe(false)
36 expect(snapshot.observations[0]!.capped).toBe(false)
37 for (const bucket of Object.values(snapshot.observations[0]!.costs)) {
38 expect(bucket.calls).toBe(1)
39 expect(Number.isFinite(bucket.totalMs) && bucket.totalMs >= 0).toBe(true)
40 expect(bucket.maxMs).toBe(bucket.totalMs)
41 }
42 expect(markers).toHaveLength(4)
43 expect(JSON.stringify({ snapshot, markers })).not.toContain('private-callback-payload')
44 expect(JSON.stringify({ snapshot, markers })).not.toContain('bytes')
45 }
46 finally { clearTimeout(timer); source.terminate() }
47})
48
49test('observer failures preserve the original callback return, throw identity and opaque fields', () => {
50 const first = worker(), second = worker()
51 const observer = createExecutionCallbackObserver(() => { throw new Error('Diagnostic marker unavailable') })
52 let getterCalls = 0, received: unknown
53 const payload = { kind: 'result', get privateField() { getterCalls++; return 'unread-private-value' } }
54 const event = new MessageEvent('message', { data: payload }), result = { originalReturn: true }, originalError = new Error('Original callback failure')
55 try {
56 first.onmessage = function (value) { expect(this).toBe(first); received = value.data; return result }
57 observer.observe(first, 1)
58 expect(first.onmessage!.call(first, event)).toBe(result)
59 expect(received).toBe(payload)
60 expect(getterCalls).toBe(0)
61 second.onmessage = function (value) { expect(this).toBe(second); expect(value).toBe(event); throw originalError }
62 observer.observe(second, 2)
63 let thrown: unknown
64 try { second.onmessage!.call(second, event) }
65 catch (error) { thrown = error }
66 expect(thrown).toBe(originalError)
67 expect(observer.snapshot().unavailable).toBe(true)
68 expect(observer.snapshot().observations.map(value => value.costs.result.calls)).toEqual([1, 1])
69 expect(JSON.stringify(observer.snapshot())).not.toContain('unread-private-value')
70 expect(getterCalls).toBe(0)
71 }
72 finally { first.terminate(); second.terminate() }
73})
74
75test('the observer caps actual Worker reservations without changing later callbacks and returns detached snapshots', () => {
76 const first = worker(), second = worker(), third = worker()
77 const observer = createExecutionCallbackObserver(() => {})
78 const callback = () => {}
79 try {
80 first.onmessage = callback; second.onmessage = callback; third.onmessage = callback
81 observer.observe(first, 1); observer.observe(first, 99); observer.observe(second, 2); observer.observe(third, 3)
82 const snapshot = observer.snapshot()
83 expect(snapshot.overflow).toBe(true)
84 expect(snapshot.unavailable).toBe(false)
85 expect(snapshot.observations.map(value => value.ordinal)).toEqual([1, 2])
86 expect(third.onmessage).toBe(callback)
87 snapshot.observations[0]!.costs.event.calls = 9999
88 expect(observer.snapshot().observations[0]!.costs.event.calls).toBe(0)
89 }
90 finally { first.terminate(); second.terminate(); third.terminate() }
91})
92
93test('marker latency and diagnostic exceptions cannot inflate the original synchronous callback duration', () => {
94 const source = worker(), returned = { original: true }, settled: number[] = []
95 let clock = 0
96 const observer = createExecutionCallbackObserver(value => {
97 if (value.phase === 'settled' && value.callbackMs !== null) settled.push(value.callbackMs)
98 clock += 1000
99 throw new Error('Slow diagnostic sink failed')
100 }, () => clock)
101 try {
102 source.onmessage = () => { clock += 7; return returned }
103 observer.observe(source, 1)
104 expect(source.onmessage!.call(source, new MessageEvent('message', { data: { kind: 'result' } }))).toBe(returned)
105 expect(observer.snapshot().observations[0]!.costs.result).toEqual({ calls: 1, totalMs: 7, maxMs: 7 })
106 expect(settled).toEqual([7])
107 expect(observer.snapshot().unavailable).toBe(true)
108 }
109 finally { source.terminate() }
110})
111
112test('repeated terminal-kind messages cannot produce an unbounded diagnostic console stream', () => {
113 const source = worker(), markers: unknown[] = []
114 const observer = createExecutionCallbackObserver(value => markers.push(value))
115 let originalCalls = 0
116 try {
117 source.onmessage = () => { originalCalls++ }
118 observer.observe(source, 1)
119 for (let index = 0; index < 10; index++) source.onmessage!.call(source, new MessageEvent('message', { data: { kind: 'result' } }))
120 expect(originalCalls).toBe(10)
121 expect(markers).toHaveLength(4)
122 expect(observer.snapshot().observations[0]!.capped).toBe(true)
123 expect(observer.snapshot().observations[0]!.costs.result.calls).toBe(10)
124 expect(observer.snapshot().unavailable).toBe(false)
125 }
126 finally { source.terminate() }
127})