diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 40178ad..c90c323 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -73,6 +73,18 @@ jobs: ruff check python tests examples ruff format --check python tests examples + - name: Public documentation language + if: matrix.os == 'ubuntu-latest' && matrix.python == '3.11' + working-directory: sdk-python + shell: bash + run: | + if rg -n -i '\b(boundary|path|surface|authority|projection|lowering|flow|layer|contracts?)\b' \ + README.md RELEASE_NOTES.md docs examples \ + --glob '*.md' --glob '*.mdx'; then + echo "Public documentation must name the API, queue, process, service, or request directly." + exit 1 + fi + - name: Python type contract if: matrix.os == 'ubuntu-latest' && matrix.python == '3.11' working-directory: sdk-python diff --git a/README.md b/README.md index 62e0bb5..f8ceb45 100644 --- a/README.md +++ b/README.md @@ -34,8 +34,10 @@ recorded. Both behaviors are off by default. ## Handle permissions and source changes PocketStation does not prompt during import or discovery. Check microphone -permission without prompting, then let Source opening report the authoritative -result. Application capture and microphone capture use separate permissions. +permission without prompting through +`pocketstation.sources.microphone_permission_observation()`, then let Source +opening report the authoritative result. Application capture and microphone +capture use separate permissions. When an application or device disappears, PocketStation reports the change and does not switch to another Source. Stop the current Session, discover again, @@ -55,7 +57,7 @@ output as independent recorded stems. Provider events and media events share one monotonic timeline, so you can see whether delay occurred before the model, inside the provider, in local output, or after Relay delivery. -See the [voice-agent debugger instructions](examples/README.md#debug-a-voice-agent-interruption-from-the-media-boundary). +See the [voice-agent debugger instructions](examples/README.md#debug-a-voice-agent-interruption). Run repository examples from a source checkout or source archive. The installed `pocketstation-demo` command is the packaged application-and-microphone demo. @@ -78,7 +80,7 @@ faster-whisper Operator, connects both stems to its audio input, and prints each transcript with its original source identity. It does not start Relay or write a recording. -The Session runs this path concurrently: +The Session runs these jobs concurrently: ```text voice application ─┐ @@ -152,14 +154,14 @@ cancelled, and invalid-buffer outcomes explicitly. ## Create an integration -PocketStation uses four open boundaries: +PocketStation provides four integration APIs: -| Boundary | Use it when | +| API | Use it when | |---|---| | `Source` | Media or signals enter the Session. | | `Operator` | Work transforms media or emits typed signals. | | `Connector` | Media or signals leave for an external system. | -| `Endpoint` | You need the lower-level outbound execution contract. | +| `Endpoint` | You need direct control of an outbound worker. | Pass one function when the destination is already open: @@ -174,7 +176,7 @@ destination = pks_aio.Connector(send=send_audio) application.send_to(destination) ``` -Subclass the synchronous or asyncio contract when the provider opens and +Subclass the synchronous or asyncio Connector when the provider opens and closes resources. The provider class owns its connection; the Session owns bounded delivery, lineage, observations, drain, abort, and joined shutdown: @@ -207,8 +209,8 @@ destination. See [Create an integration](docs/guides/integrations.md) for deadlines, failures, and the advanced SPI. Python provider callbacks execute on bounded off-realtime workers. They cannot -be used as native capture callbacks. Compiled native extensions remain the path -for native provider code, and process sidecars remain available when crash +be used as native capture callbacks. Use compiled native extensions for native +provider code, or a process sidecar when crash isolation is required. ## Use Relay from Python @@ -227,7 +229,7 @@ redacted secrets, and matching synchronous and asyncio APIs. asyncio namespace provides awaitable lifecycle, stream, Relay, provider, and audio-input operations without creating another audio queue. -Python callbacks still cross the interpreter boundary. Capture, routing, +Python callbacks still enter the interpreter. Capture, routing, recording, and Relay transport remain native-speed; arbitrary Python model code does not have the same execution cost as Rust. @@ -236,7 +238,7 @@ does not have the same execution cost as Rust. | Area | Support | |---|---| | Python | 3.11 and newer | -| macOS Apple silicon | Installed wheel, application capture, physical microphone, 10 ms voice path, Relay, Chromium, and multistem recording tested | +| macOS Apple silicon | Installed wheel, application capture, physical microphone, 10 ms voice capture, Relay, Chromium, and multistem recording tested | | Linux | Core application selection and 10 ms capture tested; installed Python distribution qualification in progress | | Windows 11 ARM64 | Core application selection and 10 ms capture tested in a VM; installed Python distribution and physical-device qualification in progress | | WAN and TURN | Not yet qualified | @@ -246,7 +248,7 @@ Connector `0.1.2`. The Rust-to-Python audio read currently copies native samples into Python-owned bytes before exposing a `memoryview`. The view avoids another Python-side copy; -the complete boundary is not zero-copy. +the call into Python still copies samples and is not zero-copy. ## Develop the SDK @@ -266,16 +268,22 @@ uv run mypy python tests/qualification/typing_contract.py examples API ownership. - [Write application-owned audio](docs/guides/application-audio.md) — bounded PCM input and selective output cancellation. +- [Process audio and typed signals](docs/guides/process-audio-and-signals.md) — + Operators, named ports, generated audio, and finite model work. - [Record and observe a Session](docs/guides/record-and-observe.md) — multistem outcomes, route metrics, and lifecycle events. +- [Keep each source identifiable](docs/concepts/source-identity-and-time.md) — + selection, persistence, timestamps, generations, and discontinuities. +- [Read events, metrics, outcomes, and errors](docs/reference/events-and-errors.md) + — setup failure, live observations, and terminal results. - [Prepare each platform](docs/operations/platform-support.md) — permissions, source persistence, explicit rediscovery, and fallback policy. - [`examples/README.md`](examples/README.md) — runnable examples and prerequisites. - `pocketstation.capture` — concise application and microphone capture. - `pocketstation.session` — Session declarations and lifecycle. -- `pocketstation.graph` — stems, ports, routes, and signal contracts. +- `pocketstation.graph` — stems, ports, routes, and signal specifications. - `pocketstation.connector` — outbound provider authoring. - `pocketstation.operator_authoring` — computation authoring. - `pocketstation.source_authoring` — inbound provider authoring. -- `pocketstation.aio` — asyncio projection of the same engine. +- `pocketstation.aio` — asyncio APIs for the same engine. diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 59af478..98ba073 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -1,5 +1,19 @@ # PocketStation for Python release notes +## Unreleased + +### Added + +Advanced integrations can now configure `RouteSettings` as accepted media plus +a separate `DeliveryPolicy`. Connector, Endpoint, and signal-subscription APIs +accept the clearer `route_settings=` keyword. Runtime metrics expose +`RouteObservability`, `RouteLatencyMeasurement`, `RouteDeliveryMetrics`, and +`SignalQueueMetrics`. + +Existing `EdgeContract`, `EdgeObservabilityLevel`, `RouteLatencyBoundary`, +`EdgeMetrics`, `TypedEdgeMetrics`, and `edge=` uses remain compatible throughout +the 0.1.x series. + ## 0.1.2 — 2026-09-01 Capture code can now send source-aware audio to an external system through one @@ -60,7 +74,7 @@ python -m pip install --upgrade pocketstation==0.1.2 ``` Python Connectors run outside native realtime partitions, but each frame still -crosses the Python interpreter boundary. Use a native Connector when that cost +enters the Python interpreter. Use a native Connector when that cost is not appropriate for the destination. ## 0.1.1 — 2026-09-01 @@ -98,7 +112,7 @@ does not execute on native capture callbacks. ### Voice interruption example `examples/debug_voice_ai.py` connects a physical microphone directly to OpenAI -Realtime, routes generated speech through the normal Session audio path, and +Realtime, routes generated speech through the normal Session audio stream, and records microphone input, generated output, and browser playback separately. The resulting timeline distinguishes provider cancellation from Core output cancellation and receiver delivery. @@ -110,7 +124,7 @@ provider-history truncation as unavailable instead of inferring them. ### Supported and qualified environments - macOS Apple silicon has installed-wheel evidence for application capture, - physical microphone input, the 10 ms voice path, Relay, Chromium, and + physical microphone input, the 10 ms voice configuration, Relay, Chromium, and multistem recording. - Linux and Windows have Core application-selection and 10 ms capture evidence. Installed Python distributions are qualified separately by the release diff --git a/docs/README.md b/docs/README.md index 815f497..b14c0f3 100644 --- a/docs/README.md +++ b/docs/README.md @@ -13,6 +13,7 @@ need the lower-level API. ## Create an integration - [Write application-owned audio into a Session](guides/application-audio.md) +- [Process audio and typed signals](guides/process-audio-and-signals.md) - [Record stems and inspect Session delivery](guides/record-and-observe.md) - [Compose a bounded voice workflow](guides/voice.md) - [Publish a named AudioBus through Relay](guides/relay.md) @@ -21,6 +22,8 @@ need the lower-level API. ## Understand the system - [Session ownership, bounds, and shutdown](concepts/session-and-bounds.md) +- [Source identity, timestamps, and discontinuities](concepts/source-identity-and-time.md) +- [Media and delivery settings for each route](concepts/route-settings.md) ## Operate and upgrade @@ -35,13 +38,14 @@ need the lower-level API. - [`pocketstation.aio`](../python/pocketstation/aio/__init__.py) — asyncio over the same native Session. - [`pocketstation.voice`](../python/pocketstation/voice/__init__.py) — - provider-neutral voice composition contracts. + provider-neutral voice composition protocols. - [`pocketstation.graph`](../python/pocketstation/graph.py) — stems, routes, ports, and signals. - [`pocketstation.observations`](../python/pocketstation/observations.py) — runtime metrics and outcomes. - [Python API map](reference/api-map.md) — public entry points and advanced modules by task. +- [Events, metrics, outcomes, and errors](reference/events-and-errors.md) For supported platforms and current qualification limits, read the [repository README](../README.md#platform-support). diff --git a/docs/concepts/route-settings.md b/docs/concepts/route-settings.md new file mode 100644 index 0000000..e2399ac --- /dev/null +++ b/docs/concepts/route-settings.md @@ -0,0 +1,72 @@ +# Set media and delivery behavior for a route + +Every destination receives data through its own finite queue. PocketStation +chooses suitable settings for common capture, recording, Connector, and signal +subscription work. Use `RouteSettings` when an integration needs to state those +choices explicitly. + +`RouteSettings` keeps two decisions together without mixing their APIs: + +- `MediaCaps` describes the media accepted by the route. +- `DeliveryPolicy` describes timing, queue pressure, loss, copying, and + observations. + +## Use a preset first + +Realtime PCM should keep capture moving when one destination falls behind: + +```python +from pocketstation.graph import RouteSettings + +settings = RouteSettings.realtime_audio() +``` + +Typed signals use a finite asynchronous queue: + +```python +settings = RouteSettings.bounded_async() +``` + +Connected ports still negotiate the concrete sample rate, channel layout, and +signal schema. + +## Change delivery without changing media + +```python +from pocketstation.graph import ( + BackpressurePolicy, + DeliveryPolicy, + RouteSettings, +) + +delivery = DeliveryPolicy.realtime_audio().with_backpressure( + BackpressurePolicy.DROP_OLDEST +) +settings = RouteSettings.realtime_audio().with_delivery_policy(delivery) +``` + +Pass the result to a Connector destination: + +```python +destination = session.destination(connector, route_settings=settings) +application.send(destination) +``` + +Use `DROP_OLDEST` only when fresh realtime media matters more than queued +media. Use the default `DROP_NEWEST` when preserving items already accepted by +the destination is more important. A blocking policy is not valid on a capture +callback or realtime worker. + +## Read what the Session compiled + +Advanced Connector and Endpoint preparation objects expose `route_settings`. +`RouteMetrics.delivery` returns `RouteDeliveryMetrics` for queue depth, +attempted delivery, drops, and discontinuities. Typed signals use +`SignalQueueMetrics`. Check those observations before increasing capacity: a +larger finite queue can hold older audio without solving the slow destination. + +`EdgeContract` remains an import-compatible name for `RouteSettings` in the +0.1.x series. Existing `edge=` keyword arguments continue to work. New code +should use `route_settings=` so the decision is clear at the call site. +`EdgeMetrics` and `TypedEdgeMetrics` remain compatibility names for the clearer +metrics types. diff --git a/docs/concepts/session-and-bounds.md b/docs/concepts/session-and-bounds.md index 1732093..49f1a44 100644 --- a/docs/concepts/session-and-bounds.md +++ b/docs/concepts/session-and-bounds.md @@ -21,7 +21,7 @@ destinations does not mix that identity or recapture the source. ## Every crossing is finite Audio input, polling, Python provider work, signals, Relay, and recording use -declared capacities. When a boundary is full, PocketStation returns or records +declared capacities. When a queue is full, PocketStation returns or records pressure according to that route's policy. It does not hide pressure in an unbounded `asyncio.Queue`. @@ -36,7 +36,7 @@ bounded off-realtime workers. Native capture callbacks remain allocation-free, lock-free, blocking-free, async-free, log-free, and panic-free. Use a compiled extension when code must stay native. Use a process sidecar when -crash isolation matters. Neither boundary creates a second Session engine. +crash isolation matters. Neither option creates a second Session engine. ## Stop and cancel are different diff --git a/docs/concepts/source-identity-and-time.md b/docs/concepts/source-identity-and-time.md new file mode 100644 index 0000000..0f7ea1b --- /dev/null +++ b/docs/concepts/source-identity-and-time.md @@ -0,0 +1,71 @@ +# Keep each source identifiable + +PocketStation keeps application audio, microphone audio, and application-owned +PCM separate even when they use the same Operator, Connector, RelaySession, or +recording. A frame does not become anonymous when it leaves capture. + +## Read the identities carried by a frame + +`AudioFrame` exposes the identifiers needed to correlate live delivery with +provider observations and recorded stems: + +| Value | Meaning | +|---|---| +| `source_id` | The Source instance that produced the media. | +| `stream_id` | One output stream declared by that Source. | +| `stem_id` | The independently routed and recorded audio stem. | +| `sequence_number` | The frame order within the current source generation. | +| `timestamp_start_ns` | The first sample time in its declared clock domain. | +| `source_generation` | The current attachment of a recoverable Source. | +| `discontinuity_epoch` | A change that prevents the timeline from being treated as continuous. | + +Use these values together. A display name such as `Zoom` helps a person choose +an application; it is not a media identity. + +## Select an application before capture starts + +The concise API accepts an exact display name, an application identifier, or a +positive process ID: + +```python +import pocketstation as pks +from pocketstation.sources import SourceQuery + +with pks.capture(application="Zoom") as live: + for frame in live.audio: + print(frame.source_id, frame.stream_id, frame.sequence_number) +``` + +PocketStation rejects zero matches and ambiguous matches. It does not select +the first process silently. + +Use discovery when the application must display choices or remember one: + +```python +matches = pks.discover_sources(SourceQuery.application("Zoom")) +if len(matches) != 1: + raise RuntimeError("Select one running Zoom source") + +selected = matches[0] +source = pks.Source.from_discovered(selected) +``` + +Store the discovered selector only for its reported persistence scope. A +process ID ends with that process. A platform application identity may survive +a normal restart, but should still be resolved again before a new Session. + +## Treat discontinuity as data + +A sequence gap, timestamp gap, source restart, or dropped route frame changes +what downstream code may infer. PocketStation records those changes instead of +inserting silent continuity. + +An Operator can preserve the input lineage and add derivation for its output. +A Connector receives the source-aware `AudioFrame`. A recording manifest keeps +the same source and stem identity beside the WAV file. + +Do not rewrite a missing timestamp as zero, merge two source generations, or +claim that a remote receiver played a sample from a sender timestamp alone. + +Continue with [recording and observations](../guides/record-and-observe.md) or +[platform source persistence](../operations/platform-support.md). diff --git a/docs/getting-started/capture.md b/docs/getting-started/capture.md index 0e6d44c..26ecb57 100644 --- a/docs/getting-started/capture.md +++ b/docs/getting-started/capture.md @@ -16,7 +16,7 @@ Install the package: python -m pip install pocketstation ``` -## Run the shortest capture path +## Capture an application ```python import pocketstation diff --git a/docs/guides/application-audio.md b/docs/guides/application-audio.md index 66249d2..eeafba7 100644 --- a/docs/guides/application-audio.md +++ b/docs/guides/application-audio.md @@ -55,7 +55,7 @@ assistant.write(samples, generation=generation) generation.cancel() ``` -Core discards matching frames that are still in its bounded sender paths. +Core discards matching frames that are still waiting in its sender queues. Microphone capture, recording, and unrelated outputs continue. This operation cannot recall audio already accepted by a remote service or diff --git a/docs/guides/integrations.md b/docs/guides/integrations.md index 57de06e..4dac8d2 100644 --- a/docs/guides/integrations.md +++ b/docs/guides/integrations.md @@ -1,10 +1,10 @@ # Create a Source, Operator, Connector, or Endpoint -Choose the boundary by the direction and ownership of the work. All four use +Choose the API by the direction of the work. All four use the same Session compiler, finite queues, lifecycle, observations, and joined shutdown. -| Boundary | Use it when | +| API | Use it when | |---|---| | `Source` | Media or signals enter the Session. | | `Operator` | Computation transforms media or emits signals. | @@ -22,8 +22,8 @@ another Session, or hide an unbounded queue behind a provider callback. ## Choose a Connector for outbound delivery -A Connector is the outbound boundary between one PocketStation Session and an -external system. Use it to publish source-aware audio to a WebSocket, call +A Connector sends data from one PocketStation Session to an external system. +Use it to publish source-aware audio to a WebSocket, call transport, monitoring service, storage API, or provider SDK without rebuilding capture, buffering, routing, and shutdown in every integration. @@ -44,7 +44,7 @@ does not create a second media engine or skip Core. Do not use a Connector for work that changes audio into another signal. A transcriber is an `Operator`. Audio arriving from a provider is a `Source` or an application-owned `AudioInput`. A duplex integration can compose inbound -and outbound boundaries while the Session remains the only runtime. +and outbound APIs while the Session remains the only runtime. ## Send audio to a provider @@ -119,7 +119,7 @@ raised them. Raise `ConnectorError` when the provider has a stable error code or retryability classification. Other routes continue independently when one Connector is slow or fails. -## Lifecycle and delivery contract +## Lifecycle and delivery behavior | Method | Provider responsibility | PocketStation responsibility | |---|---|---| @@ -167,5 +167,5 @@ errors, logs, metrics, observations, or object representations. Build and install the distribution into a clean environment. Run the provider through a normal Session, cause saturation and cancellation, and verify joined -shutdown. A mock proves only the adapter contract; a network integration needs +shutdown. A mock proves only the adapter calls; a network integration needs provider and receiver evidence. diff --git a/docs/guides/process-audio-and-signals.md b/docs/guides/process-audio-and-signals.md new file mode 100644 index 0000000..61344c8 --- /dev/null +++ b/docs/guides/process-audio-and-signals.md @@ -0,0 +1,80 @@ +# Process audio and typed signals + +Use an `Operator` when work consumes Session media or signals and emits a +derived result. Transcription, classification, translation, speech synthesis, +and policy checks are Operators when their outputs remain inside the Session. + +A Python Operator runs on a bounded off-realtime worker. Capture and unrelated +destinations continue while the Operator works. The Operator does not create a +second Session or call Python from a native capture callback. + +## Declare the inputs and outputs + +Each port has a name, a `SignalSpec`, and media requirements. The declaration +lets PocketStation reject incompatible connections before capture begins. + +```python +from pocketstation.aio.operator_authoring import operator +from pocketstation.graph import PortSpec, SignalSpec +from pocketstation.operator_authoring import OperatorEmission, OperatorManifest + +request = SignalSpec.text(role="request") +result = SignalSpec.text(role="result.final") + +manifest = OperatorManifest( + "io.example.uppercase.v1", + inputs=(PortSpec.input("input", request),), + outputs=(PortSpec.output("output", result),), +) + +@operator(manifest) +async def uppercase(input_port, envelope): + assert input_port == "input" + return (OperatorEmission.text(str(envelope.payload).upper(), signal=result),) +``` + +Register the implementation once, declare an instance, then connect named +ports: + +```python +operator = session.register_operator(uppercase).declare() +source.output("events").connect(operator.input("input")) +results = session.subscribe(operator.output("output"), signal=result) +``` + +`SignalSpec` supplies runtime and cross-language identity. Python type hints +help local development but are not used as a wire format. + +## Emit generated audio + +An Operator can return one exact PCM frame with `OperatorEmission.audio()`. +Call `reenter_audio()` on the declared output to turn that generated signal +into a normal audio stem: + +```python +generated = operator.output("audio").reenter_audio() +generated.record("generated") +generated.send_to(destination) +``` + +The reentered stem can be recorded, polled, sent to a Connector, or connected +to another Operator. Core assigns its stream and stem identity and retains the +derivation from the Operator output. + +## Keep model work finite + +Set finite queue capacity and processing time in `OperatorManifest`. Implement +`cancel()` when provider work can be interrupted and `close()` to release the +provider exactly once. Return terminal signals only when the provider has +actually produced a terminal result. + +Slow inference may saturate its own input. Inspect Operator input metrics, +route drops, and discontinuities. Do not run model inference in a frame iterator +that must keep another destination drained. + +Use a `Connector` instead when the result leaves the Session and the provider +does not emit a Session signal. Use `AudioInput` when application or provider +code already owns PCM that should enter the Session. + +Continue with [creating an integration](integrations.md) or [voice +composition](voice.md). diff --git a/docs/guides/record-and-observe.md b/docs/guides/record-and-observe.md index b0ff515..78a1fc9 100644 --- a/docs/guides/record-and-observe.md +++ b/docs/guides/record-and-observe.md @@ -52,7 +52,7 @@ a browser played the same sample. Leaving the context requests normal stop and drains accepted bounded work. Call `cancel()` on the explicit `RunningSession` API when active provider or -sidecar work must abort. Both paths join Session-owned workers before returning +sidecar work must abort. Both shutdown modes join Session workers before returning a terminal `StopResult`. After shutdown, inspect: diff --git a/docs/guides/voice.md b/docs/guides/voice.md index 0dff4ca..aa96ac5 100644 --- a/docs/guides/voice.md +++ b/docs/guides/voice.md @@ -1,12 +1,12 @@ # Compose a bounded voice workflow -`pocketstation.voice` defines provider-neutral voice contracts. +`pocketstation.voice` defines provider-neutral Python protocols. `pocketstation.aio.Session` composes those providers around one native Session. The package does not contain a model provider, capture engine, Relay implementation, or agent framework. Start with a declared asyncio Session, one input stem, one `AudioInput` for -generated speech, and provider objects that implement the selected contracts. +generated speech, and provider objects that implement the selected protocols. ## Choose one provider shape diff --git a/docs/operations/platform-support.md b/docs/operations/platform-support.md index 0d6722b..efc155c 100644 --- a/docs/operations/platform-support.md +++ b/docs/operations/platform-support.md @@ -12,13 +12,13 @@ do not rely on a sibling Rust checkout. ## Check permission without prompting -`pocketstation.microphone_permission_observation()` reads the current state -without showing consent UI: +`microphone_permission_observation()` reads the current state without showing +consent UI: ```python -import pocketstation as pks +from pocketstation.sources import microphone_permission_observation -permission = pks.microphone_permission_observation() +permission = microphone_permission_observation() print(permission.value) ``` @@ -40,8 +40,9 @@ after a restart: ```python import pocketstation as pks +from pocketstation.sources import SourceQuery -matches = pks.discover_sources(pks.SourceQuery.application("Zoom")) +matches = pks.discover_sources(SourceQuery.application("Zoom")) if len(matches) != 1: raise RuntimeError("Select one running Zoom source") @@ -118,7 +119,7 @@ permissions explicitly. An installed import and component test establish package correctness. A device claim needs the physical device. A latency claim needs p50, p95, p99, and -maximum measurements from the same frame definition and clock boundary. +maximum measurements from the same frame definition and clock domain. The 10 ms profile sets PocketStation's frame cadence. It does not guarantee sub-10 ms capture-to-Python, network, browser, or acoustic latency. WAN and TURN diff --git a/docs/reference/api-map.md b/docs/reference/api-map.md index a079282..166cf06 100644 --- a/docs/reference/api-map.md +++ b/docs/reference/api-map.md @@ -1,7 +1,7 @@ # Find the Python API for a task Use the package root for capture and Session lifecycle. Import advanced graph, -provider, Relay, and diagnostic contracts from the module that owns them. +provider, Relay, and diagnostic APIs from the module that implements them. ## Start and stop a Session @@ -14,9 +14,9 @@ provider, Relay, and diagnostic contracts from the module that owns them. | Discover running sources | `pocketstation.discover_sources` | | Feed application-owned PCM | `Session.audio_input` | -`capture()` is the short path. Use `Session` when you need more than one -destination, custom route policies, provider composition, Relay, or detailed -observations. Both paths use the same Rust engine. +`capture()` is the convenience API. Use `Session` when you need more than one +destination, explicit route settings, provider composition, Relay, or detailed +observations. Both forms use the same Rust engine. ## Compose and route media @@ -27,6 +27,12 @@ observations. Both paths use the same Rust engine. | Source declarations and discovery | `pocketstation.sources` | | Application-owned PCM | `pocketstation.audio_input` | | Runtime events, metrics, and outcomes | `pocketstation.observations` | +| Accepted media and delivery behavior | `pocketstation.graph.RouteSettings` and `DeliveryPolicy` | + +Read [source identity and time](../concepts/source-identity-and-time.md) before +persisting selectors or correlating provider events with recorded media. +Read [route settings](../concepts/route-settings.md) before changing queue or +loss behavior. ## Connect external systems @@ -47,7 +53,7 @@ callbacks never call Python. ## Build a voice workflow -`pocketstation.voice` contains the provider-neutral contracts. +`pocketstation.voice` contains the provider-neutral Python protocols. `pocketstation.aio.Session.conversation()` composes either: - a `StreamingTranscriber`, `ResponseModel`, and `SpeechSynthesizer`, with an @@ -61,8 +67,8 @@ depending on interruption or playout observations. ## Handle failures Start with `PocketStationError`, `CaptureError`, and `SessionError` from the -package root. Advanced modules expose errors for their own boundary. Preserve +package root. Advanced modules expose errors for the feature they implement. Preserve the structured error and inspect the Session outcome before retrying. Continue with [Session ownership, bounds, and shutdown](../concepts/session-and-bounds.md) -or [troubleshooting](../troubleshooting.md). +or [events, metrics, outcomes, and errors](events-and-errors.md). diff --git a/docs/reference/events-and-errors.md b/docs/reference/events-and-errors.md new file mode 100644 index 0000000..b8e8636 --- /dev/null +++ b/docs/reference/events-and-errors.md @@ -0,0 +1,80 @@ +# Read events, metrics, outcomes, and errors + +PocketStation reports setup failures, live observations, and terminal results +separately. A Session that started successfully may still receive no media, +drop frames on one route, fail one provider, or finish an incomplete recording. + +## Setup failures + +| Operation | Main error | +|---|---| +| Select or open a Source | `CaptureError` | +| Declare an invalid Session | `SessionError` | +| Write incompatible application PCM | `AudioInputWriteError` | +| Configure a provider | feature-specific `PocketStationError` subclass | + +Preserve the stable error code and human-readable message. Do not retry every +failure. Permission denial, ambiguous selection, invalid media, and missing +credentials require a different action from a temporary provider failure. + +## Session events + +`running.events` is a finite stream of `SessionEvent` values. Events identify +the component and Session time for lifecycle changes, Source failures, Endpoint +failures, rollback, finalization, and terminal state. + +The event queue has its own capacity and drop observations. A dropped diagnostic +event does not prove that audio was dropped; inspect media route metrics too. + +## Metrics + +`running.metrics()` returns one immutable `SessionMetrics` snapshot. It groups +Source, route, polling, Operator, Endpoint, application-owned audio, recording, +and sidecar measurements. + +For a route, inspect: + +- declared and current queue depth; +- peak queue depth; +- delivered and dropped frames; +- discontinuity counts; and +- latency only when the measurement definition and unit are present. + +Missing measurements remain `None`. Sender time, Relay receive time, browser +jitter-buffer time, and acoustic playout are different observations. +`RouteMetrics.source_latency_measurement` returns a +`RouteLatencyMeasurement`; `source_latency_unit` gives its unit. +`RouteLatencyBoundary` and `source_latency_boundary` remain 0.1.x compatibility +names. + +## Terminal results + +After stop or cancellation, inspect `StopResult` before reporting success: + +```python +result = running.stop() +if not result.success and result.terminal_event is not None: + for failure in result.terminal_event.failures: + print(failure.error_code, failure.stage, failure.message) +``` + +When recording was requested, also require a complete `RecordingOutcome` and +inspect every stem. When a Connector or voice provider was required, inspect +its final result separately. + +Normal close requests a drain of accepted finite work. Cancellation asks active +workers to abort before the same joined shutdown. Neither mode may leave an +unjoined provider task or child process and still report success. + +## Voice observations + +`VoiceEvent` and `ConversationOutcome` keep transcript, response, synthesis, +output cancellation, and provider failure observations on the Session +timeline. They distinguish provider task cancellation, Core queued-output +cancellation, Connector queue clearing, receiver observation, and acoustic +hearing. + +An unavailable receiver or acoustic observation must remain unavailable. It is +not equivalent to zero delay or successful interruption. + +Continue with [troubleshooting](../troubleshooting.md) for recovery steps. diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index ea0cc36..12686a1 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -1,6 +1,6 @@ # Troubleshoot capture, delivery, and shutdown -Start with the boundary that did not produce the expected result. PocketStation +Start with the component that did not produce the expected result. PocketStation keeps source opening, bounded delivery, provider work, Relay delivery, recording, and receiver playout as separate observations. @@ -24,7 +24,8 @@ process or device. ## Permission changes are not reflected -`microphone_permission_observation()` never prompts. Treat `NOT_OBSERVABLE` as +`pocketstation.sources.microphone_permission_observation()` never prompts. +Treat `NOT_OBSERVABLE` as unknown and use the typed Source open result. On macOS, restart the Python host after changing screen-recording or microphone consent. On Windows and Linux, verify that the current desktop user or service can access the selected audio @@ -54,7 +55,7 @@ as separate results. Neither result proves loudspeaker playout. ## Generated speech continues after interruption -Check each boundary separately: +Check each component separately: 1. provider response cancellation; 2. Core pending-output cancellation; @@ -70,7 +71,7 @@ does not expose it. Use finite provider and network deadlines. Close application-owned inputs when no more frames will arrive. Request normal stop to drain accepted work, or cancel when active asynchronous work must abort. Then inspect the `StopResult`, -provider outcome, and recording outcome for the boundary that did not join. +provider outcome, and recording outcome for the component that did not stop. When reporting an issue, include the PocketStation versions, operating system, source selector, Session events, route metrics, and structured terminal error. diff --git a/examples/README.md b/examples/README.md index 80fb3a4..78f169e 100644 --- a/examples/README.md +++ b/examples/README.md @@ -20,7 +20,7 @@ The example asks which running application to capture. PocketStation owns the bounded route and shutdown; the surrounding WebSocket context owns its connection. -## Debug a voice-agent interruption from the media boundary +## Debug a voice-agent interruption [`debug_voice_ai.py`](debug_voice_ai.py) connects PocketStation directly to OpenAI Realtime. PocketStation owns the physical microphone, generated @@ -28,7 +28,7 @@ assistant PCM, browser delivery, independent recording stems, bounded queues, and sender-side output cancellation. The provider owns speech recognition and the model response. -The example observes three separate audio paths: +The example observes three separate audio streams: - the microphone sent to the model; - generated assistant audio sent to Relay; diff --git a/python/pocketstation/__init__.py b/python/pocketstation/__init__.py index d75b983..f5d2339 100644 --- a/python/pocketstation/__init__.py +++ b/python/pocketstation/__init__.py @@ -1,6 +1,6 @@ """PocketStation's concise Python entry point. -Advanced graph, authoring, Relay, extension, and diagnostic contracts live in +Advanced graph, authoring, Relay, extension, and diagnostic APIs live in their named modules. They are not duplicated at the package root. """ diff --git a/python/pocketstation/_api.py b/python/pocketstation/_api.py index 753fd47..7d11af0 100644 --- a/python/pocketstation/_api.py +++ b/python/pocketstation/_api.py @@ -124,6 +124,7 @@ ClockDomain, Codec, CopyPolicy, + DeliveryPolicy, DeliverySemantics, DerivedStream, EdgeContract, @@ -142,6 +143,8 @@ OperatorInstance, PortDirection, PortSpec, + RouteObservability, + RouteSettings, SampleFormat, SignalKind, SignalSpec, @@ -186,7 +189,9 @@ RecordingDiscontinuityKind, RecordingState, RelayPublishOutcome, + RouteDeliveryMetrics, RouteLatencyBoundary, + RouteLatencyMeasurement, RouteLatencyUnit, RouteObservationInterval, SessionComponent, @@ -204,6 +209,7 @@ SessionTraceRecorderOutcome, SessionTraceRecordType, SessionTraceValidation, + SignalQueueMetrics, SourceMetrics, TerminationDisposition, TypedEdgeMetrics, @@ -402,6 +408,7 @@ "ControlClient", "ControlPlaneError", "CopyPolicy", + "DeliveryPolicy", "DeliverySemantics", "DerivedRouteMetrics", "DerivedStream", @@ -501,11 +508,15 @@ "RelayRoute", "RelaySession", "RelayTimeoutError", + "RouteDeliveryMetrics", "RouteId", "RouteLatencyBoundary", + "RouteLatencyMeasurement", "RouteLatencyUnit", "RouteMetrics", + "RouteObservability", "RouteObservationInterval", + "RouteSettings", "RunningEndpointDriver", "RunningSession", "RuntimeCompatibility", @@ -561,6 +572,7 @@ "SignalKind", "SignalLineage", "SignalPayload", + "SignalQueueMetrics", "SignalReadResult", "SignalSpec", "SignalStream", diff --git a/python/pocketstation/aio/audio_input.py b/python/pocketstation/aio/audio_input.py index b51921d..3054773 100644 --- a/python/pocketstation/aio/audio_input.py +++ b/python/pocketstation/aio/audio_input.py @@ -1,4 +1,4 @@ -"""Asyncio projection of bounded application-owned PCM input.""" +"""Asyncio API for bounded application-owned PCM input.""" from __future__ import annotations diff --git a/python/pocketstation/aio/connector.py b/python/pocketstation/aio/connector.py index 764b17a..6a20b9b 100644 --- a/python/pocketstation/aio/connector.py +++ b/python/pocketstation/aio/connector.py @@ -53,7 +53,7 @@ from ..connector import ( RegisteredConnector as SyncRegisteredConnector, ) -from ..graph import EdgeContract, Endpoint +from ..graph import EdgeContract, Endpoint, RouteSettings @dataclass(frozen=True, slots=True) @@ -710,9 +710,14 @@ def declare( self, configuration: ConnectorConfigurationInput = (), *, + route_settings: RouteSettings | None = None, edge: EdgeContract | None = None, ) -> Endpoint: - return self._registered.declare(configuration, edge=edge) + return self._registered.declare( + configuration, + route_settings=route_settings, + edge=edge, + ) async def observations(self) -> tuple[ConnectorRuntimeObservations, ...]: return await asyncio.to_thread(self._registered.observations) diff --git a/python/pocketstation/aio/endpoint_authoring.py b/python/pocketstation/aio/endpoint_authoring.py index b00512e..f797e3f 100644 --- a/python/pocketstation/aio/endpoint_authoring.py +++ b/python/pocketstation/aio/endpoint_authoring.py @@ -1,4 +1,4 @@ -"""Bounded asyncio projection of Core's advanced Endpoint lifecycle.""" +"""Bounded asyncio API for Core's advanced Endpoint lifecycle.""" from __future__ import annotations @@ -27,7 +27,7 @@ from ..endpoint_authoring import PreparedEndpointDriver as SyncPreparedEndpointDriver from ..endpoint_authoring import RegisteredEndpoint as SyncRegisteredEndpoint from ..endpoint_authoring import RunningEndpointDriver as SyncRunningEndpointDriver -from ..graph import EdgeContract, Endpoint +from ..graph import EdgeContract, Endpoint, RouteSettings from ..observations import EndpointFailureStage _Result = TypeVar("_Result") @@ -222,9 +222,14 @@ def declare( self, configuration: EndpointConfigurationInput = (), *, + route_settings: RouteSettings | None = None, edge: EdgeContract | None = None, ) -> Endpoint: - return self._registered.declare(configuration, edge=edge) + return self._registered.declare( + configuration, + route_settings=route_settings, + edge=edge, + ) def _wait_for_provider( diff --git a/python/pocketstation/aio/operator_authoring.py b/python/pocketstation/aio/operator_authoring.py index 2511d85..6aa5179 100644 --- a/python/pocketstation/aio/operator_authoring.py +++ b/python/pocketstation/aio/operator_authoring.py @@ -46,7 +46,7 @@ def __post_init__(self) -> None: class OperatorNode: async def prepare(self, context: OperatorPrepareContext) -> None: - """Observe compiled port and edge contracts before processing.""" + """Observe compiled ports and route settings before processing.""" async def process( self, input_port: str, envelope: SignalEnvelope[object] diff --git a/python/pocketstation/aio/session.py b/python/pocketstation/aio/session.py index 1445a8e..b254d10 100644 --- a/python/pocketstation/aio/session.py +++ b/python/pocketstation/aio/session.py @@ -32,6 +32,7 @@ from ..graph import ( EdgeContract, Endpoint, + RouteSettings, SignalSpec, Stem, _GraphSessionDeclarations, @@ -456,10 +457,15 @@ def destination( connector: Connector | SyncConnector, configuration: ConnectorConfigurationInput = (), *, + route_settings: RouteSettings | None = None, edge: EdgeContract | None = None, ) -> Endpoint: """Declare one Connector destination using an idempotent registration.""" - return self.register_connector(connector).declare(configuration, edge=edge) + return self.register_connector(connector).declare( + configuration, + route_settings=route_settings, + edge=edge, + ) def register_endpoint( self, endpoint: EndpointProvider | SyncEndpointProvider diff --git a/python/pocketstation/audio_input.py b/python/pocketstation/audio_input.py index 34cb80a..dbd9da0 100644 --- a/python/pocketstation/audio_input.py +++ b/python/pocketstation/audio_input.py @@ -16,7 +16,7 @@ @dataclass(frozen=True, slots=True) class AudioInputConfig: - """Finite PCM contract shared by the convenient and advanced APIs.""" + """Finite PCM configuration shared by the concise and advanced APIs.""" name: str sample_rate_hz: int = 48_000 @@ -133,7 +133,7 @@ def try_write( except BufferError as error: # PyO3 rejects an incompatible buffer format before entering the # native method, so it cannot attach PocketStation's coded error. - # Keep that binding detail out of the public SDK contract. + # Keep that binding detail out of the public SDK API. raise AudioInputBufferError( "samples must be a C-contiguous float32 buffer", "audio_input.invalid_buffer", diff --git a/python/pocketstation/compatibility.py b/python/pocketstation/compatibility.py index 99a3c0b..d47c8af 100644 --- a/python/pocketstation/compatibility.py +++ b/python/pocketstation/compatibility.py @@ -7,7 +7,7 @@ @dataclass(frozen=True, slots=True) class RuntimeCompatibility: - """Exact native components and interpreter contract embedded by the wheel.""" + """Native component and interpreter versions embedded by the wheel.""" sdk_version: str core_version: str diff --git a/python/pocketstation/connector.py b/python/pocketstation/connector.py index 264a30e..15b0a8f 100644 --- a/python/pocketstation/connector.py +++ b/python/pocketstation/connector.py @@ -38,7 +38,9 @@ Multiplicity, PortDirection, PortSpec, + RouteSettings, SignalSpec, + _select_route_settings, ) from .signal import SignalEnvelope @@ -286,7 +288,7 @@ def value(self) -> str | bool | int: unsigned = self._native.as_unsigned_integer() if unsigned is not None: return unsigned - raise AssertionError("native Connector value has no compatible projection") + raise AssertionError("native Connector value has no compatible Python wrapper") def __repr__(self) -> str: if self.kind is ConnectorConfigurationValueKind.SECRET: @@ -447,7 +449,7 @@ class ConnectorRequirement: @dataclass(frozen=True, slots=True) class ConnectorManifest: - """Provider-neutral outbound Connector contract compiled by Core.""" + """Provider-neutral outbound Connector compiled by Core.""" operator_id: str package_version: str @@ -554,9 +556,13 @@ def signal(self) -> SignalSpec[object]: def media(self) -> MediaCaps: return MediaCaps._from_native(self._native.media) + @property + def route_settings(self) -> RouteSettings: + return RouteSettings(self._native.edge) + @property def edge(self) -> EdgeContract: - return EdgeContract(self._native.edge) + return self.route_settings @property def configuration(self) -> Mapping[str, ConnectorConfigurationValue]: @@ -1230,16 +1236,21 @@ def declare( self, configuration: ConnectorConfigurationInput = (), *, + route_settings: RouteSettings | None = None, edge: EdgeContract | None = None, ) -> Endpoint: """Declare one configured endpoint using the registered implementation.""" native_configuration = self._connector.manifest.configuration.configuration( configuration ) - selected_edge = edge or _default_edge(self._connector.manifest) + selected_settings = _select_route_settings( + route_settings, + edge, + _default_route_settings(self._connector.manifest), + ) native = _native_call( lambda: self._session._native.declare_connector( - self._native, native_configuration, selected_edge._native + self._native, native_configuration, selected_settings._native ) ) return Endpoint(native) @@ -1309,10 +1320,10 @@ def _first_duplicate(values: Iterable[str]) -> str | None: return None -def _default_edge(manifest: ConnectorManifest) -> EdgeContract: +def _default_route_settings(manifest: ConnectorManifest) -> RouteSettings: if len(manifest.inputs) == 1 and manifest.inputs[0].signal.is_audio: - return EdgeContract.realtime_audio() - return EdgeContract.bounded_async() + return RouteSettings.realtime_audio() + return RouteSettings.bounded_async() __all__ = [ diff --git a/python/pocketstation/conversation.py b/python/pocketstation/conversation.py index dd0d3a7..5600391 100644 --- a/python/pocketstation/conversation.py +++ b/python/pocketstation/conversation.py @@ -1,6 +1,6 @@ """Compatibility imports for the former conversation module. -New code should import these contracts from :mod:`pocketstation.voice`. +New code should import these protocols from :mod:`pocketstation.voice`. """ from .voice import ( diff --git a/python/pocketstation/endpoint_authoring.py b/python/pocketstation/endpoint_authoring.py index cf6bac9..681614c 100644 --- a/python/pocketstation/endpoint_authoring.py +++ b/python/pocketstation/endpoint_authoring.py @@ -1,4 +1,4 @@ -"""Advanced Python projection of Core's generic Endpoint lifecycle.""" +"""Advanced Python API for Core's generic Endpoint lifecycle.""" from __future__ import annotations @@ -35,7 +35,15 @@ _RegisteredEndpoint as _NativeRegisteredEndpoint, ) from .errors import PocketStationError, _native_call -from .graph import EdgeContract, Endpoint, MediaCaps, PortSpec, SignalSpec +from .graph import ( + EdgeContract, + Endpoint, + MediaCaps, + PortSpec, + RouteSettings, + SignalSpec, + _select_route_settings, +) from .observations import EndpointFailureRetryability, EndpointFailureStage from .signal import SignalEnvelope @@ -237,7 +245,7 @@ def mark_worker_failure(self) -> None: class EndpointPortInput: - """One compiled input port, receiver, route identity, and edge contract.""" + """One compiled input port, receiver, route identity, and settings.""" __slots__ = ("_native",) @@ -256,9 +264,13 @@ def signal(self) -> SignalSpec[object]: def media(self) -> MediaCaps: return MediaCaps._from_native(self._native.media) + @property + def route_settings(self) -> RouteSettings: + return RouteSettings(self._native.edge) + @property def edge(self) -> EdgeContract: - return EdgeContract(self._native.edge) + return self.route_settings @property def context(self) -> EndpointPrepareContext: @@ -421,13 +433,18 @@ def declare( self, configuration: EndpointConfigurationInput = (), *, + route_settings: RouteSettings | None = None, edge: EdgeContract | None = None, ) -> Endpoint: values = _configuration(configuration) - selected_edge = edge or _default_edge(self._provider.manifest) + selected_settings = _select_route_settings( + route_settings, + edge, + _default_route_settings(self._provider.manifest), + ) native = _native_call( lambda: self._session._native.declare_registered_endpoint( - self._native, values, selected_edge._native + self._native, values, selected_settings._native ) ) return Endpoint(native) @@ -449,10 +466,10 @@ def _configuration(values: EndpointConfigurationInput) -> dict[str, str]: return result -def _default_edge(manifest: EndpointManifest) -> EdgeContract: +def _default_route_settings(manifest: EndpointManifest) -> RouteSettings: if len(manifest.inputs) == 1 and manifest.inputs[0].signal.is_audio: - return EdgeContract.realtime_audio() - return EdgeContract.bounded_async() + return RouteSettings.realtime_audio() + return RouteSettings.bounded_async() __all__ = [ diff --git a/python/pocketstation/errors.py b/python/pocketstation/errors.py index ed93b58..160d0d7 100644 --- a/python/pocketstation/errors.py +++ b/python/pocketstation/errors.py @@ -29,7 +29,7 @@ class SessionDeclarationError(SessionError, ValueError): @dataclass(frozen=True, slots=True) class SessionCompileDiagnostic: - """Machine-readable location and contract facts for a compile failure.""" + """Machine-readable location and validation details for a compile failure.""" code: str node_index: int | None = None @@ -67,15 +67,15 @@ class CaptureError(SessionStartError): class GraphError(PocketStationError, ValueError): - """A graph, signal, port, edge, or media contract is invalid.""" + """A graph, signal, port, edge, or media declaration is invalid.""" class SourceError(PocketStationError): - """An externally authored Source contract or lifecycle failed.""" + """An externally authored Source failed validation or lifecycle work.""" class OperatorError(PocketStationError): - """An externally authored Operator contract or lifecycle failed.""" + """An externally authored Operator failed validation or lifecycle work.""" class ConnectorRuntimeError(PocketStationError): @@ -118,7 +118,7 @@ class SidecarBackpressureError(SidecarError): class SidecarProtocolError(SidecarError): - """The child violated the frozen PKSS wire or handshake contract.""" + """The child violated the frozen PKSS wire protocol or handshake.""" class SidecarTimeoutError(SidecarError): @@ -126,7 +126,7 @@ class SidecarTimeoutError(SidecarError): class ExtensionError(PocketStationError): - """A compiled extension descriptor or ABI contract was rejected.""" + """A compiled extension descriptor or ABI requirement was rejected.""" class AudioInputError(PocketStationError): diff --git a/python/pocketstation/extensions.py b/python/pocketstation/extensions.py index fd3dea6..6004d94 100644 --- a/python/pocketstation/extensions.py +++ b/python/pocketstation/extensions.py @@ -30,7 +30,7 @@ class ExtensionPortDirection(StrEnum): @dataclass(frozen=True, slots=True) class ExtensionAbiVersion: - """Native authority for the linked PocketStation extension ABI.""" + """Loaded native library that implements the PocketStation extension ABI.""" struct_size_bytes: int abi_major: int diff --git a/python/pocketstation/graph.py b/python/pocketstation/graph.py index ea1bf85..ef948b0 100644 --- a/python/pocketstation/graph.py +++ b/python/pocketstation/graph.py @@ -95,7 +95,7 @@ class BinaryFormat(StrEnum): @dataclass(frozen=True, slots=True) class SignalSpec(Generic[_PayloadT_co]): - """Stable language-neutral signal identity, role, and schema contract.""" + """Stable language-neutral signal identity, role, and schema.""" kind: SignalKind format: SignalFormat | None = None @@ -348,7 +348,7 @@ def any(cls) -> MediaCaps: @classmethod def for_signal(cls, signal: SignalSpec[object]) -> MediaCaps: - """Select the wildcard media contract for a signal.""" + """Select wildcard media requirements for a signal.""" if signal.kind is SignalKind.PCM_AUDIO: return cls.audio() if signal.kind is SignalKind.ENCODED_AUDIO: @@ -374,7 +374,7 @@ def is_compatible_with(self, other: MediaCaps) -> bool: return self._native.is_compatible_with(other._native) def negotiate(self, other: MediaCaps) -> MediaCaps | None: - """Return Core's narrow compatible media contract, if one exists.""" + """Return Core's narrow compatible media requirements, if any.""" native = self._native.negotiate(other._native) return None if native is None else type(self)._from_native(native) @@ -427,7 +427,7 @@ def input( multiplicity: Multiplicity = Multiplicity.ONE, required: bool = True, ) -> PortSpec: - """Declare an input port, inferring the normal media contract.""" + """Declare an input port and infer its normal media requirements.""" return cls( name, PortDirection.INPUT, @@ -447,7 +447,7 @@ def output( multiplicity: Multiplicity = Multiplicity.ONE, required: bool = True, ) -> PortSpec: - """Declare an output port, inferring the normal media contract.""" + """Declare an output port and infer its normal media requirements.""" return cls( name, PortDirection.OUTPUT, @@ -495,7 +495,7 @@ class CopyPolicy(StrEnum): COPY_TO_BRANCH_POOL = "copy-to-branch-pool" -class EdgeObservabilityLevel(StrEnum): +class RouteObservability(StrEnum): OFF = "off" COUNTERS = "counters" FULL = "full" @@ -503,24 +503,115 @@ class EdgeObservabilityLevel(StrEnum): @property def rank(self) -> int: return { - EdgeObservabilityLevel.OFF: 0, - EdgeObservabilityLevel.COUNTERS: 1, - EdgeObservabilityLevel.FULL: 2, + RouteObservability.OFF: 0, + RouteObservability.COUNTERS: 1, + RouteObservability.FULL: 2, }[self] -@dataclass(frozen=True, slots=True) -class EdgeContract: - """Configure a bounded edge with the public Rust policy modifiers.""" +EdgeObservabilityLevel = RouteObservability + + +@dataclass(frozen=True, slots=True, eq=False) +class DeliveryPolicy: + """Choose how a route behaves when delivery slows or fails.""" _native: _NativeEdgeContract = field(repr=False, compare=False) @classmethod - def realtime_audio(cls) -> EdgeContract: + def realtime_audio(cls) -> DeliveryPolicy: return cls(_NativeEdgeContract.realtime_audio()) @classmethod - def bounded_async(cls) -> EdgeContract: + def bounded_async(cls) -> DeliveryPolicy: + return cls(_NativeEdgeContract.bounded_async()) + + @property + def clock(self) -> ClockDomain: + return ClockDomain(self._native.clock) + + @property + def latency_budget_ms(self) -> int | None: + return self._native.latency_budget_ms + + @property + def jitter_budget_ms(self) -> int | None: + return self._native.jitter_budget_ms + + @property + def backpressure(self) -> BackpressurePolicy: + return BackpressurePolicy(self._native.backpressure) + + @property + def delivery(self) -> DeliverySemantics: + return DeliverySemantics(self._native.delivery) + + @property + def loss(self) -> LossPolicy: + return LossPolicy(self._native.loss) + + @property + def copy_policy(self) -> CopyPolicy: + return CopyPolicy(self._native.copy_policy) + + @property + def observability(self) -> RouteObservability: + return RouteObservability(self._native.observability) + + @property + def max_payload_bytes(self) -> int | None: + return self._native.max_payload_bytes + + def with_backpressure(self, policy: BackpressurePolicy) -> DeliveryPolicy: + return type(self)( + _native_call(lambda: self._native.with_backpressure(policy.value)) + ) + + def with_copy_policy(self, policy: CopyPolicy) -> DeliveryPolicy: + return type(self)( + _native_call(lambda: self._native.with_copy_policy(policy.value)) + ) + + def with_jitter_budget_ms(self, budget_ms: int | None) -> DeliveryPolicy: + return type(self)(self._native.with_jitter_budget_ms(budget_ms)) + + def with_max_payload_bytes(self, maximum_bytes: int) -> DeliveryPolicy: + return type(self)( + _native_call(lambda: self._native.with_max_payload_bytes(maximum_bytes)) + ) + + def _values(self) -> tuple[object, ...]: + return ( + self.clock, + self.latency_budget_ms, + self.jitter_budget_ms, + self.backpressure, + self.delivery, + self.loss, + self.copy_policy, + self.observability, + self.max_payload_bytes, + ) + + def __eq__(self, other: object) -> bool: + return isinstance(other, DeliveryPolicy) and self._values() == other._values() + + def __hash__(self) -> int: + return hash(self._values()) + + +@dataclass(frozen=True, slots=True, eq=False) +class RouteSettings: + """Choose the media accepted by a route and how that route delivers it.""" + + _native: _NativeEdgeContract = field(repr=False, compare=False) + + @classmethod + def realtime_audio(cls) -> RouteSettings: + return cls(_NativeEdgeContract.realtime_audio()) + + @classmethod + def bounded_async(cls) -> RouteSettings: return cls(_NativeEdgeContract.bounded_async()) @property @@ -556,34 +647,63 @@ def copy_policy(self) -> CopyPolicy: return CopyPolicy(self._native.copy_policy) @property - def observability(self) -> EdgeObservabilityLevel: - return EdgeObservabilityLevel(self._native.observability) + def observability(self) -> RouteObservability: + return RouteObservability(self._native.observability) @property def max_payload_bytes(self) -> int | None: return self._native.max_payload_bytes - def with_media(self, media: MediaCaps) -> EdgeContract: + @property + def delivery_policy(self) -> DeliveryPolicy: + return DeliveryPolicy(self._native) + + def with_media(self, media: MediaCaps) -> RouteSettings: return type(self)(self._native.with_media(media._native)) - def with_backpressure(self, policy: BackpressurePolicy) -> EdgeContract: + def with_delivery_policy(self, policy: DeliveryPolicy) -> RouteSettings: + return type(self)(policy._native.with_media(self.media._native)) + + def with_backpressure(self, policy: BackpressurePolicy) -> RouteSettings: return type(self)( _native_call(lambda: self._native.with_backpressure(policy.value)) ) - def with_copy_policy(self, policy: CopyPolicy) -> EdgeContract: + def with_copy_policy(self, policy: CopyPolicy) -> RouteSettings: return type(self)( _native_call(lambda: self._native.with_copy_policy(policy.value)) ) - def with_jitter_budget_ms(self, budget_ms: int | None) -> EdgeContract: + def with_jitter_budget_ms(self, budget_ms: int | None) -> RouteSettings: return type(self)(self._native.with_jitter_budget_ms(budget_ms)) - def with_max_payload_bytes(self, maximum_bytes: int) -> EdgeContract: + def with_max_payload_bytes(self, maximum_bytes: int) -> RouteSettings: return type(self)( _native_call(lambda: self._native.with_max_payload_bytes(maximum_bytes)) ) + def _values(self) -> tuple[object, ...]: + return (self.media, self.delivery_policy) + + def __eq__(self, other: object) -> bool: + return isinstance(other, RouteSettings) and self._values() == other._values() + + def __hash__(self) -> int: + return hash(self._values()) + + +EdgeContract: TypeAlias = RouteSettings + + +def _select_route_settings( + route_settings: RouteSettings | None, + edge: EdgeContract | None, + default: RouteSettings, +) -> RouteSettings: + if route_settings is not None and edge is not None: + raise ValueError("pass route_settings or edge, not both") + return route_settings or edge or default + ConfigurationInput: TypeAlias = Mapping[str, str] | Iterable[tuple[str, str]] @@ -658,6 +778,10 @@ class EndpointDescriptor: input_edge: EdgeContract | None = None _native: _NativeEndpointDescriptor = field(init=False, repr=False, compare=False) + @property + def route_settings(self) -> RouteSettings | None: + return self.input_edge + def __post_init__(self) -> None: native = _native_call( lambda: _NativeEndpointDescriptor( @@ -767,8 +891,8 @@ def send_to( This is the concise one-destination form. Use ``Session.register_connector(...).declare(...)`` when one Connector - implementation needs multiple configurations or explicit edge - contracts. + implementation needs multiple configurations or explicit + route settings. """ return self.send( self._destination(connector), @@ -805,7 +929,7 @@ def through( class Stem(_RoutableStream): - """Independent source-aware PCM path declared on a Session.""" + """Independent source-aware PCM stream declared on a Session.""" __slots__ = ("_native",) _native: _NativeStem @@ -1019,7 +1143,7 @@ def connector( ) def browser(self, receiver_uri: str) -> Endpoint: - """Declare the frozen browser/remote receiver endpoint contract.""" + """Declare the frozen browser or remote-receiver Endpoint.""" return _native_call(lambda: Endpoint(self._native.browser(receiver_uri))) def subscribe( @@ -1027,6 +1151,7 @@ def subscribe( stream: DerivedStream | SourceOutput, *, signal: SignalSpec[_PayloadT], + route_settings: RouteSettings | None = None, edge: EdgeContract | None = None, ) -> BusSubscription[_PayloadT]: """Declare one bounded, exclusive typed-signal subscription. @@ -1036,17 +1161,17 @@ def subscribe( """ from .signal import BusSubscription - contract = ( - EdgeContract.bounded_async().with_media(_media_for_signal(signal)) - if edge is None - else edge + settings = _select_route_settings( + route_settings, + edge, + RouteSettings.bounded_async().with_media(_media_for_signal(signal)), ) if isinstance(stream, DerivedStream): native = _native_call( lambda: self._native.subscribe_derived( stream._native, signal._native, - contract._native, + settings._native, ) ) elif isinstance(stream, SourceOutput): @@ -1054,7 +1179,7 @@ def subscribe( lambda: self._native.subscribe_source_output( stream._native, signal._native, - contract._native, + settings._native, ) ) else: @@ -1117,6 +1242,7 @@ def _media_for_signal(signal: SignalSpec[object]) -> MediaCaps: "ClockDomain", "Codec", "CopyPolicy", + "DeliveryPolicy", "DeliverySemantics", "DerivedStream", "EdgeContract", @@ -1135,6 +1261,8 @@ def _media_for_signal(signal: SignalSpec[object]) -> MediaCaps: "OperatorInstance", "PortDirection", "PortSpec", + "RouteObservability", + "RouteSettings", "SampleFormat", "SignalKind", "SignalSpec", diff --git a/python/pocketstation/observations.py b/python/pocketstation/observations.py index ae4aa7c..7c614c2 100644 --- a/python/pocketstation/observations.py +++ b/python/pocketstation/observations.py @@ -147,10 +147,15 @@ class RouteObservationInterval(StrEnum): ROUTE_LIFETIME_TO_SNAPSHOT = "route-lifetime-to-snapshot" -class RouteLatencyBoundary(StrEnum): +class RouteLatencyMeasurement(StrEnum): + """Identify the timestamps used to calculate route latency.""" + SOURCE_TIMESTAMP_TO_ROUTE_RECEIVE = "source-monotonic-timestamp-to-route-receive" +RouteLatencyBoundary = RouteLatencyMeasurement + + class RouteLatencyUnit(StrEnum): NANOSECONDS = "nanoseconds" @@ -266,7 +271,7 @@ def _from_native(cls, failure: _NativeSessionFailure) -> SessionFailure: @dataclass(frozen=True, slots=True) class SessionEvent: - """One immutable projection of an authoritative native Session event.""" + """One immutable Python view of a native Session event.""" kind: SessionEventType lifecycle_state: SessionLifecycleState | None @@ -362,7 +367,7 @@ class LatencyHistogram: @dataclass(frozen=True, slots=True) -class EdgeMetrics: +class RouteDeliveryMetrics: queue_capacity_frames: int queue_depth_frames: int queue_peak_frames: int @@ -389,7 +394,7 @@ class EdgeMetrics: discarded_output_frames_total: int | None @classmethod - def _from_native(cls, value: _NativeEdgeMetrics) -> EdgeMetrics: + def _from_native(cls, value: _NativeEdgeMetrics) -> RouteDeliveryMetrics: return cls( queue_capacity_frames=value.queue_capacity_frames, queue_depth_frames=value.queue_depth_frames, @@ -436,6 +441,9 @@ def _from_native(cls, value: _NativeEdgeMetrics) -> EdgeMetrics: ) +EdgeMetrics = RouteDeliveryMetrics + + @dataclass(frozen=True, slots=True) class EndpointMetrics: observation_stage: EndpointObservationStage @@ -451,18 +459,22 @@ class EndpointMetrics: class RouteMetrics: route_id: int endpoint_id: int - edge: EdgeMetrics + edge: RouteDeliveryMetrics endpoint: EndpointMetrics frames_attempted_total: int observation_interval: RouteObservationInterval drop_rate_pct: float - source_latency_boundary: RouteLatencyBoundary + source_latency_boundary: RouteLatencyMeasurement source_latency_unit: RouteLatencyUnit @property def queue_capacity_frames(self) -> int: return self.edge.queue_capacity_frames + @property + def delivery(self) -> RouteDeliveryMetrics: + return self.edge + @property def frames_delivered_total(self) -> int: return self.edge.frames_delivered_total @@ -471,9 +483,13 @@ def frames_delivered_total(self) -> int: def frames_dropped_total(self) -> int: return self.edge.frames_dropped_total + @property + def source_latency_measurement(self) -> RouteLatencyMeasurement: + return self.source_latency_boundary + @classmethod def _from_native(cls, value: _NativeRouteMetrics) -> RouteMetrics: - edge = EdgeMetrics._from_native(cast(_NativeEdgeMetrics, value)) + edge = RouteDeliveryMetrics._from_native(cast(_NativeEdgeMetrics, value)) return cls( route_id=value.route_id, endpoint_id=value.endpoint_id, @@ -494,7 +510,9 @@ def _from_native(cls, value: _NativeRouteMetrics) -> RouteMetrics: value.drop_observation_interval ), drop_rate_pct=value.drop_rate_pct, - source_latency_boundary=RouteLatencyBoundary(value.source_latency_boundary), + source_latency_boundary=RouteLatencyMeasurement( + value.source_latency_boundary + ), source_latency_unit=RouteLatencyUnit(value.source_latency_unit), ) @@ -582,7 +600,7 @@ def _from_native(cls, value: _NativeExternalSourceMetrics) -> ExternalSourceMetr @dataclass(frozen=True, slots=True) -class TypedEdgeMetrics: +class SignalQueueMetrics: capacity_signals: int max_payload_bytes: int maximum_buffered_payload_bytes: int @@ -593,10 +611,13 @@ class TypedEdgeMetrics: dropped_total: int @classmethod - def _from_native(cls, value: _NativeTypedEdgeMetrics) -> TypedEdgeMetrics: + def _from_native(cls, value: _NativeTypedEdgeMetrics) -> SignalQueueMetrics: return cls(**{name: getattr(value, name) for name in cls.__dataclass_fields__}) +TypedEdgeMetrics = SignalQueueMetrics + + @dataclass(frozen=True, slots=True) class OperatorWorkerMetrics: input_attempted_total: int @@ -622,26 +643,37 @@ def _from_native(cls, value: _NativeOperatorWorkerMetrics) -> OperatorWorkerMetr @dataclass(frozen=True, slots=True) class OperatorInputMetrics: port_name: str - edge: EdgeMetrics + edge: RouteDeliveryMetrics + + @property + def delivery(self) -> RouteDeliveryMetrics: + return self.edge @classmethod def _from_native(cls, value: _NativeOperatorInputMetrics) -> OperatorInputMetrics: - return cls(port_name=value.port_name, edge=EdgeMetrics._from_native(value.edge)) + return cls( + port_name=value.port_name, + edge=RouteDeliveryMetrics._from_native(value.edge), + ) @dataclass(frozen=True, slots=True) class OperatorMetrics: operator_instance_id: int - input_edge: EdgeMetrics + input_edge: RouteDeliveryMetrics worker: OperatorWorkerMetrics finalization_failures_total: int input_ports: tuple[OperatorInputMetrics, ...] + @property + def input_delivery(self) -> RouteDeliveryMetrics: + return self.input_edge + @classmethod def _from_native(cls, value: _NativeOperatorMetrics) -> OperatorMetrics: return cls( operator_instance_id=value.operator_instance_id, - input_edge=EdgeMetrics._from_native(value.input_edge), + input_edge=RouteDeliveryMetrics._from_native(value.input_edge), worker=OperatorWorkerMetrics._from_native(value.worker), finalization_failures_total=value.finalization_failures_total, input_ports=tuple( @@ -654,7 +686,7 @@ def _from_native(cls, value: _NativeOperatorMetrics) -> OperatorMetrics: class DerivedRouteMetrics: route_id: int endpoint_id: int - output: TypedEdgeMetrics + output: SignalQueueMetrics endpoint: EndpointMetrics @classmethod @@ -662,7 +694,7 @@ def _from_native(cls, value: _NativeDerivedRouteMetrics) -> DerivedRouteMetrics: return cls( route_id=value.route_id, endpoint_id=value.endpoint_id, - output=TypedEdgeMetrics._from_native(value.output), + output=SignalQueueMetrics._from_native(value.output), endpoint=EndpointMetrics( observation_stage=EndpointObservationStage( value.endpoint_observation_stage @@ -1246,7 +1278,9 @@ def iterate() -> Iterator[SessionEvent]: "RecordingState", "RecordingStemOutcome", "RelayPublishOutcome", + "RouteDeliveryMetrics", "RouteLatencyBoundary", + "RouteLatencyMeasurement", "RouteLatencyUnit", "RouteMetrics", "RouteObservationInterval", @@ -1267,6 +1301,7 @@ def iterate() -> Iterator[SessionEvent]: "SessionTraceRecordType", "SessionTraceRecorderOutcome", "SessionTraceValidation", + "SignalQueueMetrics", "SourceMetrics", "StopResult", "TerminationDisposition", diff --git a/python/pocketstation/operator_authoring.py b/python/pocketstation/operator_authoring.py index 9d414df..2cf8a53 100644 --- a/python/pocketstation/operator_authoring.py +++ b/python/pocketstation/operator_authoring.py @@ -20,6 +20,7 @@ OperatorInstance, PortDirection, PortSpec, + RouteSettings, SignalSpec, ) from .signal import SignalAudioPayload, SignalEnvelope @@ -31,7 +32,7 @@ def operator(self, operator: Operator) -> OperatorInstance: ... @dataclass(frozen=True, slots=True) class OperatorManifest: - """Validated contract for one off-realtime Python Operator.""" + """Validated definition for one off-realtime Python Operator.""" operator_id: str inputs: tuple[PortSpec, ...] @@ -77,6 +78,10 @@ class OperatorPortContext: media: MediaCaps edge: EdgeContract + @property + def route_settings(self) -> RouteSettings: + return self.edge + @classmethod def _from_native(cls, value: _NativeOperatorPortContext) -> OperatorPortContext: return cls( @@ -148,7 +153,7 @@ class OperatorNode: """Off-realtime computation hosted by Core's Operator worker.""" def prepare(self, context: OperatorPrepareContext) -> None: - """Observe compiled port and edge contracts before processing.""" + """Observe compiled ports and route settings before processing.""" def process( self, input_port: str, envelope: SignalEnvelope[object] diff --git a/python/pocketstation/session.py b/python/pocketstation/session.py index f06f242..dd2b909 100644 --- a/python/pocketstation/session.py +++ b/python/pocketstation/session.py @@ -31,6 +31,7 @@ from .graph import ( EdgeContract, Endpoint, + RouteSettings, Stem, _GraphSessionDeclarations, ) @@ -386,6 +387,7 @@ def destination( connector: Connector, configuration: ConnectorConfigurationInput = (), *, + route_settings: RouteSettings | None = None, edge: EdgeContract | None = None, ) -> Endpoint: """Declare one Connector destination using an idempotent registration. @@ -394,7 +396,11 @@ def destination( :meth:`register_connector` remains available when one implementation must declare several independently configured Endpoints. """ - return self.register_connector(connector).declare(configuration, edge=edge) + return self.register_connector(connector).declare( + configuration, + route_settings=route_settings, + edge=edge, + ) def register_endpoint(self, endpoint: EndpointProvider) -> RegisteredEndpoint: """Register one advanced Python implementation of Core's Endpoint SPI. diff --git a/python/pocketstation/signal.py b/python/pocketstation/signal.py index c6a1398..46f09dc 100644 --- a/python/pocketstation/signal.py +++ b/python/pocketstation/signal.py @@ -13,7 +13,7 @@ from ._native import _SignalLineage as _NativeSignalLineage from ._native import _SignalSubscriptionMetrics as _NativeSignalSubscriptionMetrics from ._native import _SignalTiming as _NativeSignalTiming -from .graph import EdgeContract, SignalSpec +from .graph import EdgeContract, RouteSettings, SignalSpec from .identity import ( ClockDomainId, ConnectorId, @@ -243,9 +243,13 @@ def signal(self) -> SignalSpec[_PayloadT_co]: SignalSpec[_PayloadT_co], SignalSpec._from_native(self._native.signal) ) + @property + def route_settings(self) -> RouteSettings: + return RouteSettings(self._native.edge) + @property def edge(self) -> EdgeContract: - return EdgeContract(self._native.edge) + return self.route_settings class EndOfStream: diff --git a/python/pocketstation/source_authoring.py b/python/pocketstation/source_authoring.py index d46ea16..aa15dd3 100644 --- a/python/pocketstation/source_authoring.py +++ b/python/pocketstation/source_authoring.py @@ -28,7 +28,7 @@ def source( @dataclass(frozen=True, slots=True) class SourceManifest: - """Stable contract for one Python-authored typed Source implementation.""" + """Interface for one Python-authored typed Source implementation.""" source_type_id: str outputs: tuple[PortSpec, ...] @@ -232,7 +232,7 @@ def create(self, configuration: Mapping[str, str]) -> SourceDriver: @dataclass(frozen=True, slots=True) class SourceProvider: - """Reusable Python implementation of one Core Source contract.""" + """Reusable Python implementation of one Core Source.""" manifest: SourceManifest factory: SourceFactory diff --git a/python/pocketstation/sources.py b/python/pocketstation/sources.py index 82ecfa0..f06980a 100644 --- a/python/pocketstation/sources.py +++ b/python/pocketstation/sources.py @@ -479,7 +479,7 @@ def from_discovered(cls, source: DiscoveredSource) -> Source: """Build the strongest supported Session declaration from discovery. Output devices and system mix remain discovery-only in the stable 1.1 - Session declaration contract. + Session declaration requirements. """ stable_id = source.stable_id if stable_id.kind is SourceKind.APPLICATION: diff --git a/python/pocketstation/voice/__init__.py b/python/pocketstation/voice/__init__.py index 6149262..e4e63a7 100644 --- a/python/pocketstation/voice/__init__.py +++ b/python/pocketstation/voice/__init__.py @@ -1,7 +1,7 @@ -"""Provider-neutral contracts and composition for live voice applications. +"""Provider-neutral protocols and composition for live voice applications. -Provider packages implement these contracts. PocketStation continues to own -the native Session, source-aware audio paths, bounded routing, recording, +Provider packages implement these protocols. PocketStation continues to own +the native Session, source-aware audio routes, bounded routing, recording, Relay delivery, and output cancellation. """ diff --git a/python/pocketstation/voice/duplex.py b/python/pocketstation/voice/duplex.py index 8aeeea5..b27bf96 100644 --- a/python/pocketstation/voice/duplex.py +++ b/python/pocketstation/voice/duplex.py @@ -1,4 +1,4 @@ -"""Contracts for stateful providers that accept and produce live audio.""" +"""Protocols for stateful providers that accept and produce live audio.""" from __future__ import annotations diff --git a/python/pocketstation/voice/events.py b/python/pocketstation/voice/events.py index 98e2320..34591ea 100644 --- a/python/pocketstation/voice/events.py +++ b/python/pocketstation/voice/events.py @@ -7,7 +7,7 @@ @dataclass(frozen=True, slots=True) class VoiceEvent: - """One measured voice lifecycle or media-boundary event.""" + """One measured voice lifecycle or media-timing event.""" kind: str timestamp_ns: int diff --git a/python/pocketstation/voice/response.py b/python/pocketstation/voice/response.py index c19e207..3060d51 100644 --- a/python/pocketstation/voice/response.py +++ b/python/pocketstation/voice/response.py @@ -1,4 +1,4 @@ -"""Incremental response contracts for provider-neutral voice composition.""" +"""Incremental response protocols for provider-neutral voice composition.""" from __future__ import annotations diff --git a/python/pocketstation/voice/speech_detection.py b/python/pocketstation/voice/speech_detection.py index 598ff3a..da7f835 100644 --- a/python/pocketstation/voice/speech_detection.py +++ b/python/pocketstation/voice/speech_detection.py @@ -1,4 +1,4 @@ -"""Speech-activity events and detector integration contracts.""" +"""Speech-activity events and detector integration protocols.""" from __future__ import annotations @@ -19,7 +19,7 @@ @dataclass(frozen=True, slots=True) class SpeechActivity: - """One speech boundary observed from a source-aware audio stream.""" + """One speech-activity update observed from a source-aware audio stream.""" kind: SpeechActivityKind source_id: SourceId diff --git a/python/pocketstation/voice/synthesis.py b/python/pocketstation/voice/synthesis.py index f04504b..f11e115 100644 --- a/python/pocketstation/voice/synthesis.py +++ b/python/pocketstation/voice/synthesis.py @@ -1,4 +1,4 @@ -"""Streaming speech-synthesis contracts over the existing PCM input boundary.""" +"""Streaming speech-synthesis protocols that write to an existing PCM input.""" from __future__ import annotations diff --git a/python/pocketstation/voice/transcription.py b/python/pocketstation/voice/transcription.py index 2ce88c2..424bf15 100644 --- a/python/pocketstation/voice/transcription.py +++ b/python/pocketstation/voice/transcription.py @@ -1,4 +1,4 @@ -"""Streaming transcript revisions and transcriber integration contracts.""" +"""Streaming transcript revisions and transcriber integration protocols.""" from __future__ import annotations diff --git a/python/pocketstation_demo/openai_realtime.py b/python/pocketstation_demo/openai_realtime.py index 3cfe656..54e8a2f 100644 --- a/python/pocketstation_demo/openai_realtime.py +++ b/python/pocketstation_demo/openai_realtime.py @@ -95,7 +95,7 @@ def __post_init__(self) -> None: @dataclass(frozen=True, slots=True) class RealtimeVoiceObservations: - """Finite provider and PocketStation boundary counters.""" + """Finite provider and PocketStation delivery counters.""" input_ready: bool input_frames_sent: int @@ -309,7 +309,7 @@ def print_report(self) -> None: @property def observations(self) -> RealtimeVoiceObservations: - """Return measured provider and generated-audio boundary counters.""" + """Return measured provider and generated-audio delivery counters.""" if self._voice is None: raise RuntimeError("the OpenAI Realtime connection has not been declared") return self._voice.observations @@ -620,7 +620,7 @@ def outcome( self, disposition: Literal["completed", "stopped", "cancelled", "failed"], ) -> ConversationOutcome: - """Return measured provider and media-boundary facts collected so far.""" + """Return measured provider and media delivery facts collected so far.""" history = _conversation_history(self._event_records) events = tuple(_voice_event(record) for record in self._event_records) turns_started = sum(message.role == "user" for message in history) @@ -707,7 +707,7 @@ def print_report(self) -> None: " cancellation decision " f"{max(cancelled - interruption, 0) / 1_000_000:.1f} ms" ) - print(f" provider boundary {self.observations}") + print(f" provider delivery {self.observations}") print( " browser playout cutoff unavailable: the receiver has not returned " "a played-sample acknowledgement" diff --git a/tests/test_connector.py b/tests/test_connector.py index da6617f..5cd10b4 100644 --- a/tests/test_connector.py +++ b/tests/test_connector.py @@ -26,6 +26,7 @@ ConnectorShutdownMode, ConnectorWorker, PocketStationError, + RouteSettings, Session, ) from pocketstation.connector import connector @@ -226,6 +227,31 @@ def test_session_destination_reuses_one_connector_registration() -> None: assert session.register_connector(provider).session_id == session.id +def test_session_destination_accepts_route_settings_and_rejects_duplicate_names() -> ( + None +): + manifest = ConnectorManifest.audio( + "io.pocketstation.test.route-settings.v1", + package_version="1.0.0", + ) + provider = Connector.from_handler( + manifest, + lambda _item, _context: ConnectorDeliveryOutcome.DELIVERED, + ) + session = Session() + settings = RouteSettings.realtime_audio() + + endpoint = session.destination(provider, route_settings=settings) + + assert endpoint.session_id == session.id + with pytest.raises(ValueError, match="route_settings or edge"): + session.destination( + provider, + route_settings=settings, + edge=settings, + ) + + def test_session_destination_does_not_merge_different_connector_implementations() -> ( None ): diff --git a/tests/test_graph.py b/tests/test_graph.py index 993bc98..881094c 100644 --- a/tests/test_graph.py +++ b/tests/test_graph.py @@ -11,6 +11,7 @@ ClockDomain, Codec, CopyPolicy, + DeliveryPolicy, DeliverySemantics, EdgeContract, EdgeObservabilityLevel, @@ -25,6 +26,8 @@ PocketStationError, PortDirection, PortSpec, + RouteObservability, + RouteSettings, Session, SessionStartError, SignalSpec, @@ -168,11 +171,12 @@ def test_edge_presets_and_modifiers_preserve_bounded_contracts() -> None: assert realtime.delivery is DeliverySemantics.ORDERED assert realtime.loss is LossPolicy.CONCEAL_FOR_AUDIO assert realtime.copy_policy is CopyPolicy.SHARE_READ_ONLY - assert realtime.observability is EdgeObservabilityLevel.COUNTERS + assert realtime.observability is RouteObservability.COUNTERS assert realtime.max_payload_bytes is None assert realtime.clock.is_realtime assert not ClockDomain.INHERITED.is_realtime - assert EdgeObservabilityLevel.FULL.rank > realtime.observability.rank + assert EdgeObservabilityLevel is RouteObservability + assert RouteObservability.FULL.rank > realtime.observability.rank bounded = EdgeContract.bounded_async() assert bounded.clock is ClockDomain.INHERITED @@ -194,6 +198,37 @@ def test_edge_presets_and_modifiers_preserve_bounded_contracts() -> None: assert bounded.backpressure is BackpressurePolicy.BOUNDED_QUEUE +def test_route_settings_apply_delivery_policy_without_changing_media() -> None: + media = MediaCaps.audio( + AudioCaps( + sample_rate_hz=48_000, + frame_samples=480, + channel_layout=ChannelLayout.MONO, + ) + ) + delivery = ( + DeliveryPolicy.bounded_async() + .with_backpressure(BackpressurePolicy.DROP_OLDEST) + .with_copy_policy(CopyPolicy.COPY_TO_BRANCH_POOL) + .with_jitter_budget_ms(25) + .with_max_payload_bytes(4096) + ) + + settings = ( + RouteSettings.realtime_audio().with_media(media).with_delivery_policy(delivery) + ) + + assert EdgeContract is RouteSettings + assert settings.media == media + assert settings != RouteSettings.realtime_audio() + assert settings.clock is ClockDomain.INHERITED + assert delivery != DeliveryPolicy.realtime_audio() + assert settings.delivery_policy.backpressure is BackpressurePolicy.DROP_OLDEST + assert settings.delivery_policy.copy_policy is CopyPolicy.COPY_TO_BRANCH_POOL + assert settings.delivery_policy.jitter_budget_ms == 25 + assert settings.delivery_policy.max_payload_bytes == 4096 + + def test_configuration_values_are_immutable_snapshots() -> None: original = {"model": "small"} operator = OperatorConfiguration(original) diff --git a/tests/test_metrics.py b/tests/test_metrics.py index 29efca2..69544ee 100644 --- a/tests/test_metrics.py +++ b/tests/test_metrics.py @@ -7,8 +7,11 @@ import pocketstation._native as _native import pytest from pocketstation._api import ( + EdgeMetrics, EndpointObservationStage, PocketStationError, + RouteDeliveryMetrics, + RouteLatencyMeasurement, Session, Source, ) @@ -41,11 +44,18 @@ def test_metrics_preserve_bounded_source_route_and_polled_audio_truth(tmp_path) assert metrics.polled_audio.queue_capacity_frames > 0 assert metrics.event_queue.capacity_count > 0 assert all(route.edge.queue_capacity_frames > 0 for route in metrics.routes) + assert EdgeMetrics is RouteDeliveryMetrics + assert all(route.delivery is route.edge for route in metrics.routes) assert all( route.endpoint.observation_stage is EndpointObservationStage.LIVE for route in metrics.routes ) assert all(route.source_latency_unit == "nanoseconds" for route in metrics.routes) + assert all( + route.source_latency_measurement + is RouteLatencyMeasurement.SOURCE_TIMESTAMP_TO_ROUTE_RECEIVE + for route in metrics.routes + ) with pytest.raises(FrozenInstanceError): metrics.polled_audio.queue_capacity_frames = 0 diff --git a/tests/test_public_api.py b/tests/test_public_api.py index 68658b2..66b0422 100644 --- a/tests/test_public_api.py +++ b/tests/test_public_api.py @@ -34,7 +34,7 @@ def test_root_exports_are_a_small_intentional_entry_point() -> None: } -def test_advanced_contracts_are_not_duplicated_at_the_package_root() -> None: +def test_advanced_types_are_not_duplicated_at_the_package_root() -> None: assert hasattr(pocketstation, "Connector") assert not hasattr(pocketstation, "ConnectorManifest") assert not hasattr(pocketstation, "OperatorProvider")