From a7488e33d5f7a7b3cb503fa0ac2992fa7bfd3999 Mon Sep 17 00:00:00 2001 From: Raphael Avocegamou Date: Thu, 3 Sep 2026 08:28:37 +0200 Subject: [PATCH 1/9] api: use one route settings vocabulary --- README.md | 28 ++++---- docs/README.md | 2 +- docs/concepts/route-settings.md | 6 -- docs/concepts/session-and-bounds.md | 8 +-- docs/guides/application-audio.md | 2 +- docs/guides/integrations.md | 6 +- docs/guides/process-audio-and-signals.md | 2 +- docs/guides/record-and-observe.md | 3 +- docs/guides/voice.md | 2 +- docs/reference/api-map.md | 4 +- docs/reference/events-and-errors.md | 3 - docs/troubleshooting.md | 7 +- examples/README.md | 10 +-- native/src/connector/driver.rs | 20 +++--- native/src/connector/values.rs | 4 +- native/src/connector/worker.rs | 10 +-- native/src/endpoint_authoring.rs | 26 ++++---- native/src/graph.rs | 50 +++++++-------- native/src/observations.rs | 30 ++++----- native/src/operator_authoring/driver.rs | 16 ++--- native/src/operator_authoring/values.rs | 18 +++--- native/src/session.rs | 47 ++++---------- native/src/signals.rs | 34 +++++----- native/src/source_authoring/values.rs | 4 +- python/pocketstation/_api.py | 10 --- python/pocketstation/_native.pyi | 46 ++++++------- python/pocketstation/aio/connector.py | 4 +- .../pocketstation/aio/endpoint_authoring.py | 4 +- python/pocketstation/aio/session.py | 3 - python/pocketstation/compatibility.py | 2 +- python/pocketstation/connector.py | 9 +-- python/pocketstation/endpoint_authoring.py | 9 +-- python/pocketstation/graph.py | 48 +++----------- python/pocketstation/observations.py | 64 ++++++------------- python/pocketstation/operator_authoring.py | 9 +-- python/pocketstation/session.py | 3 - python/pocketstation/signal.py | 8 +-- python/pocketstation/voice/conversation.py | 6 +- tests/qualification/runtime_resources.py | 4 +- tests/test_connector.py | 8 +-- tests/test_conversation_interruptions.py | 5 +- tests/test_graph.py | 29 +++++---- tests/test_metrics.py | 8 +-- tests/test_signal_streams.py | 8 ++- 44 files changed, 254 insertions(+), 375 deletions(-) diff --git a/README.md b/README.md index f8ceb45..ba0b9db 100644 --- a/README.md +++ b/README.md @@ -80,12 +80,12 @@ 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 these jobs concurrently: +The Session preserves each source while the transcriber processes both: ```text -voice application ─┐ - ├─ one bounded faster-whisper Operator ─ transcripts -physical microphone┘ +voice application ──┐ + ├─ PocketStation Session ─ faster-whisper ─ source-aware transcripts +physical microphone ┘ ``` The complete composition is visible in @@ -131,9 +131,9 @@ with pocketstation.capture( print(frame.source_id, frame.stem_id) ``` -The iterator reads a bounded native endpoint. A slow Python consumer produces -observable pressure and discontinuities; it does not create an unbounded Python -audio queue. +The iterator reads frames from a native Endpoint. If Python stops reading, the +Endpoint reports its queue depth, dropped frames, and discontinuities instead +of allowing memory use to grow without a limit. ## Send application-owned audio into a Session @@ -178,7 +178,7 @@ application.send_to(destination) 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: +route delivery, lineage, observations, drain, abort, and joined shutdown: ```python class WebSocketConnector(pks_aio.Connector): @@ -208,7 +208,7 @@ PocketStation calls `start()` once, interleaves both source-aware stems through 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 +Python provider callbacks execute on off-realtime workers. They cannot be used as native capture callbacks. Use compiled native extensions for native provider code, or a process sidecar when crash isolation is required. @@ -220,8 +220,8 @@ The shared Rust `pocketstation-relay` connector publishes media. The Go Relay service forwards WebRTC audio. Python does not encode Opus, write RTP, or own a second media plane. -The control client uses finite request deadlines, bounded response bodies, -redacted secrets, and matching synchronous and asyncio APIs. +The control client limits request duration and response size, redacts secrets, +and provides matching synchronous and asyncio APIs. ## Sync and asyncio @@ -243,7 +243,7 @@ does not have the same execution cost as Rust. | 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 | -The native binding uses PocketStation Core `1.1.4` and the shared Relay +The native binding uses PocketStation Core `1.1.5` and the shared Relay Connector `0.1.2`. The Rust-to-Python audio read currently copies native samples into Python-owned @@ -266,8 +266,8 @@ uv run mypy python tests/qualification/typing_contract.py examples guidance. - [`docs/README.md`](docs/README.md) — task guides, concepts, operations, and API ownership. -- [Write application-owned audio](docs/guides/application-audio.md) — bounded - PCM input and selective output cancellation. +- [Write application-owned audio](docs/guides/application-audio.md) — PCM input + with explicit queue capacity 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 diff --git a/docs/README.md b/docs/README.md index b14c0f3..b8ad4d9 100644 --- a/docs/README.md +++ b/docs/README.md @@ -15,7 +15,7 @@ need the lower-level API. - [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) +- [Compose a voice workflow](guides/voice.md) - [Publish a named AudioBus through Relay](guides/relay.md) - [Create a Source, Operator, Connector, or Endpoint](guides/integrations.md) diff --git a/docs/concepts/route-settings.md b/docs/concepts/route-settings.md index e2399ac..e143431 100644 --- a/docs/concepts/route-settings.md +++ b/docs/concepts/route-settings.md @@ -64,9 +64,3 @@ Advanced Connector and Endpoint preparation objects expose `route_settings`. 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 49f1a44..09b90ca 100644 --- a/docs/concepts/session-and-bounds.md +++ b/docs/concepts/session-and-bounds.md @@ -8,7 +8,7 @@ captures and routes audio through finite queues. application ─┐ microphone ──┼─ Session ─┬─ Python or native Operator owned PCM ───┘ ├─ Connector or Endpoint - ├─ bounded frame iterator + ├─ frame iterator └─ multistem recording ``` @@ -25,14 +25,14 @@ 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`. -A slow Python consumer can still lose frames at its own bounded endpoint. -Inspect the route counters and discontinuities whenever complete delivery +A slow Python consumer can still lose frames when its route queue fills. +Inspect the queue depth, dropped-frame counters, and discontinuities whenever complete delivery matters. ## Python does not run on capture callbacks Python-authored Sources, Operators, Connectors, and Endpoints execute on -bounded off-realtime workers. Native capture callbacks remain allocation-free, +off-realtime workers. Their queues have configured capacities. 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 diff --git a/docs/guides/application-audio.md b/docs/guides/application-audio.md index eeafba7..649c3d9 100644 --- a/docs/guides/application-audio.md +++ b/docs/guides/application-audio.md @@ -63,4 +63,4 @@ playback device. A Connector and receiver need their own clear operation and playout acknowledgement before an application can claim audible cancellation. Continue with [multistem recording and observations](record-and-observe.md) or -[bounded voice composition](voice.md). +[voice composition](voice.md). diff --git a/docs/guides/integrations.md b/docs/guides/integrations.md index 4dac8d2..73ed980 100644 --- a/docs/guides/integrations.md +++ b/docs/guides/integrations.md @@ -29,7 +29,7 @@ capture, buffering, routing, and shutdown in every integration. ```text application ─┐ -microphone ──┼→ independent bounded Session routes +microphone ──┼→ independent Session routes generated ───┘ ↓ one Connector lifecycle ↓ @@ -101,7 +101,7 @@ microphone.send_to(destination) ``` One Connector object is one provider lifecycle. PocketStation creates two -bounded routes, calls `start()` once, preserves each frame's source and stem +routes with separate delivery queues, calls `start()` once, preserves each frame's source and stem identity, and calls `stop()` once. Two Connector objects create two independent destinations. @@ -124,7 +124,7 @@ Connector is slow or fails. | Method | Provider responsibility | PocketStation responsibility | |---|---|---| | `start()` | Open and authenticate the configured destination. | Run on the managed worker after the Session start gate, apply a finite async deadline, retain failure in the terminal outcome, and close once. | -| `send(frame)` | Encode or publish one frame without retaining it indefinitely. | Deliver off realtime from a bounded route and preserve frame lineage. | +| `send(frame)` | Encode or publish one frame without retaining it indefinitely. | Deliver off realtime from a route with a configured queue capacity and preserve frame lineage. | | `stop()` | Close sockets, files, tasks, and provider resources. | Call once after drain, abort, startup rollback, timeout, or delivery failure. | `AudioFrame` includes source, stream, stem, sequence, timestamp, clock, diff --git a/docs/guides/process-audio-and-signals.md b/docs/guides/process-audio-and-signals.md index 61344c8..b42edf4 100644 --- a/docs/guides/process-audio-and-signals.md +++ b/docs/guides/process-audio-and-signals.md @@ -4,7 +4,7 @@ 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 +A Python Operator runs on an 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. diff --git a/docs/guides/record-and-observe.md b/docs/guides/record-and-observe.md index 78a1fc9..d859e6a 100644 --- a/docs/guides/record-and-observe.md +++ b/docs/guides/record-and-observe.md @@ -50,7 +50,8 @@ a browser played the same sample. ## Stop or cancel deliberately -Leaving the context requests normal stop and drains accepted bounded work. +Leaving the context requests normal stop and drains work already accepted by +the route queues. Call `cancel()` on the explicit `RunningSession` API when active provider or sidecar work must abort. Both shutdown modes join Session workers before returning a terminal `StopResult`. diff --git a/docs/guides/voice.md b/docs/guides/voice.md index aa96ac5..753943a 100644 --- a/docs/guides/voice.md +++ b/docs/guides/voice.md @@ -1,4 +1,4 @@ -# Compose a bounded voice workflow +# Compose a voice workflow `pocketstation.voice` defines provider-neutral Python protocols. `pocketstation.aio.Session` composes those providers around one native Session. diff --git a/docs/reference/api-map.md b/docs/reference/api-map.md index 166cf06..16c0b9a 100644 --- a/docs/reference/api-map.md +++ b/docs/reference/api-map.md @@ -48,7 +48,7 @@ loss behavior. | Run a managed process | `pocketstation.sidecar` | | Load a trusted native extension | `pocketstation.extensions` | -Provider callbacks run on bounded off-realtime workers. Native capture +Provider callbacks run on off-realtime workers. Native capture callbacks never call Python. ## Build a voice workflow @@ -61,7 +61,7 @@ callbacks never call Python. - one `DuplexVoiceModel`. Provider implementations remain in example or separately installed provider -packages. Read [Compose a bounded voice workflow](../guides/voice.md) before +packages. Read [Compose a voice workflow](../guides/voice.md) before depending on interruption or playout observations. ## Handle failures diff --git a/docs/reference/events-and-errors.md b/docs/reference/events-and-errors.md index b8e8636..32ff498 100644 --- a/docs/reference/events-and-errors.md +++ b/docs/reference/events-and-errors.md @@ -44,9 +44,6 @@ 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: diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 12686a1..86ffbf5 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -1,7 +1,7 @@ # Troubleshoot capture, delivery, and shutdown Start with the component that did not produce the expected result. PocketStation -keeps source opening, bounded delivery, provider work, Relay delivery, +reports source opening, route delivery, provider work, Relay delivery, recording, and receiver playout as separate observations. ## No application audio arrives @@ -38,8 +38,9 @@ application receives. ## Python misses frames Inspect route capacity, queue depth, delivered frames, drops, and -discontinuities. The Python iterator is bounded; a consumer that does not read -on time can lose frames according to its route policy. +discontinuities. The iterator uses the route's configured queue capacity; a +consumer that does not read on time can lose frames according to its delivery +policy. Move expensive model work into an Operator or provider worker. Do not perform inference in the loop that must keep the frame Endpoint drained. diff --git a/examples/README.md b/examples/README.md index 78f169e..77c4cfa 100644 --- a/examples/README.md +++ b/examples/README.md @@ -16,15 +16,15 @@ python -m pip install 'pocketstation[voice-agent-debug]' python examples/send_audio_to_websocket.py ``` -The example asks which running application to capture. PocketStation owns the -bounded route and shutdown; the surrounding WebSocket context owns its +The example asks which running application to capture. PocketStation owns route +delivery and shutdown; the surrounding WebSocket context owns its connection. ## 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 -assistant PCM, browser delivery, independent recording stems, bounded queues, +assistant PCM, browser delivery, independent recording stems, queue limits, and sender-side output cancellation. The provider owns speech recognition and the model response. @@ -58,7 +58,7 @@ Speak, interrupt the assistant while it is replying, then press `Ctrl-C`. Use headphones for this run: the example does not provide acoustic echo cancellation. -The final report includes source continuity, bounded queue drops, provider +The final report includes source continuity, queue depth and drops, provider events, and PocketStation's output-cancellation events. The receiver reports WebRTC statistics, but it does not acknowledge the exact sample played through the loudspeaker. The example therefore reports acoustic hearing and exact @@ -88,7 +88,7 @@ python -m pip install 'pocketstation[transcription]' ``` The first run may download the configured faster-whisper model. Model work runs -on a bounded off-realtime Operator worker, not on a capture callback. +on an off-realtime Operator worker, not on a capture callback. ## Stream application audio to a browser diff --git a/native/src/connector/driver.rs b/native/src/connector/driver.rs index 721f1d4..e4cb31c 100644 --- a/native/src/connector/driver.rs +++ b/native/src/connector/driver.rs @@ -19,7 +19,7 @@ use super::values::{ PythonConnectorManifest, }; use crate::errors::coded_reason; -use crate::graph::{PythonEdgeContract, PythonEndpoint, PythonMediaCaps, PythonSignalSpec}; +use crate::graph::{PythonEndpoint, PythonMediaCaps, PythonRouteSettings, PythonSignalSpec}; use crate::signals::{copy_envelope, python_envelope, PythonSignalEnvelope}; use crate::streams::{owned_endpoint_audio_frame, python_audio_frame, PythonAudioFrame}; @@ -37,7 +37,7 @@ pub(crate) struct PythonConnectorInputDescriptor { pub(super) signal_wire_id: String, pub(super) signal: Py, pub(super) media: Py, - pub(super) edge: Py, + pub(super) route_settings: Py, pub(super) configuration: Vec<(String, Py)>, } @@ -63,8 +63,8 @@ impl PythonConnectorInputDescriptor { } #[getter] - fn edge(&self, py: Python<'_>) -> Py { - self.edge.clone_ref(py) + fn route_settings(&self, py: Python<'_>) -> Py { + self.route_settings.clone_ref(py) } } @@ -443,11 +443,11 @@ pub(crate) fn declare_connector( registered: &PythonRegisteredConnector, session: &Session, configuration: &PythonConnectorConfiguration, - edge: &PythonEdgeContract, + route_settings: &PythonRouteSettings, ) -> PyResult { registered .registered - .declare(session, configuration.value.clone(), edge.value) + .declare(session, configuration.value.clone(), route_settings.value) .map(|handle| PythonEndpoint { handle }) .map_err(|error| { PyValueError::new_err(coded_reason( @@ -477,10 +477,10 @@ fn python_input_descriptor( value: input.media(), }, )?; - let edge = Py::new( + let route_settings = Py::new( py, - PythonEdgeContract { - value: input.edge_contract(), + PythonRouteSettings { + value: input.route_settings(), }, )?; Py::new( @@ -493,7 +493,7 @@ fn python_input_descriptor( signal_wire_id: input.signal_spec().wire_id().to_owned(), signal, media, - edge, + route_settings, configuration, }, ) diff --git a/native/src/connector/values.rs b/native/src/connector/values.rs index 961302d..2801961 100644 --- a/native/src/connector/values.rs +++ b/native/src/connector/values.rs @@ -8,7 +8,7 @@ use pocketstation::connector::{ ConnectorRequirement, ConnectorSecret, }; use pocketstation::{ - ExecutionPartition, NodeDescriptor, NodeTypeId, OperatorId, PortDirection, SafetyContract, + ExecutionPartition, ExecutionSafety, NodeDescriptor, NodeTypeId, OperatorId, PortDirection, }; use pyo3::exceptions::PyValueError; use pyo3::prelude::*; @@ -320,7 +320,7 @@ impl PythonConnectorManifest { inputs, Vec::new(), ExecutionPartition::AsyncWorker, - SafetyContract::AllocationAllowed, + ExecutionSafety::AllocationAllowed, true, ) .map_err(|error| invalid_connector(error.to_string()))?; diff --git a/native/src/connector/worker.rs b/native/src/connector/worker.rs index 095a925..a3ff6b0 100644 --- a/native/src/connector/worker.rs +++ b/native/src/connector/worker.rs @@ -24,7 +24,7 @@ use super::values::{ configuration_values, PythonConnectorConfigurationValue, PythonConnectorManifest, }; use crate::errors::coded_reason; -use crate::graph::{PythonEdgeContract, PythonMediaCaps, PythonSignalSpec}; +use crate::graph::{PythonMediaCaps, PythonRouteSettings, PythonSignalSpec}; use crate::signals::{copy_envelope, python_envelope}; use crate::streams::{owned_endpoint_audio_frame_for_route, python_audio_frame, PythonAudioFrame}; @@ -378,10 +378,10 @@ fn python_input_descriptor( value: *input.media(), }, )?; - let edge = Py::new( + let route_settings = Py::new( py, - PythonEdgeContract { - value: *input.edge_contract(), + PythonRouteSettings { + value: *input.route_settings(), }, )?; Py::new( @@ -397,7 +397,7 @@ fn python_input_descriptor( signal_wire_id: input.signal_spec().wire_id().to_owned(), signal, media, - edge, + route_settings, configuration, }, ) diff --git a/native/src/endpoint_authoring.rs b/native/src/endpoint_authoring.rs index 031204c..d823fc6 100644 --- a/native/src/endpoint_authoring.rs +++ b/native/src/endpoint_authoring.rs @@ -6,8 +6,8 @@ use pocketstation::{ ConfigError, EndpointCancellationOutcome, EndpointDriverFactory, EndpointDriverFinalization, EndpointDriverObservations, EndpointFailure, EndpointFailureRetryability, EndpointFailureStage, EndpointInputOrigin, EndpointPortInput, EndpointPreparationGroup, EndpointReceiver, - EndpointShutdownMode, EndpointStartGate, ExecutionPartition, NodeDefinition, NodeDescriptor, - NodeTypeId, OperatorId, PreparedEndpointDriver, RunningEndpointDriver, SafetyContract, Session, + EndpointShutdownMode, EndpointStartGate, ExecutionPartition, ExecutionSafety, NodeDefinition, + NodeDescriptor, NodeTypeId, OperatorId, PreparedEndpointDriver, RunningEndpointDriver, Session, }; use pyo3::exceptions::PyValueError; use pyo3::prelude::*; @@ -15,7 +15,7 @@ use pyo3::types::PyDict; use crate::errors::{coded_reason, session_endpoint_error}; use crate::graph::{ - PythonEdgeContract, PythonEndpoint, PythonMediaCaps, PythonPortSpec, PythonSignalSpec, + PythonEndpoint, PythonMediaCaps, PythonPortSpec, PythonRouteSettings, PythonSignalSpec, }; use crate::signals::{copy_envelope, python_envelope, PythonSignalEnvelope}; use crate::streams::{owned_endpoint_audio_frame_for_route, python_audio_frame, PythonAudioFrame}; @@ -64,7 +64,7 @@ impl PythonEndpointManifest { inputs, Vec::new(), ExecutionPartition::External, - SafetyContract::ExternalService, + ExecutionSafety::ExternalService, true, ) .map_err(|error| invalid_endpoint(error.to_string()))?; @@ -441,7 +441,7 @@ struct PythonEndpointPortInput { port_name: String, signal: Py, media: Py, - edge: Py, + route_settings: Py, context: Py, receiver: Py, } @@ -459,8 +459,8 @@ impl PythonEndpointPortInput { } #[getter] - fn edge(&self, py: Python<'_>) -> Py { - self.edge.clone_ref(py) + fn route_settings(&self, py: Python<'_>) -> Py { + self.route_settings.clone_ref(py) } #[getter] @@ -526,7 +526,7 @@ pub(crate) fn declare_endpoint( session: &Session, registered: &PythonRegisteredEndpoint, configuration: HashMap, - edge: &PythonEdgeContract, + route_settings: &PythonRouteSettings, ) -> PyResult { if registered.session_id != session.id() { return Err(PyValueError::new_err(coded_reason( @@ -545,7 +545,7 @@ pub(crate) fn declare_endpoint( registered.operator_id.clone(), ) .with_configuration(configuration) - .with_input_edge(edge.value), + .with_route_settings(route_settings.value), ) .map(|handle| PythonEndpoint { handle }) .map_err(crate::errors::session_error) @@ -568,10 +568,10 @@ fn python_port_input( value: *input.media(), }, )?; - let edge = Py::new( + let route_settings = Py::new( py, - PythonEdgeContract { - value: *input.edge_contract(), + PythonRouteSettings { + value: *input.route_settings(), }, )?; let prepare = input.context(); @@ -628,7 +628,7 @@ fn python_port_input( port_name, signal, media, - edge, + route_settings, context, receiver, }, diff --git a/native/src/graph.rs b/native/src/graph.rs index 110933e..0739bde 100644 --- a/native/src/graph.rs +++ b/native/src/graph.rs @@ -3,11 +3,11 @@ use std::collections::HashMap; use pocketstation::connector::ConnectorSecret; use pocketstation::{ AudioCaps, BackpressurePolicy, BinaryFormat, ChannelLayout, Codec, CopyPolicy, - DerivedStreamHandle, EdgeContract, EndpointConfiguration, EndpointDescriptor, EndpointHandle, - EventFormat, MediaCaps, Multiplicity, Operator, OperatorConfiguration, OperatorId, - OperatorInputHandle, OperatorInstanceHandle, PortDirection, PortSpec, SampleFormat, - SignalClass, SignalSpec, SourceConfiguration, SourceInstanceHandle, SourceOutputHandle, - SourceTypeId, StemHandle, TextFormat, + DerivedStreamHandle, EndpointConfiguration, EndpointDescriptor, EndpointHandle, EventFormat, + MediaCaps, Multiplicity, Operator, OperatorConfiguration, OperatorId, OperatorInputHandle, + OperatorInstanceHandle, PortDirection, PortSpec, RouteSettings, SampleFormat, SignalClass, + SignalSpec, SourceConfiguration, SourceInstanceHandle, SourceOutputHandle, SourceTypeId, + StemHandle, TextFormat, }; use pocketstation_relay::{RelayPublishReceiptKey, RelayRouteConfiguration}; use pyo3::exceptions::{PyRuntimeError, PyValueError}; @@ -497,25 +497,25 @@ impl PythonPortSpec { } } -#[pyclass(name = "_EdgeContract", frozen)] +#[pyclass(name = "_RouteSettings", frozen)] #[derive(Clone, Copy)] -pub(crate) struct PythonEdgeContract { - pub(crate) value: EdgeContract, +pub(crate) struct PythonRouteSettings { + pub(crate) value: RouteSettings, } #[pymethods] -impl PythonEdgeContract { +impl PythonRouteSettings { #[staticmethod] fn realtime_audio() -> Self { Self { - value: EdgeContract::realtime_audio(), + value: RouteSettings::realtime_audio(), } } #[staticmethod] fn bounded_async() -> Self { Self { - value: EdgeContract::bounded_async(), + value: RouteSettings::bounded_async(), } } @@ -608,9 +608,9 @@ impl PythonEdgeContract { #[getter] fn observability(&self) -> &'static str { match self.value.observability() { - pocketstation::EdgeObservabilityLevel::Off => "off", - pocketstation::EdgeObservabilityLevel::Counters => "counters", - pocketstation::EdgeObservabilityLevel::Full => "full", + pocketstation::RouteObservability::Off => "off", + pocketstation::RouteObservability::Counters => "counters", + pocketstation::RouteObservability::Full => "full", } } @@ -629,12 +629,12 @@ pub(crate) struct PythonEndpointDescriptor { #[pymethods] impl PythonEndpointDescriptor { #[new] - #[pyo3(signature = (node_type_id, operator_id, configuration, input_edge=None))] + #[pyo3(signature = (node_type_id, operator_id, configuration, route_settings=None))] fn new( node_type_id: String, operator_id: String, configuration: HashMap, - input_edge: Option<&PythonEdgeContract>, + route_settings: Option<&PythonRouteSettings>, ) -> PyResult { let configuration = configuration.into_iter().fold( EndpointConfiguration::new(), @@ -645,8 +645,8 @@ impl PythonEndpointDescriptor { OperatorId::new(operator_id), ) .with_configuration(configuration); - if let Some(input_edge) = input_edge { - value = value.with_input_edge(input_edge.value); + if let Some(route_settings) = route_settings { + value = value.with_route_settings(route_settings.value); } Ok(Self { value }) } @@ -784,7 +784,7 @@ fn publish_route( configuration .connector_configuration() .map_err(|error| PyValueError::new_err(error.to_string()))?, - EdgeContract::realtime_audio(), + RouteSettings::realtime_audio(), ) .map_err(|error| PyValueError::new_err(error.to_string()))?; let route_id = send(endpoint)?; @@ -1308,11 +1308,11 @@ fn graph_conformance_manifest( true, ) .map_err(|error| error.to_string())?; - let input_edge = EdgeContract::bounded_async() + let input_route_settings = RouteSettings::bounded_async() .with_media(input_media) .with_backpressure(BackpressurePolicy::DropNewest) .with_copy_policy(CopyPolicy::CopyToBranchPool); - let output_edge = EdgeContract::bounded_async() + let output_route_settings = RouteSettings::bounded_async() .with_media(output_media) .with_copy_policy(CopyPolicy::CopyToBranchPool); pocketstation::AsyncOperatorManifest::new( @@ -1325,12 +1325,12 @@ fn graph_conformance_manifest( vec![input], vec![output], pocketstation::ExecutionPartition::AsyncWorker, - pocketstation::SafetyContract::AllocationAllowed, + pocketstation::ExecutionSafety::AllocationAllowed, false, ) .map_err(|error| error.to_string())?, - input_edge, - output_edge, + input_route_settings, + output_route_settings, 16, pocketstation::OperatorPermissionPolicy { network_allowed: false, @@ -1350,7 +1350,7 @@ pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> { module.add_class::()?; module.add_class::()?; module.add_class::()?; - module.add_class::()?; + module.add_class::()?; module.add_class::()?; module.add_class::()?; module.add_class::()?; diff --git a/native/src/observations.rs b/native/src/observations.rs index 30258b1..e2ac269 100644 --- a/native/src/observations.rs +++ b/native/src/observations.rs @@ -349,7 +349,7 @@ pub(crate) struct PythonOperatorInputMetrics { #[pyo3(get)] port_name: String, #[pyo3(get)] - edge: Py, + delivery: Py, } #[pyclass(name = "_OperatorWorkerMetrics", frozen)] @@ -389,7 +389,7 @@ pub(crate) struct PythonOperatorMetrics { #[pyo3(get)] operator_instance_id: u64, #[pyo3(get)] - input_edge: Py, + input_delivery: Py, #[pyo3(get)] worker: Py, #[pyo3(get)] @@ -407,8 +407,8 @@ impl PythonOperatorMetrics { } } -#[pyclass(name = "_TypedEdgeMetrics", frozen)] -pub(crate) struct PythonTypedEdgeMetrics { +#[pyclass(name = "_SignalQueueMetrics", frozen)] +pub(crate) struct PythonSignalQueueMetrics { #[pyo3(get)] capacity_signals: u64, #[pyo3(get)] @@ -434,7 +434,7 @@ pub(crate) struct PythonDerivedRouteMetrics { #[pyo3(get)] endpoint_id: u64, #[pyo3(get)] - output: Py, + output: Py, #[pyo3(get)] endpoint_observation_stage: String, #[pyo3(get)] @@ -493,8 +493,8 @@ pub(crate) struct PythonAudioReentryMetrics { joined: bool, } -#[pyclass(name = "_EdgeMetrics", frozen)] -pub(crate) struct PythonEdgeMetrics { +#[pyclass(name = "_RouteDeliveryMetrics", frozen)] +pub(crate) struct PythonRouteDeliveryMetrics { #[pyo3(get)] queue_capacity_frames: u64, #[pyo3(get)] @@ -744,7 +744,7 @@ pub(crate) struct PythonRouteMetrics { #[pyo3(get)] drop_rate_pct: f64, #[pyo3(get)] - source_latency_boundary: String, + source_latency_measurement: String, #[pyo3(get)] source_latency_unit: String, } @@ -1671,7 +1671,7 @@ pub(crate) fn python_session_event( }) } -impl From for PythonEdgeMetrics { +impl From for PythonRouteDeliveryMetrics { fn from(edge: pocketstation::EdgeObservations) -> Self { Self { queue_capacity_frames: edge.queue_capacity_frames, @@ -1802,7 +1802,7 @@ fn python_operator_metrics( py, PythonOperatorInputMetrics { port_name: input.port_name.clone(), - edge: Py::new(py, PythonEdgeMetrics::from(input.edge))?, + delivery: Py::new(py, PythonRouteDeliveryMetrics::from(input.edge))?, }, ) }) @@ -1812,7 +1812,7 @@ fn python_operator_metrics( py, PythonOperatorMetrics { operator_instance_id: operator.operator_instance_id.value(), - input_edge: Py::new(py, PythonEdgeMetrics::from(operator.input_edge))?, + input_delivery: Py::new(py, PythonRouteDeliveryMetrics::from(operator.input_delivery))?, worker: Py::new( py, PythonOperatorWorkerMetrics { @@ -1850,7 +1850,7 @@ fn python_derived_route_metrics( endpoint_id: route.endpoint_id.get(), output: Py::new( py, - PythonTypedEdgeMetrics { + PythonSignalQueueMetrics { capacity_signals: route.output.capacity_signals, max_payload_bytes: route.output.max_payload_bytes, maximum_buffered_payload_bytes: route.output.maximum_buffered_payload_bytes, @@ -2132,7 +2132,7 @@ pub(crate) fn python_session_metrics( .endpoint_finalization_failures_total, drop_observation_interval: "route-lifetime-to-snapshot".to_owned(), drop_rate_pct: route.drop_rate_pct, - source_latency_boundary: "source-monotonic-timestamp-to-route-receive" + source_latency_measurement: "source-monotonic-timestamp-to-route-receive" .to_owned(), source_latency_unit: "nanoseconds".to_owned(), }, @@ -2317,11 +2317,11 @@ pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> { module.add_class::()?; module.add_class::()?; module.add_class::()?; - module.add_class::()?; + module.add_class::()?; module.add_class::()?; module.add_class::()?; module.add_class::()?; - module.add_class::()?; + module.add_class::()?; module.add_class::()?; module.add_class::()?; module.add_class::()?; diff --git a/native/src/operator_authoring/driver.rs b/native/src/operator_authoring/driver.rs index 3e73b3c..68412ff 100644 --- a/native/src/operator_authoring/driver.rs +++ b/native/src/operator_authoring/driver.rs @@ -13,7 +13,7 @@ use pyo3::types::PyDict; use super::values::{PythonOperatorEmission, PythonOperatorManifest, PythonOperatorPayload}; use crate::errors::coded_reason; -use crate::graph::{PythonEdgeContract, PythonMediaCaps, PythonSignalSpec}; +use crate::graph::{PythonMediaCaps, PythonRouteSettings, PythonSignalSpec}; use crate::signals::{copy_envelope, python_envelope}; pub(crate) fn register_operator( @@ -48,7 +48,7 @@ pub(crate) struct PythonOperatorPortContext { capacity_signals: usize, signal: Py, media: Py, - edge: Py, + route_settings: Py, } #[pymethods] @@ -64,8 +64,8 @@ impl PythonOperatorPortContext { } #[getter] - fn edge(&self, py: Python<'_>) -> Py { - self.edge.clone_ref(py) + fn route_settings(&self, py: Python<'_>) -> Py { + self.route_settings.clone_ref(py) } } @@ -407,7 +407,7 @@ fn audio_output_spec( )) })?; if manifest - .output_edge() + .output_route_settings() .max_payload_bytes() .is_some_and(|maximum| payload_bytes > maximum) { @@ -475,10 +475,10 @@ fn python_port_context( value: value.media(), }, )?, - edge: Py::new( + route_settings: Py::new( py, - PythonEdgeContract { - value: value.edge_contract(), + PythonRouteSettings { + value: value.route_settings(), }, )?, }, diff --git a/native/src/operator_authoring/values.rs b/native/src/operator_authoring/values.rs index f646001..35f6e82 100644 --- a/native/src/operator_authoring/values.rs +++ b/native/src/operator_authoring/values.rs @@ -1,10 +1,10 @@ use std::sync::Arc; use pocketstation::{ - AsyncOperatorManifest, BackpressurePolicy, CopyPolicy, EdgeContract, ExecutionPartition, + AsyncOperatorManifest, BackpressurePolicy, CopyPolicy, ExecutionPartition, ExecutionSafety, MediaCaps, NodeDescriptor, NodeTypeId, OperatorCancellationPolicy, OperatorDeadlinePolicy, OperatorFailurePolicy, OperatorId, OperatorOutputRolePolicy, OperatorPermissionPolicy, - PortDirection, SafetyContract, SemanticRole, SignalPayload, SignalSpec, + PortDirection, RouteSettings, SemanticRole, SignalPayload, SignalSpec, }; use pyo3::buffer::PyBuffer; use pyo3::exceptions::PyValueError; @@ -64,17 +64,17 @@ impl PythonOperatorManifest { } let input_media = common_media(&inputs, "input")?; let output_media = common_media(&outputs, "output")?; - let input_edge = if matches!(input_media, MediaCaps::Audio(_)) { - EdgeContract::realtime_audio() + let input_route_settings = if matches!(input_media, MediaCaps::Audio(_)) { + RouteSettings::realtime_audio() .with_media(input_media) .with_copy_policy(CopyPolicy::CopyToBranchPool) } else { - EdgeContract::bounded_async() + RouteSettings::bounded_async() .with_media(input_media) .with_backpressure(BackpressurePolicy::DropNewest) .with_copy_policy(CopyPolicy::CopyToBranchPool) }; - let output_edge = EdgeContract::bounded_async() + let output_route_settings = RouteSettings::bounded_async() .with_media(output_media) .with_copy_policy(CopyPolicy::CopyToBranchPool); let roles = outputs @@ -95,7 +95,7 @@ impl PythonOperatorManifest { inputs, outputs, ExecutionPartition::AsyncWorker, - SafetyContract::AllocationAllowed, + ExecutionSafety::AllocationAllowed, true, ) .map_err(|error| invalid_operator(error.to_string()))?; @@ -104,8 +104,8 @@ impl PythonOperatorManifest { revision, implementation_generation, descriptor, - input_edge, - output_edge, + input_route_settings, + output_route_settings, queue_capacity_signals, OperatorPermissionPolicy { network_allowed, diff --git a/native/src/session.rs b/native/src/session.rs index a8014ca..434fddd 100644 --- a/native/src/session.rs +++ b/native/src/session.rs @@ -23,7 +23,7 @@ use crate::errors::{ use crate::extensions::PythonNativeExtensionLibrary; use crate::graph::{ make_operator, make_source_configuration, make_source_type_id, PythonDerivedStream, - PythonEdgeContract, PythonEndpoint, PythonEndpointDescriptor, PythonOperatorInstance, + PythonEndpoint, PythonEndpointDescriptor, PythonOperatorInstance, PythonRouteSettings, PythonSignalSpec, PythonSourceInstance, PythonSourceOutput, PythonStem, }; use crate::observations::{ @@ -333,31 +333,6 @@ impl PythonSession { }) } - #[allow(deprecated)] - fn connector( - &self, - operator_id: String, - configuration: HashMap, - ) -> PyResult { - validate_nonempty("operator ID", &operator_id)?; - if configuration.keys().any(|key| key.trim().is_empty()) { - return Err(PyValueError::new_err(coded_reason( - "graph.invalid_contract", - "endpoint configuration keys cannot be empty", - ))); - } - let configuration = configuration.into_iter().fold( - pocketstation::EndpointConfiguration::new(), - |configuration, (key, value)| configuration.with(key, value), - ); - self.with_session(|session| { - session - .connector(pocketstation::OperatorId::new(operator_id), configuration) - .map(|handle| PythonEndpoint { handle }) - .map_err(session_error) - }) - } - fn browser(&self, receiver_uri: String) -> PyResult { validate_nonempty("receiver URI", &receiver_uri)?; self.with_session(|session| { @@ -424,18 +399,22 @@ impl PythonSession { &self, registered: &PythonRegisteredConnector, configuration: &PythonConnectorConfiguration, - edge: &PythonEdgeContract, + route_settings: &PythonRouteSettings, ) -> PyResult { - self.with_session(|session| declare_connector(registered, session, configuration, edge)) + self.with_session(|session| { + declare_connector(registered, session, configuration, route_settings) + }) } fn declare_registered_endpoint( &self, registered: &PythonRegisteredEndpoint, configuration: HashMap, - edge: &PythonEdgeContract, + route_settings: &PythonRouteSettings, ) -> PyResult { - self.with_session(|session| declare_endpoint(session, registered, configuration, edge)) + self.with_session(|session| { + declare_endpoint(session, registered, configuration, route_settings) + }) } fn register_sidecar(&self, spec: &PythonSidecarProcessSpec) -> PyResult { @@ -471,7 +450,7 @@ impl PythonSession { &self, stream: &PythonDerivedStream, signal: &PythonSignalSpec, - edge: &PythonEdgeContract, + route_settings: &PythonRouteSettings, ) -> PyResult { let subscription_id = self.allocate_signal_subscription_id()?; self.with_session(|session| { @@ -479,7 +458,7 @@ impl PythonSession { session, &stream.handle, signal, - edge, + route_settings, subscription_id, &self.signal_receipts, ) @@ -490,7 +469,7 @@ impl PythonSession { &self, stream: &PythonSourceOutput, signal: &PythonSignalSpec, - edge: &PythonEdgeContract, + route_settings: &PythonRouteSettings, ) -> PyResult { let subscription_id = self.allocate_signal_subscription_id()?; self.with_session(|session| { @@ -498,7 +477,7 @@ impl PythonSession { session, &stream.handle, signal, - edge, + route_settings, subscription_id, &self.signal_receipts, ) diff --git a/native/src/signals.rs b/native/src/signals.rs index cce73c7..9a5e996 100644 --- a/native/src/signals.rs +++ b/native/src/signals.rs @@ -8,16 +8,16 @@ use pocketstation::{ ConfigError, DerivedStreamHandle, EndpointCancellationOutcome, EndpointConfiguration, EndpointDriverFactory, EndpointDriverFinalization, EndpointDriverObservations, EndpointFailure, EndpointFailureStage, EndpointPortInput, EndpointReceiver, EndpointStartGate, - ExecutionPartition, Multiplicity, NodeDefinition, NodeDescriptor, NodeTypeId, OperatorId, - PortDirection, PortSpec, PreparedEndpointDriver, RouteId, RunningEndpointDriver, - SafetyContract, Session, SignalEnvelope, SignalPayload, SignalSpec, SourceOutputHandle, + ExecutionPartition, ExecutionSafety, Multiplicity, NodeDefinition, NodeDescriptor, NodeTypeId, + OperatorId, PortDirection, PortSpec, PreparedEndpointDriver, RouteId, RunningEndpointDriver, + Session, SignalEnvelope, SignalPayload, SignalSpec, SourceOutputHandle, }; use pyo3::exceptions::{PyRuntimeError, PyValueError}; use pyo3::prelude::*; use pyo3::types::{PyBytes, PyMemoryView}; use crate::errors::{coded_reason, session_endpoint_error, session_error}; -use crate::graph::{PythonEdgeContract, PythonSignalSpec}; +use crate::graph::{PythonRouteSettings, PythonSignalSpec}; const SUBSCRIPTION_INPUT_PORT: &str = "signal"; const SUBSCRIPTION_CONFIG_KEY: &str = "subscription_id"; @@ -263,7 +263,7 @@ pub(crate) struct PythonBusSubscription { #[pyo3(get)] pub(crate) route_id: u64, signal: PythonSignalSpec, - edge: PythonEdgeContract, + route_settings: PythonRouteSettings, } #[pymethods] @@ -274,8 +274,8 @@ impl PythonBusSubscription { } #[getter] - fn edge(&self) -> PythonEdgeContract { - self.edge + fn route_settings(&self) -> PythonRouteSettings { + self.route_settings } } @@ -283,7 +283,7 @@ pub(crate) fn subscribe_derived( session: &Session, stream: &DerivedStreamHandle, signal: &PythonSignalSpec, - edge: &PythonEdgeContract, + route_settings: &PythonRouteSettings, subscription_id: u64, receipts: &SignalReceipts, ) -> PyResult { @@ -291,7 +291,7 @@ pub(crate) fn subscribe_derived( session, stream.session_id().get(), signal, - edge, + route_settings, subscription_id, receipts, |endpoint| stream.send(endpoint), @@ -302,7 +302,7 @@ pub(crate) fn subscribe_source_output( session: &Session, stream: &SourceOutputHandle, signal: &PythonSignalSpec, - edge: &PythonEdgeContract, + route_settings: &PythonRouteSettings, subscription_id: u64, receipts: &SignalReceipts, ) -> PyResult { @@ -310,7 +310,7 @@ pub(crate) fn subscribe_source_output( session, stream.session_id().get(), signal, - edge, + route_settings, subscription_id, receipts, |endpoint| stream.send(endpoint), @@ -321,7 +321,7 @@ fn declare_subscription( session: &Session, stream_session_id: u64, signal: &PythonSignalSpec, - edge: &PythonEdgeContract, + route_settings: &PythonRouteSettings, subscription_id: u64, receipts: &SignalReceipts, send: impl FnOnce(pocketstation::EndpointHandle) -> Result, @@ -335,7 +335,7 @@ fn declare_subscription( signal.value.validate().map_err(|error| { PyValueError::new_err(coded_reason("graph.invalid_contract", error.to_string())) })?; - if !edge.value.media().supports_signal(&signal.value) { + if !route_settings.value.media().supports_signal(&signal.value) { return Err(PyValueError::new_err(coded_reason( "graph.invalid_contract", "BusSubscription edge media does not support its SignalSpec", @@ -350,7 +350,7 @@ fn declare_subscription( SUBSCRIPTION_INPUT_PORT, PortDirection::Input, signal.value.clone(), - edge.value.media(), + route_settings.value.media(), Multiplicity::Many, true, ) @@ -363,7 +363,7 @@ fn declare_subscription( vec![input], Vec::new(), ExecutionPartition::External, - SafetyContract::ExternalService, + ExecutionSafety::ExternalService, true, ) .map_err(|error| { @@ -392,7 +392,7 @@ fn declare_subscription( .with_configuration( EndpointConfiguration::new().with(SUBSCRIPTION_CONFIG_KEY, subscription_key), ) - .with_input_edge(edge.value), + .with_route_settings(route_settings.value), ) .map_err(session_error)?; let route_id = send(endpoint).map_err(session_error)?; @@ -405,7 +405,7 @@ fn declare_subscription( session_id: session.id().get(), route_id: route_id.get(), signal: signal.clone(), - edge: *edge, + route_settings: *route_settings, }) } diff --git a/native/src/source_authoring/values.rs b/native/src/source_authoring/values.rs index 3e476e3..3875bf9 100644 --- a/native/src/source_authoring/values.rs +++ b/native/src/source_authoring/values.rs @@ -1,5 +1,5 @@ use pocketstation::{ - ExecutionPartition, PortDirection, SafetyContract, SignalPayload, SignalSpec, SourceManifest, + ExecutionPartition, ExecutionSafety, PortDirection, SignalPayload, SignalSpec, SourceManifest, SourceTypeId, }; use pyo3::exceptions::PyValueError; @@ -48,7 +48,7 @@ impl PythonSourceManifest { implementation_generation, outputs, ExecutionPartition::BlockingWorker, - SafetyContract::AllocationAllowed, + ExecutionSafety::AllocationAllowed, ) .map(|value| Self { value }) .map_err(|error| invalid_source(error.to_string())) diff --git a/python/pocketstation/_api.py b/python/pocketstation/_api.py index 7d11af0..1460a9e 100644 --- a/python/pocketstation/_api.py +++ b/python/pocketstation/_api.py @@ -127,8 +127,6 @@ DeliveryPolicy, DeliverySemantics, DerivedStream, - EdgeContract, - EdgeObservabilityLevel, Endpoint, EndpointConfiguration, EndpointDescriptor, @@ -172,7 +170,6 @@ from .observations import ( AudioReentryMetrics, DerivedRouteMetrics, - EdgeMetrics, EndpointFailureRetryability, EndpointFailureStage, EndpointMetrics, @@ -190,7 +187,6 @@ RecordingState, RelayPublishOutcome, RouteDeliveryMetrics, - RouteLatencyBoundary, RouteLatencyMeasurement, RouteLatencyUnit, RouteObservationInterval, @@ -212,7 +208,6 @@ SignalQueueMetrics, SourceMetrics, TerminationDisposition, - TypedEdgeMetrics, ) from .operator_authoring import ( OperatorConfigValidator, @@ -413,9 +408,6 @@ "DerivedRouteMetrics", "DerivedStream", "DiscoveredSource", - "EdgeContract", - "EdgeMetrics", - "EdgeObservabilityLevel", "EndOfStream", "Endpoint", "EndpointConfiguration", @@ -510,7 +502,6 @@ "RelayTimeoutError", "RouteDeliveryMetrics", "RouteId", - "RouteLatencyBoundary", "RouteLatencyMeasurement", "RouteLatencyUnit", "RouteMetrics", @@ -617,7 +608,6 @@ "SubscriptionState", "TerminationDisposition", "TextFormat", - "TypedEdgeMetrics", "aio", "application_capture_available", "capture", diff --git a/python/pocketstation/_native.pyi b/python/pocketstation/_native.pyi index 44163bb..dd6de57 100644 --- a/python/pocketstation/_native.pyi +++ b/python/pocketstation/_native.pyi @@ -176,11 +176,11 @@ class _PortSpec: multiplicity: str required: bool -class _EdgeContract: +class _RouteSettings: @staticmethod - def realtime_audio() -> _EdgeContract: ... + def realtime_audio() -> _RouteSettings: ... @staticmethod - def bounded_async() -> _EdgeContract: ... + def bounded_async() -> _RouteSettings: ... media: _MediaCaps clock: str latency_budget_ms: int | None @@ -191,18 +191,18 @@ class _EdgeContract: copy_policy: str observability: str max_payload_bytes: int | None - def with_media(self, media: _MediaCaps) -> _EdgeContract: ... - def with_backpressure(self, value: str) -> _EdgeContract: ... - def with_copy_policy(self, value: str) -> _EdgeContract: ... - def with_jitter_budget_ms(self, value: int | None) -> _EdgeContract: ... - def with_max_payload_bytes(self, value: int) -> _EdgeContract: ... + def with_media(self, media: _MediaCaps) -> _RouteSettings: ... + def with_backpressure(self, value: str) -> _RouteSettings: ... + def with_copy_policy(self, value: str) -> _RouteSettings: ... + def with_jitter_budget_ms(self, value: int | None) -> _RouteSettings: ... + def with_max_payload_bytes(self, value: int) -> _RouteSettings: ... class BusSubscription: id: int session_id: RuntimeSessionId route_id: RouteId signal: _SignalSpec - edge: _EdgeContract + route_settings: _RouteSettings class _SignalTiming: source_timestamp_ns: int | None @@ -272,7 +272,7 @@ class _EndpointDescriptor: node_type_id: str, operator_id: str, configuration: dict[str, str], - input_edge: _EdgeContract | None = None, + route_settings: _RouteSettings | None = None, ) -> None: ... class _ConnectorConfigurationValue: @@ -368,7 +368,7 @@ class ConnectorInputDescriptor: signal_wire_id: str signal: _SignalSpec media: _MediaCaps - edge: _EdgeContract + route_settings: _RouteSettings configuration: dict[str, _ConnectorConfigurationValue] class ConnectorItem: @@ -664,7 +664,7 @@ class _SessionFailure: source_platform_status_code: int | None source_backend_class: str | None -class _EdgeMetrics: +class _RouteDeliveryMetrics: queue_capacity_frames: int queue_depth_frames: int queue_peak_frames: int @@ -749,7 +749,7 @@ class RouteMetrics: endpoint_finalization_failures_total: int drop_observation_interval: str drop_rate_pct: float - source_latency_boundary: str + source_latency_measurement: str source_latency_unit: str class _SessionSourceMetrics: @@ -798,7 +798,7 @@ class _ExternalSourceMetrics: class _OperatorInputMetrics: port_name: str - edge: _EdgeMetrics + delivery: _RouteDeliveryMetrics class _OperatorWorkerMetrics: input_attempted_total: int @@ -818,12 +818,12 @@ class _OperatorWorkerMetrics: class _OperatorMetrics: operator_instance_id: int - input_edge: _EdgeMetrics + input_delivery: _RouteDeliveryMetrics worker: _OperatorWorkerMetrics finalization_failures_total: int def input_ports(self) -> list[_OperatorInputMetrics]: ... -class _TypedEdgeMetrics: +class _SignalQueueMetrics: capacity_signals: int max_payload_bytes: int maximum_buffered_payload_bytes: int @@ -836,7 +836,7 @@ class _TypedEdgeMetrics: class _DerivedRouteMetrics: route_id: int endpoint_id: int - output: _TypedEdgeMetrics + output: _SignalQueueMetrics endpoint_observation_stage: str endpoint_frames_received_total: int endpoint_frames_delivered_total: int @@ -1183,7 +1183,7 @@ class EndpointPortInput: port_name: str signal: _SignalSpec media: _MediaCaps - edge: _EdgeContract + route_settings: _RouteSettings context: EndpointPrepareContext receiver: EndpointReceiver @@ -1199,7 +1199,7 @@ class _OperatorPortContext: capacity_signals: int signal: _SignalSpec media: _MediaCaps - edge: _EdgeContract + route_settings: _RouteSettings class _OperatorPrepareContext: execution_partition: str @@ -1287,13 +1287,13 @@ class Session: self, registered: _RegisteredConnector, configuration: _ConnectorConfiguration, - edge: _EdgeContract, + route_settings: _RouteSettings, ) -> Endpoint: ... def declare_registered_endpoint( self, registered: _RegisteredEndpoint, configuration: dict[str, str], - edge: _EdgeContract, + route_settings: _RouteSettings, ) -> Endpoint: ... def load_native_extension_library( self, @@ -1310,13 +1310,13 @@ class Session: self, stream: DerivedStream, signal: _SignalSpec, - edge: _EdgeContract, + route_settings: _RouteSettings, ) -> BusSubscription: ... def subscribe_source_output( self, stream: SourceOutput, signal: _SignalSpec, - edge: _EdgeContract, + route_settings: _RouteSettings, ) -> BusSubscription: ... def start( self, diff --git a/python/pocketstation/aio/connector.py b/python/pocketstation/aio/connector.py index 6a20b9b..5a2c1c8 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, RouteSettings +from ..graph import Endpoint, RouteSettings @dataclass(frozen=True, slots=True) @@ -711,12 +711,10 @@ def declare( configuration: ConnectorConfigurationInput = (), *, route_settings: RouteSettings | None = None, - edge: EdgeContract | None = None, ) -> Endpoint: return self._registered.declare( configuration, route_settings=route_settings, - edge=edge, ) async def observations(self) -> tuple[ConnectorRuntimeObservations, ...]: diff --git a/python/pocketstation/aio/endpoint_authoring.py b/python/pocketstation/aio/endpoint_authoring.py index f797e3f..51c5915 100644 --- a/python/pocketstation/aio/endpoint_authoring.py +++ b/python/pocketstation/aio/endpoint_authoring.py @@ -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, RouteSettings +from ..graph import Endpoint, RouteSettings from ..observations import EndpointFailureStage _Result = TypeVar("_Result") @@ -223,12 +223,10 @@ def declare( configuration: EndpointConfigurationInput = (), *, route_settings: RouteSettings | None = None, - edge: EdgeContract | None = None, ) -> Endpoint: return self._registered.declare( configuration, route_settings=route_settings, - edge=edge, ) diff --git a/python/pocketstation/aio/session.py b/python/pocketstation/aio/session.py index b254d10..dd8f7f4 100644 --- a/python/pocketstation/aio/session.py +++ b/python/pocketstation/aio/session.py @@ -30,7 +30,6 @@ from ..errors import PocketStationError, _native_call, _normalize_native_error from ..extensions import NativeExtensionLibrary from ..graph import ( - EdgeContract, Endpoint, RouteSettings, SignalSpec, @@ -458,13 +457,11 @@ def destination( 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, route_settings=route_settings, - edge=edge, ) def register_endpoint( diff --git a/python/pocketstation/compatibility.py b/python/pocketstation/compatibility.py index d47c8af..6ca4e18 100644 --- a/python/pocketstation/compatibility.py +++ b/python/pocketstation/compatibility.py @@ -19,7 +19,7 @@ class RuntimeCompatibility: RUNTIME_COMPATIBILITY = RuntimeCompatibility( sdk_version="0.1.2", - core_version="1.1.4", + core_version="1.1.5", relay_connector_version="0.1.2", python_requires=">=3.11", python_abi="abi3-py311", diff --git a/python/pocketstation/connector.py b/python/pocketstation/connector.py index 15b0a8f..f82c5f0 100644 --- a/python/pocketstation/connector.py +++ b/python/pocketstation/connector.py @@ -32,7 +32,6 @@ from ._native import _RegisteredConnector as _NativeRegisteredConnector from .errors import PocketStationError, _native_call from .graph import ( - EdgeContract, Endpoint, MediaCaps, Multiplicity, @@ -558,11 +557,7 @@ def media(self) -> MediaCaps: @property def route_settings(self) -> RouteSettings: - return RouteSettings(self._native.edge) - - @property - def edge(self) -> EdgeContract: - return self.route_settings + return RouteSettings(self._native.route_settings) @property def configuration(self) -> Mapping[str, ConnectorConfigurationValue]: @@ -1237,7 +1232,6 @@ def declare( 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( @@ -1245,7 +1239,6 @@ def declare( ) selected_settings = _select_route_settings( route_settings, - edge, _default_route_settings(self._connector.manifest), ) native = _native_call( diff --git a/python/pocketstation/endpoint_authoring.py b/python/pocketstation/endpoint_authoring.py index 681614c..3398d54 100644 --- a/python/pocketstation/endpoint_authoring.py +++ b/python/pocketstation/endpoint_authoring.py @@ -36,7 +36,6 @@ ) from .errors import PocketStationError, _native_call from .graph import ( - EdgeContract, Endpoint, MediaCaps, PortSpec, @@ -266,11 +265,7 @@ def media(self) -> MediaCaps: @property def route_settings(self) -> RouteSettings: - return RouteSettings(self._native.edge) - - @property - def edge(self) -> EdgeContract: - return self.route_settings + return RouteSettings(self._native.route_settings) @property def context(self) -> EndpointPrepareContext: @@ -434,12 +429,10 @@ def declare( configuration: EndpointConfigurationInput = (), *, route_settings: RouteSettings | None = None, - edge: EdgeContract | None = None, ) -> Endpoint: values = _configuration(configuration) selected_settings = _select_route_settings( route_settings, - edge, _default_route_settings(self._provider.manifest), ) native = _native_call( diff --git a/python/pocketstation/graph.py b/python/pocketstation/graph.py index ef948b0..836bc0b 100644 --- a/python/pocketstation/graph.py +++ b/python/pocketstation/graph.py @@ -15,10 +15,10 @@ from ._native import SourceInstance as _NativeSourceInstance from ._native import SourceOutput as _NativeSourceOutput from ._native import Stem as _NativeStem -from ._native import _EdgeContract as _NativeEdgeContract from ._native import _EndpointDescriptor as _NativeEndpointDescriptor from ._native import _MediaCaps as _NativeMediaCaps from ._native import _PortSpec as _NativePortSpec +from ._native import _RouteSettings as _NativeRouteSettings from ._native import _SignalSpec as _NativeSignalSpec from .errors import _native_call from .identity import ( @@ -509,22 +509,19 @@ def rank(self) -> int: }[self] -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) + _native: _NativeRouteSettings = field(repr=False, compare=False) @classmethod def realtime_audio(cls) -> DeliveryPolicy: - return cls(_NativeEdgeContract.realtime_audio()) + return cls(_NativeRouteSettings.realtime_audio()) @classmethod def bounded_async(cls) -> DeliveryPolicy: - return cls(_NativeEdgeContract.bounded_async()) + return cls(_NativeRouteSettings.bounded_async()) @property def clock(self) -> ClockDomain: @@ -604,15 +601,15 @@ def __hash__(self) -> int: class RouteSettings: """Choose the media accepted by a route and how that route delivers it.""" - _native: _NativeEdgeContract = field(repr=False, compare=False) + _native: _NativeRouteSettings = field(repr=False, compare=False) @classmethod def realtime_audio(cls) -> RouteSettings: - return cls(_NativeEdgeContract.realtime_audio()) + return cls(_NativeRouteSettings.realtime_audio()) @classmethod def bounded_async(cls) -> RouteSettings: - return cls(_NativeEdgeContract.bounded_async()) + return cls(_NativeRouteSettings.bounded_async()) @property def media(self) -> MediaCaps: @@ -692,17 +689,11 @@ 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 + return route_settings or default ConfigurationInput: TypeAlias = Mapping[str, str] | Iterable[tuple[str, str]] @@ -775,20 +766,16 @@ class EndpointDescriptor: node_type_id: str operator_id: str configuration: EndpointConfiguration = field(default_factory=EndpointConfiguration) - input_edge: EdgeContract | None = None + route_settings: RouteSettings | 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( self.node_type_id, self.operator_id, self.configuration._as_dict(), - None if self.input_edge is None else self.input_edge._native, + None if self.route_settings is None else self.route_settings._native, ) ) object.__setattr__(self, "_native", native) @@ -1131,17 +1118,6 @@ def endpoint(self, descriptor: EndpointDescriptor) -> Endpoint: """Declare one open Endpoint descriptor on the Session draft.""" return _native_call(lambda: Endpoint(self._native.endpoint(descriptor._native))) - def connector( - self, - operator_id: str, - configuration: EndpointConfiguration | None = None, - ) -> Endpoint: - """Declare an external connector endpoint without provider taxonomy.""" - values = EndpointConfiguration() if configuration is None else configuration - return _native_call( - lambda: Endpoint(self._native.connector(operator_id, values._as_dict())) - ) - def browser(self, receiver_uri: str) -> Endpoint: """Declare the frozen browser or remote-receiver Endpoint.""" return _native_call(lambda: Endpoint(self._native.browser(receiver_uri))) @@ -1152,7 +1128,6 @@ def subscribe( *, signal: SignalSpec[_PayloadT], route_settings: RouteSettings | None = None, - edge: EdgeContract | None = None, ) -> BusSubscription[_PayloadT]: """Declare one bounded, exclusive typed-signal subscription. @@ -1163,7 +1138,6 @@ def subscribe( settings = _select_route_settings( route_settings, - edge, RouteSettings.bounded_async().with_media(_media_for_signal(signal)), ) if isinstance(stream, DerivedStream): @@ -1245,8 +1219,6 @@ def _media_for_signal(signal: SignalSpec[object]) -> MediaCaps: "DeliveryPolicy", "DeliverySemantics", "DerivedStream", - "EdgeContract", - "EdgeObservabilityLevel", "Endpoint", "EndpointConfiguration", "EndpointDescriptor", diff --git a/python/pocketstation/observations.py b/python/pocketstation/observations.py index 7c614c2..ccc7405 100644 --- a/python/pocketstation/observations.py +++ b/python/pocketstation/observations.py @@ -21,15 +21,15 @@ from ._native import StopResult as _NativeStopResult from ._native import _AudioReentryMetrics as _NativeAudioReentryMetrics from ._native import _DerivedRouteMetrics as _NativeDerivedRouteMetrics -from ._native import _EdgeMetrics as _NativeEdgeMetrics from ._native import _ExternalSourceMetrics as _NativeExternalSourceMetrics from ._native import _OperatorInputMetrics as _NativeOperatorInputMetrics from ._native import _OperatorMetrics as _NativeOperatorMetrics from ._native import _OperatorWorkerMetrics as _NativeOperatorWorkerMetrics +from ._native import _RouteDeliveryMetrics as _NativeRouteDeliveryMetrics from ._native import _SessionFailure as _NativeSessionFailure from ._native import _SessionSourceMetrics as _NativeSessionSourceMetrics from ._native import _SessionTraceValidation as _NativeTraceValidation -from ._native import _TypedEdgeMetrics as _NativeTypedEdgeMetrics +from ._native import _SignalQueueMetrics as _NativeSignalQueueMetrics from .errors import PocketStationError, _native_call from .identity import ( EndpointId, @@ -153,9 +153,6 @@ class RouteLatencyMeasurement(StrEnum): SOURCE_TIMESTAMP_TO_ROUTE_RECEIVE = "source-monotonic-timestamp-to-route-receive" -RouteLatencyBoundary = RouteLatencyMeasurement - - class RouteLatencyUnit(StrEnum): NANOSECONDS = "nanoseconds" @@ -394,7 +391,7 @@ class RouteDeliveryMetrics: discarded_output_frames_total: int | None @classmethod - def _from_native(cls, value: _NativeEdgeMetrics) -> RouteDeliveryMetrics: + def _from_native(cls, value: _NativeRouteDeliveryMetrics) -> RouteDeliveryMetrics: return cls( queue_capacity_frames=value.queue_capacity_frames, queue_depth_frames=value.queue_depth_frames, @@ -441,9 +438,6 @@ def _from_native(cls, value: _NativeEdgeMetrics) -> RouteDeliveryMetrics: ) -EdgeMetrics = RouteDeliveryMetrics - - @dataclass(frozen=True, slots=True) class EndpointMetrics: observation_stage: EndpointObservationStage @@ -459,41 +453,35 @@ class EndpointMetrics: class RouteMetrics: route_id: int endpoint_id: int - edge: RouteDeliveryMetrics + delivery: RouteDeliveryMetrics endpoint: EndpointMetrics frames_attempted_total: int observation_interval: RouteObservationInterval drop_rate_pct: float - source_latency_boundary: RouteLatencyMeasurement + source_latency_measurement: 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 + return self.delivery.queue_capacity_frames @property def frames_delivered_total(self) -> int: - return self.edge.frames_delivered_total + return self.delivery.frames_delivered_total @property def frames_dropped_total(self) -> int: - return self.edge.frames_dropped_total - - @property - def source_latency_measurement(self) -> RouteLatencyMeasurement: - return self.source_latency_boundary + return self.delivery.frames_dropped_total @classmethod def _from_native(cls, value: _NativeRouteMetrics) -> RouteMetrics: - edge = RouteDeliveryMetrics._from_native(cast(_NativeEdgeMetrics, value)) + delivery = RouteDeliveryMetrics._from_native( + cast(_NativeRouteDeliveryMetrics, value) + ) return cls( route_id=value.route_id, endpoint_id=value.endpoint_id, - edge=edge, + delivery=delivery, endpoint=EndpointMetrics( observation_stage=EndpointObservationStage( value.endpoint_observation_stage @@ -510,8 +498,8 @@ def _from_native(cls, value: _NativeRouteMetrics) -> RouteMetrics: value.drop_observation_interval ), drop_rate_pct=value.drop_rate_pct, - source_latency_boundary=RouteLatencyMeasurement( - value.source_latency_boundary + source_latency_measurement=RouteLatencyMeasurement( + value.source_latency_measurement ), source_latency_unit=RouteLatencyUnit(value.source_latency_unit), ) @@ -611,13 +599,10 @@ class SignalQueueMetrics: dropped_total: int @classmethod - def _from_native(cls, value: _NativeTypedEdgeMetrics) -> SignalQueueMetrics: + def _from_native(cls, value: _NativeSignalQueueMetrics) -> 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 @@ -643,37 +628,29 @@ def _from_native(cls, value: _NativeOperatorWorkerMetrics) -> OperatorWorkerMetr @dataclass(frozen=True, slots=True) class OperatorInputMetrics: port_name: str - edge: RouteDeliveryMetrics - - @property - def delivery(self) -> RouteDeliveryMetrics: - return self.edge + delivery: RouteDeliveryMetrics @classmethod def _from_native(cls, value: _NativeOperatorInputMetrics) -> OperatorInputMetrics: return cls( port_name=value.port_name, - edge=RouteDeliveryMetrics._from_native(value.edge), + delivery=RouteDeliveryMetrics._from_native(value.delivery), ) @dataclass(frozen=True, slots=True) class OperatorMetrics: operator_instance_id: int - input_edge: RouteDeliveryMetrics + input_delivery: 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=RouteDeliveryMetrics._from_native(value.input_edge), + input_delivery=RouteDeliveryMetrics._from_native(value.input_delivery), worker=OperatorWorkerMetrics._from_native(value.worker), finalization_failures_total=value.finalization_failures_total, input_ports=tuple( @@ -1259,7 +1236,6 @@ def iterate() -> Iterator[SessionEvent]: __all__ = [ "AudioReentryMetrics", "DerivedRouteMetrics", - "EdgeMetrics", "EndpointFailureRetryability", "EndpointFailureStage", "EndpointMetrics", @@ -1279,7 +1255,6 @@ def iterate() -> Iterator[SessionEvent]: "RecordingStemOutcome", "RelayPublishOutcome", "RouteDeliveryMetrics", - "RouteLatencyBoundary", "RouteLatencyMeasurement", "RouteLatencyUnit", "RouteMetrics", @@ -1305,5 +1280,4 @@ def iterate() -> Iterator[SessionEvent]: "SourceMetrics", "StopResult", "TerminationDisposition", - "TypedEdgeMetrics", ] diff --git a/python/pocketstation/operator_authoring.py b/python/pocketstation/operator_authoring.py index 2cf8a53..b0d47e0 100644 --- a/python/pocketstation/operator_authoring.py +++ b/python/pocketstation/operator_authoring.py @@ -13,7 +13,6 @@ from ._native import _SignalEnvelope as _NativeSignalEnvelope from .errors import _native_call from .graph import ( - EdgeContract, MediaCaps, Operator, OperatorConfiguration, @@ -76,11 +75,7 @@ class OperatorPortContext: capacity_signals: int signal: SignalSpec[object] media: MediaCaps - edge: EdgeContract - - @property - def route_settings(self) -> RouteSettings: - return self.edge + route_settings: RouteSettings @classmethod def _from_native(cls, value: _NativeOperatorPortContext) -> OperatorPortContext: @@ -91,7 +86,7 @@ def _from_native(cls, value: _NativeOperatorPortContext) -> OperatorPortContext: capacity_signals=value.capacity_signals, signal=SignalSpec._from_native(value.signal), media=MediaCaps._from_native(value.media), - edge=EdgeContract(value.edge), + route_settings=RouteSettings(value.route_settings), ) diff --git a/python/pocketstation/session.py b/python/pocketstation/session.py index dd2b909..07a5b07 100644 --- a/python/pocketstation/session.py +++ b/python/pocketstation/session.py @@ -29,7 +29,6 @@ from .errors import PocketStationError, _native_call from .extensions import NativeExtensionLibrary from .graph import ( - EdgeContract, Endpoint, RouteSettings, Stem, @@ -388,7 +387,6 @@ def destination( configuration: ConnectorConfigurationInput = (), *, route_settings: RouteSettings | None = None, - edge: EdgeContract | None = None, ) -> Endpoint: """Declare one Connector destination using an idempotent registration. @@ -399,7 +397,6 @@ def destination( return self.register_connector(connector).declare( configuration, route_settings=route_settings, - edge=edge, ) def register_endpoint(self, endpoint: EndpointProvider) -> RegisteredEndpoint: diff --git a/python/pocketstation/signal.py b/python/pocketstation/signal.py index 46f09dc..40ad020 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, RouteSettings, SignalSpec +from .graph import RouteSettings, SignalSpec from .identity import ( ClockDomainId, ConnectorId, @@ -245,11 +245,7 @@ def signal(self) -> SignalSpec[_PayloadT_co]: @property def route_settings(self) -> RouteSettings: - return RouteSettings(self._native.edge) - - @property - def edge(self) -> EdgeContract: - return self.route_settings + return RouteSettings(self._native.route_settings) class EndOfStream: diff --git a/python/pocketstation/voice/conversation.py b/python/pocketstation/voice/conversation.py index eb6f510..784e626 100644 --- a/python/pocketstation/voice/conversation.py +++ b/python/pocketstation/voice/conversation.py @@ -930,9 +930,9 @@ async def _wait_output_drained(self, running: object) -> None: "generated audio delivery failed before output drained" ) routes_drained = bool(routes) and all( - route.edge.queue_depth_frames == 0 - and route.edge.frames_delivered_total - + (route.edge.discarded_output_frames_total or 0) + route.delivery.queue_depth_frames == 0 + and route.delivery.frames_delivered_total + + (route.delivery.discarded_output_frames_total or 0) >= self._output_frames_written for route in routes ) diff --git a/tests/qualification/runtime_resources.py b/tests/qualification/runtime_resources.py index c029552..f5dfae2 100644 --- a/tests/qualification/runtime_resources.py +++ b/tests/qualification/runtime_resources.py @@ -297,7 +297,7 @@ def qualify_slow_consumer( if len(metrics.routes) != 1: raise RuntimeError("slow-consumer qualification expected one route") route = metrics.routes[0] - if route.edge.queue_peak_frames > route.queue_capacity_frames: + if route.delivery.queue_peak_frames > route.queue_capacity_frames: raise RuntimeError("slow-consumer queue exceeded its declared capacity") if route.frames_dropped_total == 0: raise RuntimeError("slow-consumer saturation was not observed") @@ -309,7 +309,7 @@ def qualify_slow_consumer( frames_accepted_total=observations.accepted_total, input_full_retries_total=retries_total, route_capacity_frames=route.queue_capacity_frames, - route_peak_frames=route.edge.queue_peak_frames, + route_peak_frames=route.delivery.queue_peak_frames, route_drops_total=route.frames_dropped_total, stop_success=stop.success, ) diff --git a/tests/test_connector.py b/tests/test_connector.py index 5cd10b4..579f0df 100644 --- a/tests/test_connector.py +++ b/tests/test_connector.py @@ -94,7 +94,7 @@ def prepare( assert descriptor.signal_wire_id == "pks.signal.pcm-audio.v1" assert descriptor.signal.is_audio assert descriptor.media.kind.value == "audio-pcm" - assert descriptor.edge.media.kind.value == "audio-pcm" + assert descriptor.route_settings.media.kind.value == "audio-pcm" assert len(driver.items) == 1 item = driver.items[0] assert item.kind == "audio" @@ -244,12 +244,6 @@ def test_session_destination_accepts_route_settings_and_rejects_duplicate_names( 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() -> ( diff --git a/tests/test_conversation_interruptions.py b/tests/test_conversation_interruptions.py index 2d07a45..72ad5c3 100644 --- a/tests/test_conversation_interruptions.py +++ b/tests/test_conversation_interruptions.py @@ -209,7 +209,10 @@ async def synthesize( assert await running.audio.read(timeout_s=0.01) is None discarded_output_frames_total = ( metrics.polled_audio.discarded_output_frames_total - + sum(route.edge.discarded_output_frames_total or 0 for route in metrics.routes) + + sum( + route.delivery.discarded_output_frames_total or 0 + for route in metrics.routes + ) ) assert discarded_output_frames_total >= 1 assert stopped.success diff --git a/tests/test_graph.py b/tests/test_graph.py index 881094c..53338f1 100644 --- a/tests/test_graph.py +++ b/tests/test_graph.py @@ -13,8 +13,6 @@ CopyPolicy, DeliveryPolicy, DeliverySemantics, - EdgeContract, - EdgeObservabilityLevel, EndpointConfiguration, EndpointDescriptor, EventFormat, @@ -164,8 +162,8 @@ def test_port_helpers_infer_media_without_hiding_explicit_contracts() -> None: assert output_port.multiplicity is Multiplicity.MANY -def test_edge_presets_and_modifiers_preserve_bounded_contracts() -> None: - realtime = EdgeContract.realtime_audio() +def test_route_settings_presets_and_modifiers_remain_bounded() -> None: + realtime = RouteSettings.realtime_audio() assert realtime.clock is ClockDomain.CAPTURE assert realtime.backpressure is BackpressurePolicy.DROP_NEWEST assert realtime.delivery is DeliverySemantics.ORDERED @@ -175,10 +173,9 @@ def test_edge_presets_and_modifiers_preserve_bounded_contracts() -> None: assert realtime.max_payload_bytes is None assert realtime.clock.is_realtime assert not ClockDomain.INHERITED.is_realtime - assert EdgeObservabilityLevel is RouteObservability assert RouteObservability.FULL.rank > realtime.observability.rank - bounded = EdgeContract.bounded_async() + bounded = RouteSettings.bounded_async() assert bounded.clock is ClockDomain.INHERITED assert bounded.backpressure is BackpressurePolicy.BOUNDED_QUEUE assert bounded.delivery is DeliverySemantics.ORDERED @@ -218,7 +215,6 @@ def test_route_settings_apply_delivery_policy_without_changing_media() -> None: 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 @@ -261,9 +257,12 @@ def test_graph_declarations_lower_immediately_to_one_rust_session(tmp_path) -> N OperatorConfiguration({"language": "en"}), ) ) - connector = session.connector( - "org.example.connector.v1", - EndpointConfiguration({"region": "local"}), + connector = session.endpoint( + EndpointDescriptor( + "org.example.connector-node.v1", + "org.example.connector.v1", + EndpointConfiguration({"region": "local"}), + ) ) browser = session.browser("https://receiver.example.test") endpoint = session.endpoint( @@ -271,7 +270,7 @@ def test_graph_declarations_lower_immediately_to_one_rust_session(tmp_path) -> N "org.example.endpoint-node.v1", "org.example.endpoint.v1", EndpointConfiguration({"mode": "events"}), - EdgeContract.bounded_async(), + RouteSettings.bounded_async(), ) ) @@ -345,4 +344,10 @@ def test_sync_and_async_sessions_share_the_same_graph_declaration_surface() -> N assert session.id > 0 declared = session.operator(Operator("org.example.operator.v1")) assert declared.session_id == session.id - assert session.connector("org.example.connector.v1").session_id == session.id + endpoint = session.endpoint( + EndpointDescriptor( + "org.example.endpoint-node.v1", + "org.example.endpoint.v1", + ) + ) + assert endpoint.session_id == session.id diff --git a/tests/test_metrics.py b/tests/test_metrics.py index 69544ee..364f43e 100644 --- a/tests/test_metrics.py +++ b/tests/test_metrics.py @@ -7,7 +7,6 @@ import pocketstation._native as _native import pytest from pocketstation._api import ( - EdgeMetrics, EndpointObservationStage, PocketStationError, RouteDeliveryMetrics, @@ -43,9 +42,10 @@ def test_metrics_preserve_bounded_source_route_and_polled_audio_truth(tmp_path) assert metrics.polled_audio.registered_endpoints == 2 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.delivery.queue_capacity_frames > 0 for route in metrics.routes) + assert all( + isinstance(route.delivery, RouteDeliveryMetrics) for route in metrics.routes + ) assert all( route.endpoint.observation_stage is EndpointObservationStage.LIVE for route in metrics.routes diff --git a/tests/test_signal_streams.py b/tests/test_signal_streams.py index 5cdc895..eb4ff99 100644 --- a/tests/test_signal_streams.py +++ b/tests/test_signal_streams.py @@ -131,8 +131,10 @@ def test_real_session_delivers_audio_text_and_bytes_with_complete_provenance( "bytes": bytes_stream, } for name, subscription in subscriptions.items(): - assert subscription.edge.backpressure is BackpressurePolicy.BOUNDED_QUEUE - assert subscription.edge.max_payload_bytes == 1_048_576 + assert ( + subscription.route_settings.backpressure is BackpressurePolicy.BOUNDED_QUEUE + ) + assert subscription.route_settings.max_payload_bytes == 1_048_576 metrics = streams[name].metrics() assert metrics.capacity_signals > 0 assert metrics.max_payload_bytes == 1_048_576 @@ -173,7 +175,7 @@ def test_external_source_outputs_have_the_same_subscription_declaration() -> Non assert subscription.session_id == session.id assert subscription.signal == SignalSpec.text(TextFormat.JSON) - assert subscription.edge.media.supports_signal(subscription.signal) + assert subscription.route_settings.media.supports_signal(subscription.signal) assert subscription.route_id > 0 From d4e8b757bc49f7d7c81437339102be40ac835ac1 Mon Sep 17 00:00:00 2001 From: Raphael Avocegamou Date: Thu, 3 Sep 2026 08:28:57 +0200 Subject: [PATCH 2/9] release: prepare Python 0.1.3 --- README.md | 4 ++-- RELEASE_NOTES.md | 25 +++++++++++++++++++------ native/Cargo.lock | 10 +++------- native/Cargo.toml | 8 ++++---- pyproject.toml | 2 +- python/pocketstation/__init__.py | 2 +- python/pocketstation/_api.py | 2 +- python/pocketstation/compatibility.py | 6 +++--- uv.lock | 2 +- 9 files changed, 35 insertions(+), 26 deletions(-) diff --git a/README.md b/README.md index ba0b9db..89b2930 100644 --- a/README.md +++ b/README.md @@ -243,8 +243,8 @@ does not have the same execution cost as Rust. | 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 | -The native binding uses PocketStation Core `1.1.5` and the shared Relay -Connector `0.1.2`. +The native binding uses PocketStation Core `1.1.6` and the shared Relay +Connector `0.1.3`. 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; diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 98ba073..dbb7bae 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -2,6 +2,11 @@ ## Unreleased +## 0.1.3 — 2026-09-03 + +Configure route delivery with one consistent set of names across Python and +the native runtime. + ### Added Advanced integrations can now configure `RouteSettings` as accepted media plus @@ -10,9 +15,17 @@ 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. +Session operator metrics expose aggregate input delivery through +`input_delivery`, with per-port detail in `input_ports`. + +### Changed + +Connector, Endpoint, Operator, subscription, and observation APIs now use the +same route-settings vocabulary. + +```console +python -m pip install --upgrade pocketstation==0.1.3 +``` ## 0.1.2 — 2026-09-01 @@ -35,7 +48,7 @@ application.send_to(destination) Subclass `pocketstation.Connector` for a synchronous destination or `pocketstation.aio.Connector` for an asynchronous provider. The class owns its -provider connection. PocketStation owns the bounded routes, source and stem +provider connection. PocketStation owns route delivery, source and stem identity, delivery observations, drain, abort, and joined shutdown. ```python @@ -98,7 +111,7 @@ The first release includes: - synchronous and asyncio Session APIs; - exact application selection and default-microphone capture; - 10 ms and 20 ms audio profiles; -- bounded audio and typed-signal streams; +- audio and typed-signal streams with explicit queue capacities; - Python-authored Sources, Operators, Connectors, and Endpoints; - application-owned PCM input and generated-audio output cancellation; - provider-neutral voice composition with revisable transcripts; @@ -106,7 +119,7 @@ The first release includes: - source-aware multistem recording with structured outcomes. The package uses PocketStation Core for capture, routing, timing, recording, -and lifecycle. Python provider work runs on bounded off-realtime workers and +and lifecycle. Python provider work runs on off-realtime workers and does not execute on native capture callbacks. ### Voice interruption example diff --git a/native/Cargo.lock b/native/Cargo.lock index 6cac869..e43ca9d 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -1413,9 +1413,7 @@ checksum = "f6b464fbc74e149a392436b17d523f769e057cb6877f6a5c4618bc6f11800548" [[package]] name = "pocketstation" -version = "1.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c320a8850b385affb4b6f985ea1b7db91a24692a48448f2b9ea8cb7e12245eef" +version = "1.1.6" dependencies = [ "alsa", "cc", @@ -1437,7 +1435,7 @@ dependencies = [ [[package]] name = "pocketstation-python" -version = "0.1.2" +version = "0.1.3" dependencies = [ "pocketstation", "pocketstation-relay", @@ -1447,9 +1445,7 @@ dependencies = [ [[package]] name = "pocketstation-relay" -version = "0.1.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5389b4483333c9dd7b68e25dc3d2b5790fa66f75864bb02267d9fed449c64b30" +version = "0.1.3" dependencies = [ "base64", "pocketstation", diff --git a/native/Cargo.toml b/native/Cargo.toml index 931ba9a..f6816f1 100644 --- a/native/Cargo.toml +++ b/native/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "pocketstation-python" -version = "0.1.2" +version = "0.1.3" edition = "2021" publish = false description = "Native PocketStation runtime bindings for the Python SDK" @@ -16,10 +16,10 @@ default = [] conformance-fixtures = ["pocketstation/conformance-fixtures"] [dependencies] -pocketstation = "=1.1.4" -pocketstation-relay = "=0.1.2" +pocketstation = "=1.1.6" +pocketstation-relay = "=0.1.3" pyo3 = { version = "0.27", features = ["abi3-py311"] } [dev-dependencies] -pocketstation = { version = "=1.1.4", features = ["conformance-fixtures"] } +pocketstation = { version = "=1.1.6", features = ["conformance-fixtures"] } tempfile = "3" diff --git a/pyproject.toml b/pyproject.toml index 9cf234e..bc2324a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "maturin" [project] name = "pocketstation" -version = "0.1.2" +version = "0.1.3" description = "Source-aware live audio capture, processing, and routing for Python" readme = "README.md" requires-python = ">=3.11" diff --git a/python/pocketstation/__init__.py b/python/pocketstation/__init__.py index f5d2339..21564d9 100644 --- a/python/pocketstation/__init__.py +++ b/python/pocketstation/__init__.py @@ -16,7 +16,7 @@ from .session import RecordingOutcome, RunningSession, Session, StopResult from .sources import Source, discover_sources -__version__ = "0.1.2" +__version__ = "0.1.3" __all__ = [ "RUNTIME_COMPATIBILITY", diff --git a/python/pocketstation/_api.py b/python/pocketstation/_api.py index 1460a9e..fcd2d0f 100644 --- a/python/pocketstation/_api.py +++ b/python/pocketstation/_api.py @@ -322,7 +322,7 @@ SignalStream, ) -__version__ = "0.1.2" +__version__ = "0.1.3" __all__ = [ "RUNTIME_COMPATIBILITY", "STREAM_EOF", diff --git a/python/pocketstation/compatibility.py b/python/pocketstation/compatibility.py index 6ca4e18..f3ae550 100644 --- a/python/pocketstation/compatibility.py +++ b/python/pocketstation/compatibility.py @@ -18,9 +18,9 @@ class RuntimeCompatibility: RUNTIME_COMPATIBILITY = RuntimeCompatibility( - sdk_version="0.1.2", - core_version="1.1.5", - relay_connector_version="0.1.2", + sdk_version="0.1.3", + core_version="1.1.6", + relay_connector_version="0.1.3", python_requires=">=3.11", python_abi="abi3-py311", free_threaded_cpython=False, diff --git a/uv.lock b/uv.lock index f183881..efc7167 100644 --- a/uv.lock +++ b/uv.lock @@ -746,7 +746,7 @@ wheels = [ [[package]] name = "pocketstation" -version = "0.1.2" +version = "0.1.3" source = { editable = "." } dependencies = [ { name = "httpx" }, From 7e93023d86d74e3c157ee11f7bc6268070fcd98c Mon Sep 17 00:00:00 2001 From: Raphael Avocegamou Date: Thu, 3 Sep 2026 08:31:51 +0200 Subject: [PATCH 3/9] docs: explain Python integrations directly --- README.md | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index 89b2930..36b6e63 100644 --- a/README.md +++ b/README.md @@ -154,14 +154,9 @@ cancelled, and invalid-buffer outcomes explicitly. ## Create an integration -PocketStation provides four integration APIs: - -| 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 direct control of an outbound worker. | +Create a `Connector` when Session audio needs to reach an API, socket, file, or +provider. Most Python integrations need only a send function or a small class; +PocketStation supplies the worker, queue, delivery observations, and shutdown. Pass one function when the destination is already open: @@ -208,10 +203,15 @@ PocketStation calls `start()` once, interleaves both source-aware stems through destination. See [Create an integration](docs/guides/integrations.md) for deadlines, failures, and the advanced SPI. -Python provider callbacks execute on off-realtime workers. They cannot -be used as native capture callbacks. Use compiled native extensions for native -provider code, or a process sidecar when crash -isolation is required. +Python provider callbacks execute on off-realtime workers. They cannot be used +as native capture callbacks. Use a compiled native extension for native +provider code, or a managed process when crash isolation is required. + +Use a `Source` when media enters the Session, an `Operator` when work transforms +media or emits typed signals, and an `Endpoint` when an integration needs direct +control of outbound delivery. The [integration guide](docs/guides/integrations.md) +starts with the normal Connector API and introduces those advanced APIs only +when the task requires them. ## Use Relay from Python From 3eb982c6969784c5d89ce37d381414ba6befc097 Mon Sep 17 00:00:00 2001 From: Raphael Avocegamou Date: Thu, 3 Sep 2026 10:05:58 +0200 Subject: [PATCH 4/9] ci: require concrete queue limits in docs --- .github/workflows/ci.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c90c323..b993442 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -78,10 +78,10 @@ jobs: working-directory: sdk-python shell: bash run: | - if rg -n -i '\b(boundary|path|surface|authority|projection|lowering|flow|layer|contracts?)\b' \ + if rg -n -i '\b(boundary|path|surface|authority|projection|lowering|flow|layer|bounded|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." + echo "Public documentation must name the API, queue, limit, process, service, or request directly." exit 1 fi From b4d2961a03325c288a8917ec3538631282dd8ce5 Mon Sep 17 00:00:00 2001 From: Raphael Avocegamou Date: Thu, 3 Sep 2026 10:09:12 +0200 Subject: [PATCH 5/9] release: bind published native dependencies --- native/Cargo.lock | 20 ++++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/native/Cargo.lock b/native/Cargo.lock index e43ca9d..d569b9b 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -134,9 +134,9 @@ checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" [[package]] name = "aws-lc-rs" -version = "1.18.0" +version = "1.18.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ce2b2dcc879c3bae0d371e77c99f2238400ef24ec001394befa67b6e543add9e" +checksum = "b281d307588d634de920874890732659e2e7672f72b5e10e81badc1a8a83621e" dependencies = [ "aws-lc-sys", "untrusted", @@ -145,9 +145,9 @@ dependencies = [ [[package]] name = "aws-lc-sys" -version = "0.44.0" +version = "0.45.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f09fae7be8bb3174e05c6afdb34199e6dc0c7c04ba9fa237b1967adfbde27483" +checksum = "9bff6c3b54fad79a2e60b8102caf565819711497c1f5f092f49508e2f5c31b27" dependencies = [ "cc", "cmake", @@ -1414,6 +1414,8 @@ checksum = "f6b464fbc74e149a392436b17d523f769e057cb6877f6a5c4618bc6f11800548" [[package]] name = "pocketstation" version = "1.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c9f775adeeb6fedadb2d684771472ff6b1d40f39074669e6841386998d010e7f" dependencies = [ "alsa", "cc", @@ -1446,6 +1448,8 @@ dependencies = [ [[package]] name = "pocketstation-relay" version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "566a0e6fcd4fc3dc33d09795d8939f2fcb9e1da83951caedb048e2218561095c" dependencies = [ "base64", "pocketstation", @@ -1895,9 +1899,9 @@ checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" [[package]] name = "smallvec" -version = "1.15.2" +version = "1.16.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" +checksum = "b9be42f50aa861c555654aa3a37f52f4b1074bacf4e48fe0ef7fa584e80f1f0f" [[package]] name = "spki" @@ -2136,9 +2140,9 @@ dependencies = [ [[package]] name = "toml" -version = "1.1.4+spec-1.1.0" +version = "1.1.5+spec-1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3aace63f4bbcdfc2c965b059de67119c89c4017a70d633be6c104910f67056f5" +checksum = "12c0ba9680044b4ce98d391a62094047eada0d64860b80166c39f4a6b5640785" dependencies = [ "indexmap", "serde_core", From 5db2f0ff49d5328721f3775c5556077e7dbb90dc Mon Sep 17 00:00:00 2001 From: Raphael Avocegamou Date: Thu, 3 Sep 2026 10:18:17 +0200 Subject: [PATCH 6/9] style: format native observations --- native/src/observations.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/native/src/observations.rs b/native/src/observations.rs index e2ac269..2477e2e 100644 --- a/native/src/observations.rs +++ b/native/src/observations.rs @@ -1812,7 +1812,10 @@ fn python_operator_metrics( py, PythonOperatorMetrics { operator_instance_id: operator.operator_instance_id.value(), - input_delivery: Py::new(py, PythonRouteDeliveryMetrics::from(operator.input_delivery))?, + input_delivery: Py::new( + py, + PythonRouteDeliveryMetrics::from(operator.input_delivery), + )?, worker: Py::new( py, PythonOperatorWorkerMetrics { From 003f408106eefd619d4482f862d6a67e4d9cd065 Mon Sep 17 00:00:00 2001 From: Raphael Avocegamou Date: Thu, 3 Sep 2026 10:19:34 +0200 Subject: [PATCH 7/9] docs: explain transcription and audio queues directly --- README.md | 19 +++++++++---------- docs/getting-started/capture.md | 5 +++-- examples/README.md | 9 +++++---- 3 files changed, 17 insertions(+), 16 deletions(-) diff --git a/README.md b/README.md index 36b6e63..7d58bb9 100644 --- a/README.md +++ b/README.md @@ -75,22 +75,21 @@ python -m pip install 'pocketstation[transcription]' python examples/transcribe_voice_app.py ``` -The program asks which desktop voice application to inspect. It declares one -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 +The program asks which desktop voice application to inspect. It sends the +application and microphone through one faster-whisper model, then labels each +transcript with the source that produced it. It does not start Relay or write a recording. The Session preserves each source while the transcriber processes both: ```text -voice application ──┐ - ├─ PocketStation Session ─ faster-whisper ─ source-aware transcripts -physical microphone ┘ +voice application ── faster-whisper ── transcript labeled "application" +physical microphone ─ faster-whisper ─ transcript labeled "microphone" ``` The complete composition is visible in [`examples/transcribe_voice_app.py`](examples/transcribe_voice_app.py). The -example adapter imports `faster_whisper.WhisperModel` when the Operator starts; +example adapter imports `faster_whisper.WhisperModel` when transcription starts; the provider is not part of the `pocketstation` namespace. This example does not debug turn handling, interruption, agent latency, or @@ -131,9 +130,9 @@ with pocketstation.capture( print(frame.source_id, frame.stem_id) ``` -The iterator reads frames from a native Endpoint. If Python stops reading, the -Endpoint reports its queue depth, dropped frames, and discontinuities instead -of allowing memory use to grow without a limit. +The iterator receives audio through a native queue that holds 32 frames by +default. If Python stops reading and the queue fills, PocketStation drops new +frames and reports the queue depth, dropped-frame count, and discontinuity. ## Send application-owned audio into a Session diff --git a/docs/getting-started/capture.md b/docs/getting-started/capture.md index 26ecb57..1b3f12c 100644 --- a/docs/getting-started/capture.md +++ b/docs/getting-started/capture.md @@ -36,8 +36,9 @@ discover it again. For a saved selection, use in [Persist a source at its supported scope](../operations/platform-support.md#persist-a-source-at-its-supported-scope). The context manager starts one native Session and joins it when the block -exits. The iterator reads a finite native Endpoint; it does not create an -unbounded Python audio queue. +exits. The iterator receives audio through a native queue that holds 32 frames +by default. If Python stops reading and the queue fills, PocketStation drops +new frames and reports the loss. ## Add a microphone or recording diff --git a/examples/README.md b/examples/README.md index 77c4cfa..093d722 100644 --- a/examples/README.md +++ b/examples/README.md @@ -71,8 +71,9 @@ strict admission limits. Set `POCKETSTATION_CONTROL_URL` and ## Transcribe both sides of a voice application Use this example to inspect what a desktop voice application produced and what -the person said into the microphone. PocketStation keeps both sources separate -while one faster-whisper Operator transcribes them. +the person said into the microphone. One faster-whisper model transcribes both +sources, and every transcript identifies whether it came from the application +or the microphone. ```bash python examples/transcribe_voice_app.py @@ -87,8 +88,8 @@ Install the optional model dependency before the first run: python -m pip install 'pocketstation[transcription]' ``` -The first run may download the configured faster-whisper model. Model work runs -on an off-realtime Operator worker, not on a capture callback. +The first run may download the configured faster-whisper model. PocketStation +runs transcription on a dedicated worker, never on the capture callback. ## Stream application audio to a browser From 17b011338a322469db9ed27b33efa1d850e1a003 Mon Sep 17 00:00:00 2001 From: Raphael Avocegamou Date: Thu, 3 Sep 2026 11:13:23 +0200 Subject: [PATCH 8/9] release: bind Core 1.1.7 and Relay 0.1.5 --- README.md | 4 ++-- RELEASE_NOTES.md | 3 +++ native/Cargo.lock | 8 ++++---- native/Cargo.toml | 6 +++--- python/pocketstation/compatibility.py | 4 ++-- 5 files changed, 14 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index 7d58bb9..f373432 100644 --- a/README.md +++ b/README.md @@ -242,8 +242,8 @@ does not have the same execution cost as Rust. | 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 | -The native binding uses PocketStation Core `1.1.6` and the shared Relay -Connector `0.1.3`. +The native binding uses PocketStation Core `1.1.7` and the shared Relay +Connector `0.1.5`. 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; diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index dbb7bae..8314137 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -18,6 +18,9 @@ accept the clearer `route_settings=` keyword. Runtime metrics expose Session operator metrics expose aggregate input delivery through `input_delivery`, with per-port detail in `input_ports`. +This release uses PocketStation Core 1.1.7, which keeps concurrent queue-depth +observations within the configured route capacity, and Relay Connector 0.1.5. + ### Changed Connector, Endpoint, Operator, subscription, and observation APIs now use the diff --git a/native/Cargo.lock b/native/Cargo.lock index d569b9b..d875cb1 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -1413,9 +1413,9 @@ checksum = "f6b464fbc74e149a392436b17d523f769e057cb6877f6a5c4618bc6f11800548" [[package]] name = "pocketstation" -version = "1.1.6" +version = "1.1.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c9f775adeeb6fedadb2d684771472ff6b1d40f39074669e6841386998d010e7f" +checksum = "78d452027085b8556c77af89673e62815e81babe0b1f930bd8d26158427c5784" dependencies = [ "alsa", "cc", @@ -1447,9 +1447,9 @@ dependencies = [ [[package]] name = "pocketstation-relay" -version = "0.1.3" +version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "566a0e6fcd4fc3dc33d09795d8939f2fcb9e1da83951caedb048e2218561095c" +checksum = "3f2ae2918f830246e7475e0d7539c144052042fcd1af657e96dbfd4e323ddb38" dependencies = [ "base64", "pocketstation", diff --git a/native/Cargo.toml b/native/Cargo.toml index f6816f1..3026496 100644 --- a/native/Cargo.toml +++ b/native/Cargo.toml @@ -16,10 +16,10 @@ default = [] conformance-fixtures = ["pocketstation/conformance-fixtures"] [dependencies] -pocketstation = "=1.1.6" -pocketstation-relay = "=0.1.3" +pocketstation = "=1.1.7" +pocketstation-relay = "=0.1.5" pyo3 = { version = "0.27", features = ["abi3-py311"] } [dev-dependencies] -pocketstation = { version = "=1.1.6", features = ["conformance-fixtures"] } +pocketstation = { version = "=1.1.7", features = ["conformance-fixtures"] } tempfile = "3" diff --git a/python/pocketstation/compatibility.py b/python/pocketstation/compatibility.py index f3ae550..8cac81c 100644 --- a/python/pocketstation/compatibility.py +++ b/python/pocketstation/compatibility.py @@ -19,8 +19,8 @@ class RuntimeCompatibility: RUNTIME_COMPATIBILITY = RuntimeCompatibility( sdk_version="0.1.3", - core_version="1.1.6", - relay_connector_version="0.1.3", + core_version="1.1.7", + relay_connector_version="0.1.5", python_requires=">=3.11", python_abi="abi3-py311", free_threaded_cpython=False, From fd7a41bb5694d34c931c612084f5c7af50c07dd8 Mon Sep 17 00:00:00 2001 From: Raphael Avocegamou Date: Thu, 3 Sep 2026 11:15:07 +0200 Subject: [PATCH 9/9] docs: format public Python examples --- README.md | 2 ++ RELEASE_NOTES.md | 1 + docs/guides/integrations.md | 3 +++ docs/guides/process-audio-and-signals.md | 1 + 4 files changed, 7 insertions(+) diff --git a/README.md b/README.md index f373432..d0f1e94 100644 --- a/README.md +++ b/README.md @@ -163,9 +163,11 @@ Pass one function when the destination is already open: import pocketstation as pks import pocketstation.aio as pks_aio + async def send_audio(frame: pks.AudioFrame) -> None: await socket.send(frame.samples) + destination = pks_aio.Connector(send=send_audio) application.send_to(destination) ``` diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 8314137..1c0bb66 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -45,6 +45,7 @@ Use `Connector(send=...)` when the destination is already open: async def send_audio(frame): await socket.send(frame.samples) + destination = pocketstation.aio.Connector(send=send_audio) application.send_to(destination) ``` diff --git a/docs/guides/integrations.md b/docs/guides/integrations.md index 73ed980..23faaa4 100644 --- a/docs/guides/integrations.md +++ b/docs/guides/integrations.md @@ -54,9 +54,11 @@ Use one function when the provider connection is already open: import pocketstation as pks import pocketstation.aio as pks_aio + async def send_audio(frame: pks.AudioFrame) -> None: await socket.send(frame.samples) + destination = pks_aio.Connector(send=send_audio) application.send_to(destination) ``` @@ -77,6 +79,7 @@ Use a class when the integration is reused or owns provider state: import pocketstation as pks import pocketstation.aio as pks_aio + class WebSocketConnector(pks_aio.Connector): def __init__(self, url: str, token: str) -> None: self.url = url diff --git a/docs/guides/process-audio-and-signals.md b/docs/guides/process-audio-and-signals.md index b42edf4..a6f44bc 100644 --- a/docs/guides/process-audio-and-signals.md +++ b/docs/guides/process-audio-and-signals.md @@ -27,6 +27,7 @@ manifest = OperatorManifest( outputs=(PortSpec.output("output", result),), ) + @operator(manifest) async def uppercase(input_port, envelope): assert input_port == "input"