From 31fed2211ef049f0a542f84c2264c1a2ba093452 Mon Sep 17 00:00:00 2001 From: BL Date: Sat, 5 Sep 2026 14:03:36 +0900 Subject: [PATCH 1/4] docs: scope bounded source PCM observation --- ...bservation-for-existing-source-spectrum.md | 27 +++++++++++++++++++ 1 file changed, 27 insertions(+) create mode 100644 .github/ISSUE_SPECS/434-expose-bounded-read-only-pcm-observation-for-existing-source-spectrum.md diff --git a/.github/ISSUE_SPECS/434-expose-bounded-read-only-pcm-observation-for-existing-source-spectrum.md b/.github/ISSUE_SPECS/434-expose-bounded-read-only-pcm-observation-for-existing-source-spectrum.md new file mode 100644 index 000000000..2fdc3310e --- /dev/null +++ b/.github/ISSUE_SPECS/434-expose-bounded-read-only-pcm-observation-for-existing-source-spectrum.md @@ -0,0 +1,27 @@ +# Expose bounded read-only PCM observation for existing source spectrum + +## Required existing behavior and boundary + +The app already displays input-source spectrum using its2048-point FFT and a bounded pull from source rings. SDK#405 moved PCM layout authority out of adapter/app. App#101 must retain spectrum without copying private ring offsets or using Msb1RingReader, which changes audio consumer state. The SDK owns this codec-neutral read-only observation; app retains FFT/display and adapter will later map source IDs to observers. No post-effect or arbitrary graph taps. + +## Smallest API + +Export PcmSourceChunk {generation:bigint,startFrame:bigint,frames:number,endOfRegion:boolean,planes:readonly Float32Array[]} and Msb1RingObserver from @misofm/engine/browser. Constructor accepts an existing SharedArrayBuffer ring and uses the same SDK validation authority. Expose channels, frameCapacity, pull(consume:(chunk:PcmSourceChunk)=>void, maximumChunks?:number):number, counters():Msb1RingCounters, close():void. Add only already-present missing wire counters needed by existing app totals (underruns,drainBlocks,depth) to the shared counter snapshot authority; no new wire words. + +## Frozen bounded read contract + +- Independent observation cursor, no Atomics stores/adds or mutation of any playback shared word/PCM. Never advances read/write indices, attaches/detaches the ring, holds slots, or adds another producer. Observer runs on app/control thread only. +- Own one reusable planar scratch chunk and metadata object. Callback data is borrowed until return; frames bounds valid samples. After construction, pull allocates no PCM buffers/views and does bounded work. Default maximum min(capacity,32); explicit maximum integer1..32. Slow observers catch up; they never stall audio. +- Begin at oldest currently live chunk. Handle ring-index wrap, missed data, seek epoch/generation changes, stale/invalid/torn slots, mono/stereo and partial tail. On close release local ownership idempotently; subsequent pull returns0. +- Copy into scratch only after checking slot sequence/generation and current live interval. Recheck sequence/generation/live interval after copying and drop observations whose slot became reusable/overwritten. A sequence-only test is insufficient because writer may overwrite PCM before publishing new sequence. A deterministic during-copy mutation in the existing fixture must reject mixed-generation/reused data without changing shared state. This is best-effort visualization observation, not guaranteed lossless delivery. +- Snapshot counters from existing words through one SDK authority; do not claim multiword atomicity. No source document parser, session changes, ring/prelude ABI change, decoder/storage input, telemetry protocol or new framework. + +## Allowed paths and execution + +sdk/src/browser/pcm-ring.ts or one adjacent pcm-observer.ts using shared validation helpers; sdk/src/browser/index.ts; sdk/test/browser-pcm-evals.mjs; existing strict packed type/runtime smoke; sdk/README.md; this spec. No Rust, generated six artifacts, PCM worklet prelude, SDK boot/control or adapter/app edits. Begin isolated codex/dx-pcm-observation after SDK428 review PASS and its evidence checkpoint. Source baseline9ab79d13 contains reviewed405 and completed428 revision awaiting review. Astra medium implements; separate Astra medium reviewer checks. Root commits focused-green tranche before final evidence. + +## Evidence + +Existing focused SDK fixture proves representative shapes/tail, budget cap, independent observers, index wrap/overrun, seek/stale generation, deliberate slot reuse during copy, byte-identical shared buffer before/after read-only operations, reusable local planes and closed no-op. Use fixed expectations rather than a second layout implementation. Existing type/headless/generated/package gates and strict public consumer pass. Six generated engine artifacts and PCM prelude remain byte-identical. No extra browser matrix, benchmark or listening gate for this visualization-only control API. + +Matching issue misofm/engine#434. From acf305993d5370cdda45d877794f513aa87d00ff Mon Sep 17 00:00:00 2001 From: BL Date: Sat, 5 Sep 2026 14:41:38 +0900 Subject: [PATCH 2/4] feat(browser): expose bounded read-only PCM observation --- ...bservation-for-existing-source-spectrum.md | 6 + sdk/README.md | 21 +++ sdk/src/browser/pcm-ring.ts | 123 ++++++++++++++- sdk/test/browser-pcm-evals.mjs | 142 ++++++++++++++++++ sdk/test/package-tarball-smoke.mjs | 33 +++- 5 files changed, 322 insertions(+), 3 deletions(-) diff --git a/.github/ISSUE_SPECS/434-expose-bounded-read-only-pcm-observation-for-existing-source-spectrum.md b/.github/ISSUE_SPECS/434-expose-bounded-read-only-pcm-observation-for-existing-source-spectrum.md index 2fdc3310e..fd936e91c 100644 --- a/.github/ISSUE_SPECS/434-expose-bounded-read-only-pcm-observation-for-existing-source-spectrum.md +++ b/.github/ISSUE_SPECS/434-expose-bounded-read-only-pcm-observation-for-existing-source-spectrum.md @@ -25,3 +25,9 @@ sdk/src/browser/pcm-ring.ts or one adjacent pcm-observer.ts using shared validat Existing focused SDK fixture proves representative shapes/tail, budget cap, independent observers, index wrap/overrun, seek/stale generation, deliberate slot reuse during copy, byte-identical shared buffer before/after read-only operations, reusable local planes and closed no-op. Use fixed expectations rather than a second layout implementation. Existing type/headless/generated/package gates and strict public consumer pass. Six generated engine artifacts and PCM prelude remain byte-identical. No extra browser matrix, benchmark or listening gate for this visualization-only control API. Matching issue misofm/engine#434. + +## Attempt 1 focused checkpoint + +Astra implementation adds `Msb1RingObserver` beside the existing writer, sharing its private validator and existing browser export. Observer owns reusable scratch/metadata, bounds candidate attempts, catches up across index wrap and seek epochs (including equal low-word generation tags), and validates the live interval again after copying. The existing counter snapshot type now includes the three existing wire words. Close drops ring/scratch references and retains a final counter snapshot; reentrant pull is rejected to protect borrowed scratch. + +Focused evidence: `node --test sdk/test/browser-pcm-evals.mjs` PASS (8 tests, `/private/tmp/dx434-focused.log`); `sdk/node_modules/.bin/tsc --project sdk/tsconfig.json` PASS (`/private/tmp/dx434-typecheck.log`). Fixtures include mono/stereo partial tails, independent cursors, 32-attempt default, invalid-slot budget, wrap/overrun, stale/torn slots, full seek generation, constructor/view allocation traps, reusable identities and byte-for-byte shared-state preservation. Deterministic copy hooks advance READ_INDEX and reserve/zero a reused slot while its old sequence remains published, or seek during copy; both deliver zero callbacks. Existing packed smoke now exercises the public observer type and runtime. Full headless/generated/package evidence follows root's exact-path checkpoint; not yet claimed. diff --git a/sdk/README.md b/sdk/README.md index a6ca01788..beb5a3618 100644 --- a/sdk/README.md +++ b/sdk/README.md @@ -220,6 +220,27 @@ Call `await engine.close()` when the browser session is finished. It disposes th before closing its `AudioContext`, is safe to call repeatedly, and still closes the context if the host's MessagePort has already failed. +For input-source spectrum, `Msb1RingObserver` from `@misofm/engine/browser` reads an +existing feed ring without consuming audio or changing shared bytes: + +```ts +const observer = new Msb1RingObserver(feed.rings[0]); +observer.pull((chunk) => { + updateSourceSpectrum(chunk.planes, chunk.frames); // Use synchronously; scratch is reused. +}); +const { underruns, drainBlocks, depth } = observer.counters(); +observer.close(); +``` + +Each observer starts at the oldest live chunk and has an independent cursor. Pull attempts at +most `min(ringCapacity, 32)` chunks by default, or an explicit integer from 1 to 32. Metadata and +planar scratch are borrowed until the callback returns; only `frames` samples are valid. Pull +creates no PCM buffers or views. It skips stale or concurrently reused slots and catches up after +missed data or seeks; visualization is best effort. Use it on the control thread and do not +reenter `pull` from its callback. Counters read existing wire words individually, not as an atomic +multiword snapshot. Closing releases local storage; later pulls return zero and counters retain +the final snapshot. Observation never holds slots or delays playback. + ## Agents ```ts diff --git a/sdk/src/browser/pcm-ring.ts b/sdk/src/browser/pcm-ring.ts index 1b907b1bf..14b3462cb 100644 --- a/sdk/src/browser/pcm-ring.ts +++ b/sdk/src/browser/pcm-ring.ts @@ -13,7 +13,7 @@ export const MSB1_CONTROL = Object.freeze({ MAGIC: 0, VERSION: 1, CAPACITY: 2, C const SLOT = { SEQUENCE: 0, GENERATION_TAG: 1, FRAMES: 2, FLAGS: 3 } as const; const SLOT_I64 = { GENERATION: 2, START_FRAME: 3 } as const; export interface Msb1RingLayout { readonly sourceId: string; readonly channels: number; readonly frameCapacity: number; readonly capacity: number } -export interface Msb1RingCounters { readonly wrote:number; readonly overflow:number; readonly submitted:number; readonly stale:number; readonly refused:number; readonly lastResult:number; readonly seeksApplied:number; readonly torn:number; readonly errors:number; readonly occupancy:number; readonly generationTag:number; readonly submittedGenerationTag:number } +export interface Msb1RingCounters { readonly wrote:number; readonly overflow:number; readonly submitted:number; readonly stale:number; readonly refused:number; readonly lastResult:number; readonly underruns:number; readonly drainBlocks:number; readonly depth:number; readonly seeksApplied:number; readonly torn:number; readonly errors:number; readonly occupancy:number; readonly generationTag:number; readonly submittedGenerationTag:number } export function msb1RingBytes(channels:number, frameCapacity:number, capacity:number):number { return MSB1_HEADER_OFFSET + capacity * MSB1_SLOT_HEADER_BYTES + capacity * channels * frameCapacity * 4; } export function createMsb1Ring(layout: Msb1RingLayout): SharedArrayBuffer { if (!power(layout.capacity)) throw new RangeError("MSB1 capacity must be a power of two"); @@ -34,3 +34,124 @@ export class Msb1RingWriter { } function bind(shared:SharedArrayBuffer){if(!(shared instanceof SharedArrayBuffer))throw new TypeError("MSB1 requires SharedArrayBuffer");const control=new Int32Array(shared,0,32);if(Atomics.load(control,0)!==MSB1_MAGIC||control[1]!==1)throw new TypeError("Shared buffer is not MSB1");const capacity=control[2]!,channels=control[3]!,frameCapacity=control[4]!,headerOffset=control[5]!,pcmOffset=control[6]!;if(!power(capacity)||!positive(channels)||!positive(frameCapacity))throw new TypeError("MSB1 header is invalid");const headers=new Int32Array(shared,headerOffset,capacity*8),planes=Array.from({length:capacity},(_,s)=>Array.from({length:channels},(_,ch)=>new Float32Array(shared,pcmOffset+(s*channels+ch)*frameCapacity*4,frameCapacity)));return {control,controlI64:new BigInt64Array(shared,MSB1_CONTROL_I64_OFFSET,2),headers,headersI64:new BigInt64Array(shared,headerOffset,capacity*4),planes,capacity,channels,frameCapacity}} function occupancy(c:Int32Array){return (Atomics.load(c,8)-Atomics.load(c,9))&(MSB1_WRAP-1)} function positive(v:number){return Number.isSafeInteger(v)&&v>0} function power(v:number){return positive(v)&&(v&(v-1))===0} + +/** Borrowed scratch storage: only `frames` samples are valid, until the callback returns. */ +export interface PcmSourceChunk { + readonly generation: bigint; + readonly startFrame: bigint; + readonly frames: number; + readonly endOfRegion: boolean; + readonly planes: readonly Float32Array[]; +} + +/** Best-effort source observation on the control thread; never owns an audio consumer slot. */ +export class Msb1RingObserver { + readonly channels: number; + readonly frameCapacity: number; + #view: ReturnType | undefined; + #chunk: { -readonly [K in keyof PcmSourceChunk]: PcmSourceChunk[K] } | undefined; + #cursor: number | undefined; + #epoch = -1; + #tag = 0; + #pulling = false; + #closedCounters: Msb1RingCounters | undefined; + + constructor(ring: SharedArrayBuffer) { + const view = bind(ring); + this.#view = view; + this.channels = view.channels; + this.frameCapacity = view.frameCapacity; + this.#chunk = { + generation: 0n, startFrame: 0n, frames: 0, endOfRegion: false, + planes: Object.freeze(Array.from({ length: view.channels }, () => new Float32Array(view.frameCapacity))), + }; + } + + pull(consume: (chunk: PcmSourceChunk) => void, maximumChunks?: number): number { + const view = this.#view; + const chunk = this.#chunk; + if (!view || !chunk) return 0; + const budget = maximumChunks ?? Math.min(view.capacity, 32); + if (!Number.isInteger(budget) || budget < 1 || budget > 32) throw new RangeError("maximumChunks must be an integer from 1 to 32"); + if (this.#pulling) throw new TypeError("MSB1 observation cannot be reentered"); + this.#pulling = true; + let delivered = 0; + try { + const c = view.control, h = view.headers, h64 = view.headersI64; + const mask = MSB1_WRAP - 1; + // Bound attempts, including stale/torn candidates, rather than only successful callbacks. + for (let attempt = 0; attempt < budget && this.#view === view; attempt++) { + const epoch = Atomics.load(c, MSB1_CONTROL.SEEK_EPOCH); + const tag = Atomics.load(c, MSB1_CONTROL.GENERATION_TAG); + const read = Atomics.load(c, MSB1_CONTROL.READ_INDEX); + const write = Atomics.load(c, MSB1_CONTROL.WRITE_INDEX); + const live = (write - read) & mask; + if (live > view.capacity) break; + if (this.#epoch !== epoch || this.#tag !== tag || this.#cursor === undefined || ((this.#cursor - read) & mask) > live) { + this.#cursor = read; + this.#epoch = epoch; + this.#tag = tag; + } + const index = this.#cursor; + if (index === write) break; + this.#cursor = (index + 1) & mask; + const slot = index & (view.capacity - 1), word = slot * 8, wide = slot * 4; + if (Atomics.load(h, word + SLOT.SEQUENCE) !== index || Atomics.load(h, word + SLOT.GENERATION_TAG) !== tag) continue; + const frames = Atomics.load(h, word + SLOT.FRAMES); + if (frames < 1 || frames > view.frameCapacity) continue; + const flags = Atomics.load(h, word + SLOT.FLAGS); + const generation = Atomics.load(h64, wide + SLOT_I64.GENERATION); + const startFrame = Atomics.load(h64, wide + SLOT_I64.START_FRAME); + if (Number(BigInt.asIntN(32, generation)) !== tag) continue; + // A seek can retain the same low-word tag; its full generation still excludes old slots. + const seekGeneration = Atomics.load(view.controlI64, 0); + if (epoch !== 0 && Number(BigInt.asIntN(32, seekGeneration)) === tag && generation !== seekGeneration) continue; + for (let channel = 0; channel < this.channels; channel++) chunk.planes[channel]!.set(view.planes[slot]![channel]!); + // A producer clears/reuses PCM before publishing a new sequence. Sequence alone cannot + // detect that race: the candidate must STILL belong to the audio consumer's live interval. + const afterRead = Atomics.load(c, MSB1_CONTROL.READ_INDEX); + const afterWrite = Atomics.load(c, MSB1_CONTROL.WRITE_INDEX); + const afterLive = (afterWrite - afterRead) & mask; + if (afterLive > view.capacity || ((index - afterRead) & mask) >= afterLive || + Atomics.load(c, MSB1_CONTROL.SEEK_EPOCH) !== epoch || Atomics.load(c, MSB1_CONTROL.GENERATION_TAG) !== tag || + Atomics.load(h, word + SLOT.SEQUENCE) !== index || Atomics.load(h, word + SLOT.GENERATION_TAG) !== tag || + Atomics.load(h, word + SLOT.FRAMES) !== frames || Atomics.load(h, word + SLOT.FLAGS) !== flags || + Atomics.load(h64, wide + SLOT_I64.GENERATION) !== generation || Atomics.load(h64, wide + SLOT_I64.START_FRAME) !== startFrame) continue; + chunk.generation = generation; + chunk.startFrame = startFrame; + chunk.frames = frames; + chunk.endOfRegion = (flags & MSB1_FLAG_END_OF_REGION) !== 0; + consume(chunk); + delivered++; + } + return delivered; + } finally { + this.#pulling = false; + } + } + + /** Individual atomic word reads, not a multiword atomic snapshot. After close, returns the final snapshot. */ + counters(): Msb1RingCounters { + return this.#view ? ringCounters(this.#view.control) : this.#closedCounters!; + } + + close(): void { + if (!this.#view) return; + this.#closedCounters = ringCounters(this.#view.control); + this.#view = undefined; + this.#chunk = undefined; + } +} + +function ringCounters(c: Int32Array): Msb1RingCounters { + const submittedGenerationTag = Atomics.load(c, MSB1_CONTROL.SUBMITTED_GENERATION_TAG); + return { + wrote: Atomics.load(c, MSB1_CONTROL.WROTE), overflow: Atomics.load(c, MSB1_CONTROL.OVERFLOW), + submitted: Atomics.load(c, MSB1_CONTROL.SUBMITTED), stale: Atomics.load(c, MSB1_CONTROL.STALE), + refused: Atomics.load(c, MSB1_CONTROL.REFUSED), lastResult: Atomics.load(c, MSB1_CONTROL.LAST_RESULT), + underruns: Atomics.load(c, MSB1_CONTROL.UNDERRUNS), drainBlocks: Atomics.load(c, MSB1_CONTROL.DRAIN_BLOCKS), + depth: Atomics.load(c, MSB1_CONTROL.DEPTH), seeksApplied: Atomics.load(c, MSB1_CONTROL.SEEKS_APPLIED), + torn: Atomics.load(c, MSB1_CONTROL.TORN), errors: Atomics.load(c, MSB1_CONTROL.ERRORS), + occupancy: occupancy(c), generationTag: Atomics.load(c, MSB1_CONTROL.GENERATION_TAG), submittedGenerationTag, + }; +} diff --git a/sdk/test/browser-pcm-evals.mjs b/sdk/test/browser-pcm-evals.mjs index 9128db4d7..03150d5ac 100644 --- a/sdk/test/browser-pcm-evals.mjs +++ b/sdk/test/browser-pcm-evals.mjs @@ -13,6 +13,8 @@ import { createMsb1Ring, msb1RingBytes, Msb1RingWriter, + Msb1RingObserver, + MSB1_WRAP, } from "../src/browser/pcm-ring.ts"; import { attachEngineFeed, PcmFeedError, prepareEngineFeed } from "../src/browser/pcm-feed.ts"; import { BUNDLED_ENGINE_ASSETS } from "../src/assets.ts"; @@ -370,3 +372,143 @@ test("moved prelude drains odd mono/stereo rings and allocation mutation turns r mutant.process(); assert.ok(mutant.allocations.includes("Float32Array"), "runtime mutation must be caught by constructor instrumentation"); }); + +function observedFixture(channels = 2, capacity = 4, frameCapacity = 4) { + const ring = createMsb1Ring({ sourceId: "observation", channels, capacity, frameCapacity }); + const writer = new Msb1RingWriter(ring); + writer.engage(1n); + return { ring, writer, c: controls(ring), write(startFrame, frames = frameCapacity, generation = 1n) { + const planes = writer.reserve(frames); + assert.ok(planes); + for (let ch = 0; ch < channels; ch++) for (let f = 0; f < frames; f++) planes[ch][f] = Number(startFrame) + ch * 100 + f; + writer.commit({ startFrame: BigInt(startFrame), frames, generation, endOfRegion: frames < frameCapacity }); + } }; +} + +test("observer borrows reusable mono/stereo scratch without altering any shared byte", (t) => { + for (const channels of [1, 2]) { + const f = observedFixture(channels); + f.write(10); f.write(20, 2); + for (const [word, value] of [[MSB1_CONTROL.UNDERRUNS, 7], [MSB1_CONTROL.DRAIN_BLOCKS, 8], [MSB1_CONTROL.DEPTH, 9]]) Atomics.store(f.c, word, value); + const before = Buffer.from(new Uint8Array(f.ring)); + const observer = new Msb1RingObserver(f.ring), independent = new Msb1RingObserver(f.ring); + assert.equal(observer.channels, channels); assert.equal(observer.frameCapacity, 4); + let metadata, planes, calls = 0; + const native = globalThis.Float32Array; + const constructor = t.mock.method(globalThis, "Float32Array", new Proxy(native, { construct() { throw new Error("pull allocated PCM"); } })); + const subarray = t.mock.method(native.prototype, "subarray", () => { throw new Error("pull allocated a view"); }); + const slice = t.mock.method(native.prototype, "slice", () => { throw new Error("pull allocated PCM"); }); + try { + assert.equal(observer.pull((chunk) => { + if (!metadata) { metadata = chunk; planes = [...chunk.planes]; } + else assert.equal(chunk, metadata); + const start = calls++ === 0 ? 10 : 20; + assert.equal(chunk.generation, 1n); assert.equal(chunk.startFrame, BigInt(start)); + assert.equal(chunk.frames, start === 10 ? 4 : 2); assert.equal(chunk.endOfRegion, start === 20); + chunk.planes.forEach((plane, ch) => { + assert.equal(plane, planes[ch]); assert.notEqual(plane.buffer, f.ring); + assert.deepEqual([...plane], start === 10 ? [10 + ch * 100, 11 + ch * 100, 12 + ch * 100, 13 + ch * 100] : [20 + ch * 100, 21 + ch * 100, 0, 0]); + }); + }, 1), 1); + assert.equal(observer.pull((chunk) => { + assert.equal(chunk, metadata); assert.equal(chunk.planes[0], planes[0]); + assert.equal(chunk.frames, 2); assert.equal(chunk.endOfRegion, true); + assert.deepEqual([...chunk.planes[0]], [20, 21, 0, 0]); + }), 1); + } finally { constructor.mock.restore(); subarray.mock.restore(); slice.mock.restore(); } + assert.equal(independent.pull(() => {}), 2); + assert.equal(observer.pull(() => assert.fail("already observed")), 0); + assert.deepEqual(observer.counters(), { wrote: 2, overflow: 0, submitted: 0, stale: 0, refused: 0, lastResult: 0, underruns: 7, drainBlocks: 8, depth: 9, seeksApplied: 0, torn: 0, errors: 0, occupancy: 2, generationTag: 1, submittedGenerationTag: 0 }); + const finalCounters = observer.counters(); + observer.close(); observer.close(); independent.close(); + assert.equal(observer.pull(() => assert.fail("closed")), 0); + assert.deepEqual(observer.counters(), finalCounters); + assert.deepEqual(Buffer.from(new Uint8Array(f.ring)), before); + } + assert.throws(() => new Msb1RingObserver(new ArrayBuffer(128)), /SharedArrayBuffer/); + const bad = createMsb1Ring({ sourceId: "bad", channels: 1, capacity: 2, frameCapacity: 4 }); + controls(bad)[MSB1_CONTROL.MAGIC] = 0; + assert.throws(() => new Msb1RingObserver(bad), /MSB1/); +}); + +test("observer bounds candidate work, catches up, wraps and remains independent of audio consumption", () => { + const f = observedFixture(1, 64, 1); + for (let i = 0; i < 40; i++) f.write(i); + const observer = new Msb1RingObserver(f.ring); + for (const invalid of [0, -1, 33, 1.5, NaN, Infinity]) assert.throws(() => observer.pull(() => {}, invalid), /maximumChunks/); + const starts = []; + assert.equal(observer.pull((chunk) => starts.push(chunk.startFrame)), 32); + assert.equal(observer.pull((chunk) => starts.push(chunk.startFrame)), 8); + assert.deepEqual(starts, Array.from({ length: 40 }, (_, i) => BigInt(i))); + assert.equal(f.c[MSB1_CONTROL.READ_INDEX], 0); + const wrapped = observedFixture(1, 2, 1); + Atomics.store(wrapped.c, MSB1_CONTROL.READ_INDEX, MSB1_WRAP - 1); + Atomics.store(wrapped.c, MSB1_CONTROL.WRITE_INDEX, MSB1_WRAP - 1); + wrapped.write(1); wrapped.write(2); + const slow = new Msb1RingObserver(wrapped.ring); + assert.equal(slow.pull((chunk) => assert.equal(chunk.startFrame, 1n), 1), 1); + Atomics.store(wrapped.c, MSB1_CONTROL.READ_INDEX, 1); + wrapped.write(3); wrapped.write(4); + const caught = []; + assert.equal(slow.pull((chunk) => caught.push(chunk.startFrame)), 2); + assert.deepEqual(caught, [3n, 4n]); + const fresh = new Msb1RingObserver(wrapped.ring); + assert.equal(fresh.pull((chunk) => { assert.equal(chunk.startFrame, 3n); fresh.close(); }), 1); + assert.equal(fresh.pull(() => assert.fail("closed in callback")), 0); + const bounded = new Msb1RingObserver(f.ring); + headers(f.ring, 64)[2] = 0; // Invalid first candidate still consumes the explicit budget. + assert.equal(bounded.pull(() => assert.fail("invalid candidate"), 1), 0); + assert.equal(bounded.pull((chunk) => assert.equal(chunk.startFrame, 1n), 1), 1); +}); + +test("observer rejects stale, invalid and torn slots and follows full seek generations", () => { + const f = observedFixture(1, 8, 1); + for (let i = 0; i < 5; i++) f.write(i); + const h = headers(f.ring, 8); + h[1] = 2; // Stale tag. + h[8] = 0; // Torn sequence. + h[18] = 2; // Frames exceed capacity. + headers64(f.ring, 8)[14] = 2n; // Full generation disagrees with tag. + const observer = new Msb1RingObserver(f.ring); + assert.equal(observer.pull((chunk) => assert.equal(chunk.startFrame, 4n)), 1); + const generation = (1n << 32n) + 1n; // Same low tag, new seek epoch. + f.writer.seek(generation, 5n); + f.write(5, 1, generation); + assert.equal(observer.pull((chunk) => { assert.equal(chunk.generation, generation); assert.equal(chunk.startFrame, 5n); }), 1); + const signed = 0x80000001n; + f.writer.seek(signed, 6n); f.write(6, 1, signed); + assert.equal(observer.pull((chunk) => { assert.equal(chunk.generation, signed); assert.equal(chunk.startFrame, 6n); }), 1); + assert.throws(() => new Msb1RingObserver(f.ring).pull(() => { observer.pull(() => {}); throw new Error("consumer failure"); }), /consumer failure/); + const reentrant = new Msb1RingObserver(f.ring); + assert.equal(reentrant.pull(() => assert.throws(() => reentrant.pull(() => {}), /reentered/)), 1); +}); + +test("observer drops PCM reused during copy before sequence publication, and seeks during copy", (t) => { + for (const mode of ["reuse", "seek"]) { + const f = observedFixture(2, 1, 4); + f.write(10); + const observer = new Msb1RingObserver(f.ring); + const originalSet = Float32Array.prototype.set; + let mutated = false, afterMutation; + const hook = t.mock.method(Float32Array.prototype, "set", function (source, offset) { + originalSet.call(this, source, offset); + if (!mutated && source.buffer === f.ring && this.buffer !== f.ring) { + mutated = true; + if (mode === "reuse") { + Atomics.store(f.c, MSB1_CONTROL.READ_INDEX, 1); + assert.ok(f.writer.reserve(4)); // Zeroes both shared planes WITHOUT publishing a new sequence. + assert.equal(headers(f.ring, 1)[0], 0); + } else f.writer.seek(2n, 20n); + afterMutation = Buffer.from(new Uint8Array(f.ring)); + } + }); + try { assert.equal(observer.pull(() => assert.fail("delivered raced PCM")), 0); } + finally { hook.mock.restore(); } + assert.equal(mutated, true); + assert.deepEqual(Buffer.from(new Uint8Array(f.ring)), afterMutation); + if (mode === "reuse") { + f.writer.commit({ generation: 1n, startFrame: 20n, frames: 4, endOfRegion: false }); + assert.equal(observer.pull((chunk) => { assert.equal(chunk.startFrame, 20n); assert.deepEqual([...chunk.planes[0]], [0, 0, 0, 0]); }), 1); + } + } +}); diff --git a/sdk/test/package-tarball-smoke.mjs b/sdk/test/package-tarball-smoke.mjs index 11e58912a..c6c7a9c51 100644 --- a/sdk/test/package-tarball-smoke.mjs +++ b/sdk/test/package-tarball-smoke.mjs @@ -108,8 +108,8 @@ const consumer = resolve(consumerRoot, "index.ts"); await writeFile(consumer, ` import { CATALOG, session } from "@misofm/engine"; import { createOfflineEngine, loadBundledEngineAsset } from "@misofm/engine/headless"; -import { createEngine, prepareEngineFeed, attachEngineFeed, Msb1RingWriter } from "@misofm/engine/browser"; -import type { BrowserEngine } from "@misofm/engine/browser"; +import { createEngine, prepareEngineFeed, attachEngineFeed, Msb1RingWriter, Msb1RingObserver } from "@misofm/engine/browser"; +import type { BrowserEngine, PcmSourceChunk } from "@misofm/engine/browser"; import { BUNDLED_ENGINE_ASSETS } from "@misofm/engine/assets"; // @ts-expect-error arbitrary-model canonical serialization is intentionally not public import { canonicalSessionJson } from "@misofm/engine"; @@ -128,6 +128,18 @@ void packedFeed.rings; const packedWriter = new Msb1RingWriter(packedFeed.rings[0]); packedWriter.engage(1n); void packedWriter; +const packedObserver = new Msb1RingObserver(packedFeed.rings[0]); +packedObserver.pull((chunk: PcmSourceChunk) => { + const generation: bigint = chunk.generation; + const planes: readonly Float32Array[] = chunk.planes; + const frames: number = chunk.frames; + // @ts-expect-error Borrowed metadata is read-only. + chunk.frames = 3; + void [generation, planes, frames]; +}, 1); +const observedCounters: number[] = [packedObserver.counters().underruns, packedObserver.counters().drainBlocks, packedObserver.counters().depth]; +void observedCounters; +packedObserver.close(); void prepareEngineFeed(domContext, BUNDLED_ENGINE_ASSETS.pcmFeedWorklet); async function defaultBrowserContext() { const engine = await createEngine({ document: "opaque" }); @@ -198,6 +210,23 @@ assert.deepEqual(packedAttach.rings, packedFeedRuntime.rings); assert.deepEqual(packedFeedRuntime.rings.map((ring) => new Int32Array(ring)[2]), [64, 64]); for (const ring of packedFeedRuntime.rings) Atomics.store(new Int32Array(ring), 13, 1); await packedFeedRuntime.ready(); +const publicWriter = new imported["./browser"].Msb1RingWriter(packedFeedRuntime.rings[0]); +publicWriter.engage(1n); +publicWriter.reserve(3)[0].set([1, 2, 3]); +publicWriter.commit({ generation: 1n, startFrame: 11n, frames: 3, endOfRegion: true }); +const observationBefore = Buffer.from(new Uint8Array(packedFeedRuntime.rings[0])); +const publicObserver = new imported["./browser"].Msb1RingObserver(packedFeedRuntime.rings[0]); +assert.equal(publicObserver.pull((chunk) => { + assert.equal(chunk.generation, 1n); + assert.equal(chunk.startFrame, 11n); + assert.equal(chunk.frames, 3); + assert.equal(chunk.endOfRegion, true); + assert.deepEqual([...chunk.planes[0]], [1, 2, 3, 0]); +}), 1); +assert.deepEqual([publicObserver.counters().underruns, publicObserver.counters().drainBlocks, publicObserver.counters().depth], [0, 0, 0]); +publicObserver.close(); +assert.equal(publicObserver.pull(() => assert.fail("closed observer")), 0); +assert.deepEqual(Buffer.from(new Uint8Array(packedFeedRuntime.rings[0])), observationBefore); packedFeedRuntime.close(); packedFeedRuntime.close(); assert.equal(packedDisconnects, 1); From 7e1713ec78ccb1135fa9f5b92f6b948520068ee6 Mon Sep 17 00:00:00 2001 From: BL Date: Sat, 5 Sep 2026 14:44:47 +0900 Subject: [PATCH 3/4] docs: record PCM observation package validation --- ...-only-pcm-observation-for-existing-source-spectrum.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/.github/ISSUE_SPECS/434-expose-bounded-read-only-pcm-observation-for-existing-source-spectrum.md b/.github/ISSUE_SPECS/434-expose-bounded-read-only-pcm-observation-for-existing-source-spectrum.md index fd936e91c..2b22de01e 100644 --- a/.github/ISSUE_SPECS/434-expose-bounded-read-only-pcm-observation-for-existing-source-spectrum.md +++ b/.github/ISSUE_SPECS/434-expose-bounded-read-only-pcm-observation-for-existing-source-spectrum.md @@ -31,3 +31,12 @@ Matching issue misofm/engine#434. Astra implementation adds `Msb1RingObserver` beside the existing writer, sharing its private validator and existing browser export. Observer owns reusable scratch/metadata, bounds candidate attempts, catches up across index wrap and seek epochs (including equal low-word generation tags), and validates the live interval again after copying. The existing counter snapshot type now includes the three existing wire words. Close drops ring/scratch references and retains a final counter snapshot; reentrant pull is rejected to protect borrowed scratch. Focused evidence: `node --test sdk/test/browser-pcm-evals.mjs` PASS (8 tests, `/private/tmp/dx434-focused.log`); `sdk/node_modules/.bin/tsc --project sdk/tsconfig.json` PASS (`/private/tmp/dx434-typecheck.log`). Fixtures include mono/stereo partial tails, independent cursors, 32-attempt default, invalid-slot budget, wrap/overrun, stale/torn slots, full seek generation, constructor/view allocation traps, reusable identities and byte-for-byte shared-state preservation. Deterministic copy hooks advance READ_INDEX and reserve/zero a reused slot while its old sequence remains published, or seek during copy; both deliver zero callbacks. Existing packed smoke now exercises the public observer type and runtime. Full headless/generated/package evidence follows root's exact-path checkpoint; not yet claimed. + +## Attempt 1 final validation at acf30599 + +- `bash scripts/check-sdk-headless.sh /private/tmp/dx-393-current-artifacts`: PASS, 160 passed / 1 existing skip (`/private/tmp/dx434-headless.log`). +- `npm run check:generated` and `npm run check:assets` from `sdk/`: PASS (`/private/tmp/dx434-generated.log`, `/private/tmp/dx434-assets.log`). The package gate also runs the complete `scripts/check-sdk-generated.sh`, including generated banners and the host declaration mirror; PASS. +- `npm run check:package -- /private/tmp/dx-393-current-artifacts` from `sdk/`: PASS (`/private/tmp/dx434-package.log`), including 11 CLI tests, strict extracted-tarball consumer types and public observer runtime assertions. No extra browser matrix was run for this control-thread observation API. +- Direct byte comparison: all six staged generated engine artifacts equal the approved input directory; the PCM prelude equals both baseline `e3a52dce` and its staged copy (`/private/tmp/dx434-bytes.log`). No Wasm rebuild or repin. Production/source paths remained clean throughout full validation. + +Implementation and required evidence are ready for the separate independent Astra review; this record does not claim review PASS or issue closure. From 8a19a84813230713e8f1604db04be4dccf653283 Mon Sep 17 00:00:00 2001 From: BL Date: Sat, 5 Sep 2026 14:49:35 +0900 Subject: [PATCH 4/4] docs: record independent PCM observation review --- ...-read-only-pcm-observation-for-existing-source-spectrum.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/.github/ISSUE_SPECS/434-expose-bounded-read-only-pcm-observation-for-existing-source-spectrum.md b/.github/ISSUE_SPECS/434-expose-bounded-read-only-pcm-observation-for-existing-source-spectrum.md index 2b22de01e..fb3bf05f8 100644 --- a/.github/ISSUE_SPECS/434-expose-bounded-read-only-pcm-observation-for-existing-source-spectrum.md +++ b/.github/ISSUE_SPECS/434-expose-bounded-read-only-pcm-observation-for-existing-source-spectrum.md @@ -40,3 +40,7 @@ Focused evidence: `node --test sdk/test/browser-pcm-evals.mjs` PASS (8 tests, `/ - Direct byte comparison: all six staged generated engine artifacts equal the approved input directory; the PCM prelude equals both baseline `e3a52dce` and its staged copy (`/private/tmp/dx434-bytes.log`). No Wasm rebuild or repin. Production/source paths remained clean throughout full validation. Implementation and required evidence are ready for the separate independent Astra review; this record does not claim review PASS or issue closure. + +## Dedicated Astra medium PASS + +Independent review at7e1713ec passes focused/type/package checks. Removing only the post-copy live-interval guard causes the existing slot-reuse regression to fail while sequence/generation guards remain; read-only race protection is discriminating. Six generated artifacts and PCM prelude are unchanged. Report attached to the observation PR. This completes SDK source observation, not adapter mapping or app adoption.