From 04ea2f2d2112e7b81d9094dd6e8b02936e8d26a8 Mon Sep 17 00:00:00 2001 From: CJ Pais Date: Mon, 24 Aug 2026 14:34:42 +0800 Subject: [PATCH] feat(bindings): add structured stream snapshot --- bindings/python/README.md | 1 + .../python/src/transcribe_cpp/__init__.py | 5 ++ bindings/python/tests/test_streaming.py | 8 +++ bindings/rust/transcribe-cpp/README.md | 12 +++++ .../rust/transcribe-cpp/tests/streaming.rs | 11 +++++ bindings/swift/README.md | 1 + .../swift/Sources/TranscribeCpp/Session.swift | 2 +- .../Sources/TranscribeCpp/Streaming.swift | 3 ++ .../TranscribeCppTests/StreamingTests.swift | 6 +++ bindings/typescript/README.md | 8 +-- bindings/typescript/src/index.ts | 49 ++++++++++--------- bindings/typescript/src/types.ts | 7 ++- bindings/typescript/test/streaming.test.mjs | 21 +++++++- docs/bindings.md | 16 +++++- 14 files changed, 119 insertions(+), 31 deletions(-) diff --git a/bindings/python/README.md b/bindings/python/README.md index 7b5fbbe1..1e57c9b8 100644 --- a/bindings/python/README.md +++ b/bindings/python/README.md @@ -40,6 +40,7 @@ with model.session() as session, session.stream() as stream: stream.feed(chunk) text = stream.text() # .committed (stable) + .tentative stream.finalize() + result = stream.snapshot() # language, segments, words, tokens, timings ``` Long transcriptions can be cancelled from another thread with diff --git a/bindings/python/src/transcribe_cpp/__init__.py b/bindings/python/src/transcribe_cpp/__init__.py index 9312976e..bc0a6602 100644 --- a/bindings/python/src/transcribe_cpp/__init__.py +++ b/bindings/python/src/transcribe_cpp/__init__.py @@ -1413,6 +1413,11 @@ def text(self) -> StreamText: tentative=_decode(txt.tentative_text), ) + def snapshot(self) -> Result: + """Full structured snapshot of the current hypothesis (owned copies).""" + _ = self._h # validate that this stream has not been reset + return self._session._materialize() + @property def state(self) -> str: """``"idle"`` / ``"active"`` / ``"finished"`` / ``"failed"``.""" diff --git a/bindings/python/tests/test_streaming.py b/bindings/python/tests/test_streaming.py index e6081844..6c169509 100644 --- a/bindings/python/tests/test_streaming.py +++ b/bindings/python/tests/test_streaming.py @@ -29,6 +29,7 @@ def test_streaming_real(streaming_model_path, audio_pcm): stream.feed(audio_pcm[i : i + 16000]) update = stream.finalize() committed = stream.text().committed + snapshot = stream.snapshot() revision, state = stream.revision, stream.state last_status = stream.last_status assert update.is_final, update @@ -36,6 +37,11 @@ def test_streaming_real(streaming_model_path, audio_pcm): assert revision >= 1 assert last_status is None, last_status assert "country" in committed.lower(), committed + assert snapshot.text + assert isinstance(snapshot.language, str) + assert snapshot.segments + assert isinstance(snapshot.words, tuple) + assert isinstance(snapshot.tokens, tuple) def test_streaming_with_language_hint(prompted_streaming_model_path, audio_pcm): @@ -122,6 +128,8 @@ def test_stream_use_after_reset_rejected(streaming_model_path, audio_pcm): stream.feed(audio_pcm[:16000]) with pytest.raises(t.TranscribeError, match="reset"): stream.text() + with pytest.raises(t.TranscribeError, match="reset"): + stream.snapshot() def test_stream_reset_idempotent_and_session_reusable( diff --git a/bindings/rust/transcribe-cpp/README.md b/bindings/rust/transcribe-cpp/README.md index 2720d429..6120e70e 100644 --- a/bindings/rust/transcribe-cpp/README.md +++ b/bindings/rust/transcribe-cpp/README.md @@ -36,6 +36,18 @@ println!("{}", result.text); # Ok::<(), transcribe_cpp::Error>(()) ``` +Streaming exposes both UI-stable text and a fully materialized structured +snapshot: + +```rust +let mut stream = session.stream(&RunOptions::default(), &Default::default())?; +stream.feed(&chunk)?; +println!("{}", stream.text().committed); +stream.finalize()?; +let transcript = stream.snapshot(); // language, segments, words, tokens, timings +# Ok::<(), transcribe_cpp::Error>(()) +``` + Runnable examples: ```sh diff --git a/bindings/rust/transcribe-cpp/tests/streaming.rs b/bindings/rust/transcribe-cpp/tests/streaming.rs index bbf16d64..c5b8d12e 100644 --- a/bindings/rust/transcribe-cpp/tests/streaming.rs +++ b/bindings/rust/transcribe-cpp/tests/streaming.rs @@ -62,6 +62,17 @@ fn streams_jfk_committed_text() { "stream text: {:?}", text.full ); + + let snapshot = stream.snapshot(); + assert!(!snapshot.text.is_empty(), "structured snapshot is empty"); + if let Some(language) = &snapshot.language { + assert!(!language.is_empty(), "detected language must not be empty"); + } + assert!( + !snapshot.segments.is_empty(), + "structured segments are empty" + ); + let _ = (&snapshot.words, &snapshot.tokens); } #[test] diff --git a/bindings/swift/README.md b/bindings/swift/README.md index 0b32a47f..32ac0590 100644 --- a/bindings/swift/README.md +++ b/bindings/swift/README.md @@ -69,6 +69,7 @@ for chunk in chunks { // 16 kHz mono float32 frames if update.committedChanged { print(stream.text.committed) } } try stream.finalize() +let transcript = stream.snapshot // language, segments, words, tokens, timings ``` Runnable examples live in diff --git a/bindings/swift/Sources/TranscribeCpp/Session.swift b/bindings/swift/Sources/TranscribeCpp/Session.swift index 1a58289b..ce4850c0 100644 --- a/bindings/swift/Sources/TranscribeCpp/Session.swift +++ b/bindings/swift/Sources/TranscribeCpp/Session.swift @@ -181,7 +181,7 @@ public final class Session { } } - private func readTranscript() -> Transcript { + func readTranscript() -> Transcript { var segments: [Segment] = [] for i in 0.. { +function materialize(n: Native, acc: Accessors): Transcript { const F = n.F; const segments: Segment[] = []; @@ -683,6 +681,7 @@ const STREAM_TEARDOWN = new WeakMap< interface SessionControl { enterCompute(kind: string): void; leaveCompute(kind: string): void; + currentCompute(): string | null; isCurrentStream(stream: Stream): boolean; replaceCurrentStream(stream: Stream): void; clearCurrentStream(stream: Stream): void; @@ -722,6 +721,7 @@ export class Session { leaveCompute: (kind) => { if (this.#inFlight === kind) this.#inFlight = null; }, + currentCompute: () => this.#inFlight, isCurrentStream: (stream) => this.#activeStream === stream, replaceCurrentStream: (stream) => { if (this.#activeStream && this.#activeStream !== stream) { @@ -1039,7 +1039,6 @@ export class Stream { #keepalive: unknown[] | null; #active = true; #stale = false; // true once the session has begun a newer native stream - #inFlight = false; // true while a feed/finalize native call runs on a worker #holdsLease = true; // born holding the model's compute lease (claimed at begin) /** @internal */ @@ -1108,9 +1107,8 @@ export class Stream { // The native feed runs on a libuv worker. While it is in flight the // session must not be touched from the main thread — the C session API // is single-threaded (transcribe.h), and stream_get_text hands back - // pointers the feed may free/realloc. Flag it so the read getters fail - // fast instead of racing into a use-after-free. - this.#inFlight = true; + // pointers the feed may free/realloc. Flag the owning session so every + // result getter fails fast instead of racing into a use-after-free. this.#sessionControl.enterCompute("feed()/finalize()"); try { const status = await callAsync( @@ -1128,7 +1126,6 @@ export class Stream { } check(n, status, "transcribe_stream_feed"); } finally { - this.#inFlight = false; this.#sessionControl.leaveCompute("feed()/finalize()"); } return toStreamUpdate(u); @@ -1144,7 +1141,6 @@ export class Stream { return this.#lock.run(async () => { const u: any = {}; n.F.streamUpdateInit(u); - this.#inFlight = true; // see feed(): worker-thread compute, no concurrent reads this.#sessionControl.enterCompute("feed()/finalize()"); try { check( @@ -1153,7 +1149,6 @@ export class Stream { "transcribe_stream_finalize", ); } finally { - this.#inFlight = false; this.#sessionControl.leaveCompute("feed()/finalize()"); // Finalize ends the active stream (FINISHED on success, FAILED on // error), so the model is free again — release the lease either way. @@ -1164,16 +1159,15 @@ export class Stream { } /** - * Reads borrow session-owned snapshot memory, so they are forbidden while a - * feed()/finalize() is computing on a worker thread (concurrent use of a - * single session is undefined per transcribe.h). The natural pattern — - * `await stream.feed(chunk)` then read — is unaffected; this only rejects a - * read issued against an un-awaited feed. + * Reads borrow session-owned snapshot memory, so they are forbidden while + * any worker call is computing on this session (concurrent use is undefined + * per transcribe.h). The natural await-then-read pattern is unaffected. */ - #assertNotFeeding(what: string): void { - if (this.#inFlight) { + #assertNotComputing(what: string): void { + const compute = this.#sessionControl.currentCompute(); + if (compute) { throw new TranscribeError( - `cannot read stream ${what} while a feed()/finalize() is in flight; await it first`, + `cannot read stream ${what} while ${compute} is in flight; await it first`, ); } } @@ -1182,7 +1176,7 @@ export class Stream { get text(): StreamText { const h = this.#session.handle; // throws if the session was disposed this.#assertCurrent("read stream text"); - this.#assertNotFeeding("text"); + this.#assertNotComputing("text"); const n = this.#n; const t: any = {}; n.F.streamTextInit(t); @@ -1194,18 +1188,27 @@ export class Stream { }; } + /** Full structured snapshot of the current hypothesis (owned copies). */ + get snapshot(): Transcript { + const h = this.#session.handle; // throws if the session was disposed + this.#assertCurrent("read stream snapshot"); + if (!this.#active) throw new TranscribeError("stream has been reset"); + this.#assertNotComputing("snapshot"); + return materialize(this.#n, singleAccessors(this.#n, h)); + } + get state(): StreamState { const h = this.#session.handle; // throws if the session was disposed this.#assertCurrent("read stream state"); if (!this.#active) return "idle"; // reset() returns to idle; native reset may still be queued - this.#assertNotFeeding("state"); + this.#assertNotComputing("state"); return STREAM_STATES[this.#n.F.streamGetState(h)] ?? "idle"; } get revision(): number { const h = this.#session.handle; // throws if the session was disposed this.#assertCurrent("read stream revision"); - this.#assertNotFeeding("revision"); + this.#assertNotComputing("revision"); return this.#n.F.streamRevision(h); } @@ -1217,7 +1220,7 @@ export class Stream { get lastStatus(): TranscribeError | null { const h = this.#session.handle; // throws if the session was disposed this.#assertCurrent("read stream lastStatus"); - this.#assertNotFeeding("lastStatus"); + this.#assertNotComputing("lastStatus"); const n = this.#n; const status = n.F.streamLastStatus(h); if (status === g.TRANSCRIBE_OK) return null; diff --git a/bindings/typescript/src/types.ts b/bindings/typescript/src/types.ts index 24bcc86c..4caafe65 100644 --- a/bindings/typescript/src/types.ts +++ b/bindings/typescript/src/types.ts @@ -83,12 +83,13 @@ export interface SessionLimits { maxKvBytes: number; } -export interface TranscriptionResult { +export interface Transcript { text: string; /** The model's decoded output before family post-processing (diarization * markers, timestamp/special tokens, tag filtering, whitespace trims). * Equal to `text` modulo whitespace for families that emit clean text. */ rawText: string; + /** Model-detected language, or an empty string when none applies. */ language: string; timestampKind: TimestampKind; segments: Segment[]; @@ -96,6 +97,10 @@ export interface TranscriptionResult { words: Word[]; tokens: Token[]; timings: Timings; +} + +/** An offline transcript plus terminal run status flags. */ +export interface TranscriptionResult extends Transcript { aborted: boolean; truncated: boolean; } diff --git a/bindings/typescript/test/streaming.test.mjs b/bindings/typescript/test/streaming.test.mjs index b873de20..7c38b171 100644 --- a/bindings/typescript/test/streaming.test.mjs +++ b/bindings/typescript/test/streaming.test.mjs @@ -16,15 +16,22 @@ modelTest("streaming commits text and finalizes", STREAMING_MODEL, async () => { assert.equal(stream.lastStatus, null, "a healthy finished stream has no failure status"); const t = stream.text; assert.ok((t.committed + t.full).trim().length > 0, "expected non-empty streamed text"); + const snapshot = stream.snapshot; + assert.ok(snapshot.text.trim().length > 0, "expected non-empty structured snapshot"); + assert.ok(snapshot.segments.length > 0, "expected structured segment rows"); + assert.ok(Array.isArray(snapshot.words)); + assert.ok(Array.isArray(snapshot.tokens)); + assert.equal(typeof snapshot.language, "string"); stream.reset(); assert.equal(stream.state, "idle"); + assert.throws(() => stream.snapshot, /reset/i, "reading .snapshot after reset must throw"); s.dispose(); } finally { m.dispose(); } }); -modelTest("stream reads reject while a feed is in flight", STREAMING_MODEL, async () => { +modelTest("stream reads reject while the session is computing", STREAMING_MODEL, async () => { const m = await TranscribeModel.load(STREAMING_MODEL); try { const s = m.createSession(); @@ -34,6 +41,7 @@ modelTest("stream reads reject while a feed is in flight", STREAMING_MODEL, asyn const pending = stream.feed(chunk); // do NOT await — leave it in flight await Promise.resolve(); // let the feed's native call reach the worker assert.throws(() => stream.text, /in flight/i, "reading .text mid-feed must throw"); + assert.throws(() => stream.snapshot, /in flight/i, "reading .snapshot mid-feed must throw"); assert.throws(() => stream.state, /in flight/i); assert.throws(() => stream.revision, /in flight/i); assert.throws(() => s.limits, /in flight/i); @@ -41,6 +49,17 @@ modelTest("stream reads reject while a feed is in flight", STREAMING_MODEL, asyn await pending; // once awaited, reads are fine again assert.doesNotThrow(() => stream.text); + assert.doesNotThrow(() => stream.snapshot); + + // A finished stream wrapper still reads the same session-owned result + // storage. Guard it during a later offline call on that session too, not + // only during the wrapper's own feed/finalize calls. + await stream.finalize(); + const runPending = s.run(jfk()); + await Promise.resolve(); // let run() reach its worker + assert.throws(() => stream.snapshot, /run\(\).*in flight/i); + assert.throws(() => stream.text, /run\(\).*in flight/i); + await runPending; stream.reset(); s.dispose(); diff --git a/docs/bindings.md b/docs/bindings.md index 3c7b1176..98dcd463 100644 --- a/docs/bindings.md +++ b/docs/bindings.md @@ -65,6 +65,17 @@ automatic: Python `ctypes.c_char_p` / `.decode()`, Go `C.GoString`, Rust hands out a zero-copy view must scope it to the current callback/update turn and document that it dies at the next stream mutation. +First-class bindings expose both streaming result shapes as owned values: + +- the UI-facing `transcribe_stream_get_text()` view (full, committed, and + tentative text), and +- a full structured snapshot built from the ordinary current-result accessors + (clean/raw text, detected language, timestamp kind, segments, speaker turns, + words, tokens, and timings). + +The structured snapshot must be copied completely before the next feed or +finalize call; bindings must not return the session-owned pointers directly. + ## Diarization result contract First-class bindings expose the generic diarization surface rather than only @@ -91,8 +102,9 @@ whitespace trims). It equals the clean text modulo whitespace for families that emit clean text natively, and is the recommended replacement for `keep_special_tags` when the goal is recovering what the model emitted — unlike the flag it works for every family, covers plain-text markers, and does -not give up the clean transcript. Offline runs (single and batch) only; -streaming results do not carry it. +not give up the clean transcript. It is present on single, batch, and full +structured stream snapshots (and may be empty before a stream has produced a +successful hypothesis). When adding a new family extension, update: