diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 8e817f7..2df1249 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -6,6 +6,16 @@ on: description: Existing version tag (vX.Y.Z) committed through a reviewed PR type: string required: true + release_mode: + description: Use the ordinary main path or reviewed-epic release admission + type: choice + options: [main, reviewed-epic] + default: main + required: true + reviewed_pr: + description: Positive canonical PR number whose merge commit the tag names (reviewed-epic mode only) + type: number + required: false push: tags: ['v*'] permissions: {} @@ -18,11 +28,15 @@ jobs: outputs: tag: ${{ steps.tag.outputs.tag }} version: ${{ steps.tag.outputs.version }} + release_mode: ${{ steps.tag.outputs.release_mode }} + reviewed_pr: ${{ steps.tag.outputs.reviewed_pr }} steps: - id: tag shell: bash env: TAG: ${{ github.event_name == 'push' && github.ref_name || inputs.tag }} + RELEASE_MODE: ${{ github.event_name == 'push' && 'main' || inputs.release_mode }} + REVIEWED_PR: ${{ inputs.reviewed_pr }} run: | set -euo pipefail [[ "$TAG" =~ ^v[0-9]+\.[0-9]+\.[0-9]+([+-][0-9A-Za-z.-]+)?$ ]] || { @@ -31,8 +45,30 @@ jobs: } echo "tag=$TAG" >> "$GITHUB_OUTPUT" echo "version=${TAG#v}" >> "$GITHUB_OUTPUT" + [[ "$RELEASE_MODE" == main || "$RELEASE_MODE" == reviewed-epic ]] || { + echo "release_mode must be main or reviewed-epic" >&2 + exit 2 + } + if [[ "$RELEASE_MODE" == reviewed-epic ]]; then + [[ "$REVIEWED_PR" =~ ^[1-9][0-9]*$ ]] || { + echo "reviewed_pr must be a positive integer in reviewed-epic mode" >&2 + exit 2 + } + else + [[ -z "$REVIEWED_PR" || "$REVIEWED_PR" == 0 ]] || { + echo "reviewed_pr is only valid in reviewed-epic mode" >&2 + exit 2 + } + REVIEWED_PR="" + fi + { + echo "release_mode=$RELEASE_MODE" + # The reusable workflow input is numeric; zero is the ordinary-path sentinel. + echo "reviewed_pr=${REVIEWED_PR:-0}" + } >> "$GITHUB_OUTPUT" source-packages: needs: resolve + if: needs.resolve.outputs.release_mode == 'main' runs-on: ubuntu-latest permissions: contents: read @@ -59,12 +95,26 @@ jobs: if-no-files-found: error module-release: needs: [resolve, source-packages] + if: needs.resolve.outputs.release_mode == 'main' permissions: contents: write - uses: tinyhumansai/.github/.github/workflows/module-release.yml@4adc1080fb495098e7af1aade82ed7e6e3c69bb7 + uses: tinyhumansai/.github/.github/workflows/module-release.yml@1e28317c435a9c714002199e65d18a1067fa5777 with: tag: ${{ needs.resolve.outputs.tag }} version: ${{ needs.resolve.outputs.version }} module-crate: tinydocs-module extra-packaged-files: '["docs/specs/tinybus-module.md"]' source-artifact: source-packages + module-reviewed-epic-release: + needs: resolve + if: needs.resolve.outputs.release_mode == 'reviewed-epic' + permissions: + contents: write + pull-requests: read + checks: read + actions: read + uses: tinyhumansai/.github/.github/workflows/module-reviewed-epic-release.yml@1e28317c435a9c714002199e65d18a1067fa5777 + with: + tag: ${{ needs.resolve.outputs.tag }} + version: ${{ needs.resolve.outputs.version }} + reviewed-pr: ${{ fromJSON(needs.resolve.outputs.reviewed_pr || '0') }} diff --git a/Cargo.lock b/Cargo.lock index da16e0b..c73a9b5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -49,15 +49,6 @@ version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" -[[package]] -name = "arbitrary" -version = "1.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c3d036a3c4ab069c7b410a2ce876bd74808d2d0888a82667669f8e783a898bf1" -dependencies = [ - "derive_arbitrary", -] - [[package]] name = "arrayvec" version = "0.7.8" @@ -419,17 +410,6 @@ version = "0.5.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" -[[package]] -name = "derive_arbitrary" -version = "1.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e567bd82dcff979e4b03460c307b3cdc9e96fde3d73bed1496d2bc75d9dd62a" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.119", -] - [[package]] name = "digest" version = "0.10.7" @@ -890,6 +870,95 @@ dependencies = [ "typenum", ] +[[package]] +name = "icu_casemap" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2edfd120d8445de853cf53a1abec916f638f933a70a6f4bf8f7e2d7b7f884e9" +dependencies = [ + "icu_casemap_data", + "icu_collections", + "icu_locale_core", + "icu_properties", + "icu_provider", + "potential_utf", + "writeable", + "zerovec", +] + +[[package]] +name = "icu_casemap_data" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "059fcda0fc8d33effc6633a8cf0d8a3c17caf7d7c91918fba9918744f501e4e8" + +[[package]] +name = "icu_collections" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fa68d21081c4a05d5a901a1c62add574c77048b6a1c67be3b50ce0b60d4ca513" +dependencies = [ + "displaydoc", + "potential_utf", + "serde", + "utf8_iter", + "yoke", + "zerofrom", + "zerovec", +] + +[[package]] +name = "icu_locale_core" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d56e28588da92eee5c3201a6eff33fabdd49b62269c8938d4ff050ce4d900deb" +dependencies = [ + "displaydoc", + "litemap", + "serde", + "tinystr", + "writeable", + "zerovec", +] + +[[package]] +name = "icu_properties" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e7ca276ad3145661a65914e6daf131ca5120cd3dcee8f8f3214b8875184a148" +dependencies = [ + "displaydoc", + "icu_collections", + "icu_locale_core", + "icu_properties_data", + "icu_provider", + "zerotrie", + "zerovec", +] + +[[package]] +name = "icu_properties_data" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e590f038c1464a96894fd6d10127e90a8be4509f56ff7ecef851b15cee0b7caa" + +[[package]] +name = "icu_provider" +version = "2.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d27bbb9d3abbefac45d55f647c9de1d44aafcd1186eb91879afef17c396c3e73" +dependencies = [ + "displaydoc", + "icu_locale_core", + "serde", + "stable_deref_trait", + "writeable", + "yoke", + "zerofrom", + "zerotrie", + "zerovec", +] + [[package]] name = "image" version = "0.25.10" @@ -1002,6 +1071,12 @@ version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" +[[package]] +name = "litemap" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae" + [[package]] name = "litrs" version = "1.0.0" @@ -1327,6 +1402,17 @@ version = "0.14.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "78451badbdaebaf17f053fd9152b3ffb33b516104eacb45e7864aaa9c712f306" +[[package]] +name = "potential_utf" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d83eb9bc6d8e5cf568e7a1101d60ee05e81ed50ea106026f3d18deeb046d7661" +dependencies = [ + "serde_core", + "writeable", + "zerovec", +] + [[package]] name = "powerfmt" version = "0.2.0" @@ -1436,6 +1522,12 @@ version = "1.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a611d15b50743feb4c76b7d03edcb0e64f399c26961e4efe6975bc398be6aa3d" +[[package]] +name = "rawzip" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac9999039f74cd73c03158cddfe4362991703c74efe5ce5c8073026f80a5e48c" + [[package]] name = "read-fonts" version = "0.35.0" @@ -1842,7 +1934,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.3", "once_cell", "rustix", "windows-sys 0.61.2", @@ -1914,11 +2006,13 @@ dependencies = [ [[package]] name = "tinybus" -version = "0.1.5" +version = "0.3.2" dependencies = [ "async-trait", "flate2", + "icu_casemap", "libc", + "rawzip", "serde", "serde_json", "sha2 0.10.9", @@ -1929,13 +2023,13 @@ dependencies = [ "tokio", "toml", "tracing", + "unicode-normalization", "ureq", - "zip 2.4.2", ] [[package]] name = "tinybus-macros" -version = "0.1.5" +version = "0.3.2" dependencies = [ "proc-macro2", "quote", @@ -1944,7 +2038,7 @@ dependencies = [ [[package]] name = "tinybus-module" -version = "0.1.5" +version = "0.3.2" dependencies = [ "async-trait", "serde", @@ -1956,7 +2050,7 @@ dependencies = [ [[package]] name = "tinydocs" -version = "0.2.4" +version = "0.2.5" dependencies = [ "calamine", "docx-rs", @@ -1972,16 +2066,30 @@ dependencies = [ [[package]] name = "tinydocs-bus" -version = "0.2.4" +version = "0.2.5" dependencies = [ "serde", "serde_json", "thiserror", ] +[[package]] +name = "tinydocs-client" +version = "0.2.5" +dependencies = [ + "async-trait", + "base64 0.22.1", + "serde_json", + "sha2 0.10.9", + "tinybus", + "tinydocs-bus", + "tokio", + "tracing", +] + [[package]] name = "tinydocs-module" -version = "0.2.4" +version = "0.2.5" dependencies = [ "base64 0.23.1", "getrandom 0.4.3", @@ -1997,6 +2105,17 @@ dependencies = [ "zip 8.6.0", ] +[[package]] +name = "tinystr" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1e27c91459209c2986af3dcf603a5a74a4368754ce37414f59acc971167f643" +dependencies = [ + "displaydoc", + "serde_core", + "zerovec", +] + [[package]] name = "tinyvec" version = "1.13.3" @@ -2190,6 +2309,12 @@ version = "0.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b8c0a043c9540bae7c578c88f91dda8bd82e59ae27c21baca69c8b191aaf5a6e" +[[package]] +name = "utf8_iter" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" + [[package]] name = "uuid" version = "1.26.1" @@ -2441,6 +2566,12 @@ dependencies = [ "read-fonts 0.39.2", ] +[[package]] +name = "writeable" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc" + [[package]] name = "xattr" version = "1.6.1" @@ -2536,6 +2667,41 @@ version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" +[[package]] +name = "zerotrie" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ea269c3bd32f0a32c321907a2ae912ba6f4649bb0fc764a15627e99a7095a3f" +dependencies = [ + "displaydoc", + "yoke", + "zerofrom", + "zerovec", +] + +[[package]] +name = "zerovec" +version = "0.11.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bb0464e17806c1d976d5cba29399c7f08e516e279e2ba493f63123b5fca67dd8" +dependencies = [ + "serde", + "yoke", + "zerofrom", + "zerovec-derive", +] + +[[package]] +name = "zerovec-derive" +version = "0.11.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34df6fc39dbd26ddc9c10e6a2984476e13acce22e64e4487636ef494369225da" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.6", +] + [[package]] name = "zip" version = "0.6.6" @@ -2556,23 +2722,6 @@ dependencies = [ "zstd", ] -[[package]] -name = "zip" -version = "2.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fabe6324e908f85a1c52063ce7aa26b68dcb7eb6dbc83a2d148403c9bc3eba50" -dependencies = [ - "arbitrary", - "crc32fast", - "crossbeam-utils", - "displaydoc", - "flate2", - "indexmap", - "memchr", - "thiserror", - "zopfli", -] - [[package]] name = "zip" version = "8.6.0" diff --git a/Cargo.toml b/Cargo.toml index 9846f17..f86d02d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -22,19 +22,19 @@ exclude = [ ] [workspace] -members = ["crates/tinydocs-bus", "crates/tinydocs-module"] -default-members = [".", "crates/tinydocs-bus", "crates/tinydocs-module"] +members = ["crates/tinydocs-bus", "crates/tinydocs-client", "crates/tinydocs-module"] +default-members = [".", "crates/tinydocs-bus", "crates/tinydocs-client", "crates/tinydocs-module"] exclude = ["vendor/tinybus"] resolver = "3" [workspace.package] # Releases update this sole workspace version before packaging every crate. -version = "0.2.4" +version = "0.2.5" [dependencies] # The dependency-light TinyBus contract owns payload types and errors. Re-export # it so library and bus callers use the exact same values. -tinydocs-bus = { version = "0.2.4", path = "crates/tinydocs-bus" } +tinydocs-bus = { version = "0.2.5", path = "crates/tinydocs-bus" } # OOXML `.docx` synthesis. Optional: exclusive to the `docx` feature so a host # that only needs extraction does not pull the writer stack. docx-rs = { version = "0.4.20", optional = true } diff --git a/crates/tinydocs-client/Cargo.toml b/crates/tinydocs-client/Cargo.toml new file mode 100644 index 0000000..6c9739c --- /dev/null +++ b/crates/tinydocs-client/Cargo.toml @@ -0,0 +1,32 @@ +[package] +name = "tinydocs-client" +version.workspace = true +edition = "2024" +rust-version = "1.88" +license = "GPL-3.0-only" +description = "Typed TinyBus client for the TinyDocs module." +repository = "https://github.com/tinyhumansai/tinydocs" +documentation = "https://docs.rs/tinydocs-client" +keywords = ["tinybus", "document", "agent"] +categories = ["api-bindings"] + +[dependencies] +# Module loading and typed failure reporting stay host-owned; this is the +# published host-facing SDK without its native loader. +tinybus = { version = "0.3.2", default-features = false } +# Shared wire names and payload DTOs. +tinydocs-bus = { version = "0.2.5", path = "../tinydocs-bus" } +# The held-output protocol uses base64 chunks and verifies their SHA-256 digest. +base64 = "0.22" +sha2 = "0.10" +serde_json = "1" +# Cancellation-safe output release schedules cleanup on the caller's runtime. +tokio = { version = "1", features = ["rt", "time"] } +tracing = "0.1" + +[dev-dependencies] +async-trait = "0.1" +tokio = { version = "1", features = ["macros", "rt", "sync", "time", "test-util"] } + +[lints] +workspace = true diff --git a/crates/tinydocs-client/README.md b/crates/tinydocs-client/README.md new file mode 100644 index 0000000..efc173f --- /dev/null +++ b/crates/tinydocs-client/README.md @@ -0,0 +1,30 @@ +# TinyDocs client + +`tinydocs-client` is the host-side typed client for the `tinydocs` TinyBus +module. It translates contract methods and held outputs into typed Rust calls. +It does not load modules, decide readiness, configure reporting, or choose +deadlines; those choices belong to the embedding host. + +Each operation that produces a held output binds an ordinary +`PinnedModuleClient` for that operation. Its producer call, optional streamed +upload, output reads, releases, and cancellation cleanup stay with the same +serving peer even if the public module alias changes during the operation. +Binding does not claim native artifact proof. Stateless `ExtractDocument` and +`InspectImage` calls continue to resolve through the host's `ModuleClient` +without retaining a peer beyond the call. The host continues to own module +loading and transport lifecycle. + +`DocumentCallError` keeps unavailable, invalid-input, and execution failures +distinct. Its `Debug` and `Display` forms expose only the category. A caller +that needs TinyDocs' input detail can opt into `raw_message()` and apply its own +redaction policy before surfacing that untrusted text. + +Variable-size results use the contract's held-output protocol. The client +reads bounded base64 chunks, checks the declared length and SHA-256 digest, and +releases each output after success or failure. On cancellation, it makes a +best-effort release call on the caller's Tokio runtime. If the runtime is +already gone or that call cannot be delivered, the client cannot guarantee +that the module has physically reclaimed the output; module-side expiry remains +the fallback. The contract and behavior are exercised by +`tests/recording_broker.rs`, which serves a local recording interface over +TinyBus's in-memory broker and never contacts a live module. diff --git a/crates/tinydocs-client/src/client.rs b/crates/tinydocs-client/src/client.rs new file mode 100644 index 0000000..2468f06 --- /dev/null +++ b/crates/tinydocs-client/src/client.rs @@ -0,0 +1,406 @@ +//! Typed calls and held-output transfer for the `TinyDocs` module. + +use std::fmt::{Debug, Formatter}; + +use base64::Engine as _; +use base64::engine::general_purpose::STANDARD as BASE64; +use sha2::{Digest, Sha256}; +use tinybus::module::{ + ModuleCallError, ModuleCallFailure, ModuleClient, PinnedModuleClient, PinnedModuleObjectClient, +}; +use tinydocs_bus::names::methods; +use tinydocs_bus::spec::presentation::MAX_IMAGE_BYTES; +use tinydocs_bus::{ + DocumentFormat, DocumentSpec, ExtractDocumentSpec, ExtractedDocument, ImageFacts, OutputRef, + RenderPdfSpec, RenderedPdf, WirePresentationSpec, +}; + +use crate::DocumentCallError; + +const MODULE_ID: &str = "tinydocs"; +const READ_CHUNK_BYTES: u64 = 1024 * 1024; +const MAX_DOCUMENT_BYTES: u64 = 64 * 1024 * 1024; +const INVALID_INPUT_WIRE_NAME: &str = "ai.tinyhumans.tinydocs.Error.InvalidInput"; + +/// Typed client for the `TinyDocs` module. +/// +/// The embedding host supplies its process-owned [`ModuleClient`]. Each +/// operation that returns held outputs binds the serving peer for that +/// operation, keeping its producer, output reads, releases, and cleanup on the +/// same peer. Stateless one-call operations continue through `ModuleClient`. +#[derive(Clone)] +pub struct TinyDocsClient { + module: ModuleClient, +} + +impl Debug for TinyDocsClient { + fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("TinyDocsClient") + .finish_non_exhaustive() + } +} + +impl TinyDocsClient { + /// Create a `TinyDocs` client over the host's module client. + #[must_use] + pub fn new(module: ModuleClient) -> Self { + Self { module } + } + + /// Generate and collect a `.docx` document. + /// + /// # Errors + /// + /// Returns [`DocumentCallError`] when `TinyDocs` is unavailable, rejects the + /// specification, or fails to generate or transfer the output. + pub async fn generate_docx(&self, spec: &DocumentSpec) -> Result, DocumentCallError> { + let module = self.bind().await?; + let output = Self::output_client(&module)?; + let handle: OutputRef = module + .call_detailed(methods::GENERATE_DOCX, (spec,)) + .await + .map_err(classify_failure)?; + collect(&output, handle).await + } + + /// Generate and collect a `.pptx` document. + /// + /// `images` contains slide-image bytes concatenated in slide order. Each + /// slide's byte length in `deck` defines its range in that buffer. An empty + /// image buffer uses the text-only call without opening a stream. + /// + /// # Errors + /// + /// Returns [`DocumentCallError`] when `TinyDocs` is unavailable, rejects the + /// deck, or fails to generate or transfer the output. + pub async fn generate_pptx( + &self, + deck: &WirePresentationSpec, + images: &[u8], + ) -> Result, DocumentCallError> { + let module = self.bind().await?; + let output = Self::output_client(&module)?; + let handle: OutputRef = if images.is_empty() { + module + .call_detailed( + methods::GENERATE_PPTX, + (deck, Option::::None), + ) + .await + .map_err(classify_failure)? + } else { + module + .call_with_stream_detailed( + methods::GENERATE_PPTX, + |stream| serde_json::json!([deck, stream]), + images, + ) + .await + .map_err(classify_failure)? + }; + collect(&output, handle).await + } + + /// Extract the text layer from a streamed PDF. + /// + /// Scanned documents with no text layer produce an empty string. + /// + /// # Errors + /// + /// Returns [`DocumentCallError`] when extraction, output transfer, or UTF-8 + /// decoding fails. + pub async fn extract_text(&self, document: &[u8]) -> Result { + let module = self.bind().await?; + let output = Self::output_client(&module)?; + let handle: OutputRef = module + .call_with_stream_detailed( + methods::EXTRACT_TEXT, + |stream| serde_json::json!([stream]), + document, + ) + .await + .map_err(classify_failure)?; + let bytes = collect(&output, handle).await?; + String::from_utf8(bytes).map_err(|_| { + DocumentCallError::Failed("extracted text was not valid UTF-8".to_string()) + }) + } + + /// Extract bounded text and provenance from a streamed document. + /// + /// # Errors + /// + /// Returns [`DocumentCallError`] when extraction or the typed reply fails. + pub async fn extract_document( + &self, + document: &[u8], + spec: &ExtractDocumentSpec, + ) -> Result { + self.module + .call_with_stream_detailed( + MODULE_ID, + methods::EXTRACT_DOCUMENT, + |stream| serde_json::json!([spec, stream]), + document, + ) + .await + .map_err(classify_failure) + } + + /// Convert a streamed PDF or Office document into normalized Markdown. + /// + /// # Errors + /// + /// Returns [`DocumentCallError`] when conversion, output transfer, or UTF-8 + /// decoding fails. + pub async fn convert_markdown( + &self, + format: DocumentFormat, + document: &[u8], + ) -> Result { + let module = self.bind().await?; + let output = Self::output_client(&module)?; + let handle: OutputRef = module + .call_with_stream_detailed( + methods::CONVERT_MARKDOWN, + |stream| serde_json::json!([format, stream]), + document, + ) + .await + .map_err(classify_failure)?; + let bytes = collect(&output, handle).await?; + String::from_utf8(bytes).map_err(|_| { + DocumentCallError::Failed("converted Markdown was not valid UTF-8".to_string()) + }) + } + + /// Inspect a bounded PNG or JPEG image and return its verified facts. + /// + /// # Errors + /// + /// Returns [`DocumentCallError::InvalidInput`] for empty or oversized image + /// data, and another [`DocumentCallError`] when `TinyDocs` cannot inspect it. + pub async fn inspect_image(&self, image: &[u8]) -> Result { + if image.is_empty() || image.len() > MAX_IMAGE_BYTES { + return Err(DocumentCallError::InvalidInput(format!( + "image must contain 1 to {MAX_IMAGE_BYTES} bytes" + ))); + } + self.module + .call_with_stream_detailed( + MODULE_ID, + methods::INSPECT_IMAGE, + |stream| serde_json::json!([stream]), + image, + ) + .await + .map_err(classify_failure) + } + + /// Render selected pages of a PDF and collect every returned PNG. + /// + /// The result preserves the page order declared by `TinyDocs`. + /// + /// # Errors + /// + /// Returns [`DocumentCallError`] when rendering or any page transfer fails. + pub async fn render_pdf( + &self, + document: &[u8], + spec: &RenderPdfSpec, + ) -> Result)>, DocumentCallError> { + let module = self.bind().await?; + let output = Self::output_client(&module)?; + let rendered: RenderedPdf = module + .call_with_stream_detailed( + methods::RENDER_PDF, + |stream| serde_json::json!([spec, stream]), + document, + ) + .await + .map_err(classify_failure)?; + + let mut pending = OutputReleaseGuard { + output: output.clone(), + ids: rendered + .pages + .iter() + .map(|page| page.output.output_id.clone()) + .collect(), + }; + let mut pages = Vec::new(); + let mut first_error = None; + for page in rendered.pages { + let output_id = page.output.output_id.clone(); + // Transfer cleanup responsibility to `collect` before its first + // await. If this future is cancelled, that guard releases this + // output and `pending` releases all pages not yet visited. + pending.ids.retain(|pending_id| pending_id != &output_id); + match collect(&output, page.output).await { + Ok(bytes) if first_error.is_none() => pages.push((page.page, bytes)), + Ok(_) => {} + Err(error) => { + first_error.get_or_insert(error); + } + } + } + if let Some(error) = first_error { + Err(error) + } else { + Ok(pages) + } + } + + async fn bind(&self) -> Result { + self.module.bind(MODULE_ID).await.map_err(classify_failure) + } + + fn output_client( + module: &PinnedModuleClient, + ) -> Result { + module + .object_client(tinydocs_bus::names::OBJECT_PATH) + .map_err(classify_sdk_error) + } +} + +/// Releases retained module outputs when the collection future is cancelled. +struct OutputReleaseGuard { + output: PinnedModuleObjectClient, + ids: Vec, +} + +impl Drop for OutputReleaseGuard { + fn drop(&mut self) { + if self.ids.is_empty() { + return; + } + let Ok(runtime) = tokio::runtime::Handle::try_current() else { + return; + }; + let output = self.output.clone(); + let ids = std::mem::take(&mut self.ids); + runtime.spawn(async move { + for output_id in ids { + match output + .call_detailed::<()>(methods::RELEASE_OUTPUT, (output_id,)) + .await + { + Ok(()) => {} + Err(_) => { + tracing::debug!(module = MODULE_ID, "releasing a TinyDocs output failed"); + } + } + } + }); + } +} + +async fn collect( + output: &PinnedModuleObjectClient, + handle: OutputRef, +) -> Result, DocumentCallError> { + let mut release = OutputReleaseGuard { + output: output.clone(), + ids: vec![handle.output_id.clone()], + }; + let result = read_all(output, &handle).await; + if output + .call_detailed::<()>(methods::RELEASE_OUTPUT, (handle.output_id.clone(),)) + .await + .is_err() + { + tracing::debug!(module = MODULE_ID, "releasing a TinyDocs output failed"); + } + release.ids.clear(); + result +} + +async fn read_all( + output: &PinnedModuleObjectClient, + handle: &OutputRef, +) -> Result, DocumentCallError> { + if handle.total_bytes > MAX_DOCUMENT_BYTES { + return Err(DocumentCallError::Failed(format!( + "the module declared a {}-byte document, over the {MAX_DOCUMENT_BYTES}-byte limit", + handle.total_bytes + ))); + } + + let capacity = usize::try_from(handle.total_bytes).unwrap_or(0); + let max_capacity = usize::try_from(MAX_DOCUMENT_BYTES).unwrap_or(usize::MAX); + let mut bytes = Vec::with_capacity(capacity.min(max_capacity)); + while (bytes.len() as u64) < handle.total_bytes { + let encoded: String = output + .call_detailed( + methods::READ_OUTPUT, + ( + handle.output_id.clone(), + bytes.len() as u64, + READ_CHUNK_BYTES, + ), + ) + .await + .map_err(classify_failure)?; + let chunk = BASE64.decode(encoded).map_err(|_| { + DocumentCallError::Failed("a TinyDocs output chunk was not base64".to_string()) + })?; + if chunk.is_empty() { + return Err(DocumentCallError::Failed( + "the TinyDocs output ended before its declared length".to_string(), + )); + } + if bytes.len().saturating_add(chunk.len()) as u64 > handle.total_bytes { + return Err(DocumentCallError::Failed( + "the TinyDocs output exceeded its declared length".to_string(), + )); + } + bytes.extend_from_slice(&chunk); + } + + if sha256_hex(&bytes) != handle.sha256 { + return Err(DocumentCallError::Failed( + "the assembled TinyDocs output did not match its declared digest".to_string(), + )); + } + Ok(bytes) +} + +fn classify_failure(failure: ModuleCallFailure) -> DocumentCallError { + if failure.wire_name == INVALID_INPUT_WIRE_NAME { + DocumentCallError::InvalidInput(failure.message) + } else if failure.kind == ModuleCallError::Unavailable + || failure.kind == ModuleCallError::IncompatibleContract + || failure.wire_name.contains("ModuleUnavailable") + { + DocumentCallError::Unavailable(failure.to_string()) + } else { + DocumentCallError::Failed(failure.to_string()) + } +} + +fn classify_sdk_error(error: ModuleCallError) -> DocumentCallError { + if matches!( + error, + ModuleCallError::Unavailable | ModuleCallError::IncompatibleContract + ) { + DocumentCallError::Unavailable(error.to_string()) + } else { + DocumentCallError::Failed(error.to_string()) + } +} + +fn sha256_hex(bytes: &[u8]) -> String { + let digest = Sha256::digest(bytes); + let mut output = String::with_capacity(64); + for byte in digest { + use std::fmt::Write as _; + let _ = write!(output, "{byte:02x}"); + } + output +} + +#[cfg(test)] +#[path = "client_tests.rs"] +mod tests; diff --git a/crates/tinydocs-client/src/client_tests.rs b/crates/tinydocs-client/src/client_tests.rs new file mode 100644 index 0000000..e0aa6ae --- /dev/null +++ b/crates/tinydocs-client/src/client_tests.rs @@ -0,0 +1,96 @@ +//! Focused classification and digest tests for the `TinyDocs` client. + +use super::*; + +fn failure(kind: ModuleCallError, wire_name: &str, message: &str) -> ModuleCallFailure { + ModuleCallFailure { + kind, + wire_name: wire_name.to_string(), + message: message.to_string(), + } +} + +#[test] +fn only_the_tinydocs_invalid_input_wire_name_is_model_correctable() { + let invalid = classify_failure(failure( + ModuleCallError::ModuleFault, + INVALID_INPUT_WIRE_NAME, + "invalid section heading", + )); + assert_eq!( + invalid, + DocumentCallError::InvalidInput("invalid section heading".to_string()) + ); + + for wire_name in [ + "ai.tinyhumans.tinydocs.Error.GenerationFailed", + "ai.tinyhumans.tinydocs.Error.ExtractionFailed", + "ai.tinyhumans.tinydocs.Error.TransferFailed", + "ai.tinyhumans.tinydocs.Error.UnknownOutput", + "ai.tinyhumans.tinydocs.Error.SomethingNew", + ] { + assert!(matches!( + classify_failure(failure( + ModuleCallError::ModuleFault, + wire_name, + "untrusted remote text", + )), + DocumentCallError::Failed(_) + )); + } +} + +#[test] +fn unavailable_sdk_outcomes_stay_distinct_from_module_failures() { + assert!(matches!( + classify_failure(failure( + ModuleCallError::Unavailable, + "ai.tinyhumans.tinybus.Error.ModuleUnavailable", + "module is unavailable", + )), + DocumentCallError::Unavailable(_) + )); + assert!(matches!( + classify_sdk_error(ModuleCallError::Unavailable), + DocumentCallError::Unavailable(_) + )); + assert!(matches!( + classify_sdk_error(ModuleCallError::TransportFailed), + DocumentCallError::Failed(_) + )); +} + +fn assert_unavailable_without_remote_detail(error: &DocumentCallError, remote_detail: &str) { + assert!(matches!(error, DocumentCallError::Unavailable(_))); + assert_eq!( + error.raw_message(), + "MODULE_CALL_REPORTED: module incompatible contract" + ); + assert!(!format!("{error:?}").contains(remote_detail)); + assert!(!error.to_string().contains(remote_detail)); +} + +#[test] +fn detailed_incompatible_contract_maps_to_unavailable_without_exposing_peer_text() { + let remote_detail = "untrusted compatibility detail"; + let error = classify_failure(failure( + ModuleCallError::IncompatibleContract, + "ai.tinyhumans.tinybus.Error.IncompatibleVersion", + remote_detail, + )); + assert_unavailable_without_remote_detail(&error, remote_detail); +} + +#[test] +fn sdk_incompatible_contract_maps_to_unavailable_without_exposing_peer_text() { + let error = classify_sdk_error(ModuleCallError::IncompatibleContract); + assert_unavailable_without_remote_detail(&error, "untrusted compatibility detail"); +} + +#[test] +fn the_digest_matches_the_empty_document_vector() { + assert_eq!( + sha256_hex(b""), + "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + ); +} diff --git a/crates/tinydocs-client/src/error.rs b/crates/tinydocs-client/src/error.rs new file mode 100644 index 0000000..85d8502 --- /dev/null +++ b/crates/tinydocs-client/src/error.rs @@ -0,0 +1,62 @@ +//! `TinyDocs` call errors kept distinct for the host's tool and reporting layers. + +use std::fmt::{Debug, Display, Formatter}; + +/// Why a `TinyDocs` operation did not produce its requested result. +/// +/// Remote text is retained for an explicit caller policy, but [`Debug`] and +/// [`Display`] never include it. Module-provided text is untrusted and may +/// contain caller input or credentials. +#[derive(Clone, PartialEq, Eq)] +pub enum DocumentCallError { + /// The module is unavailable or cannot serve this contract. + Unavailable(String), + /// `TinyDocs` rejected caller-provided input. + InvalidInput(String), + /// The operation or its output transfer failed. + Failed(String), +} + +impl Debug for DocumentCallError { + fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result { + let (kind, message) = match self { + Self::Unavailable(_) => ("Unavailable", ""), + Self::InvalidInput(_) => ("InvalidInput", ""), + Self::Failed(_) => ("Failed", ""), + }; + formatter.debug_tuple(kind).field(&message).finish() + } +} + +impl DocumentCallError { + /// Return the untrusted message received from `TinyDocs`, when one exists. + /// + /// Apply an explicit product redaction policy before displaying, storing, + /// or logging this text. [`Debug`] and [`Display`] use only the error + /// category and never include this value. + #[must_use] + pub fn raw_message(&self) -> &str { + match self { + Self::Unavailable(message) | Self::InvalidInput(message) | Self::Failed(message) => { + message + } + } + } +} + +impl Display for DocumentCallError { + fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result { + let category = match self { + Self::Unavailable(_) => "unavailable", + Self::InvalidInput(_) => "input rejected", + Self::Failed(_) => "execution failed", + }; + write!(formatter, "MODULE_CALL_REPORTED: module {category}") + } +} + +impl std::error::Error for DocumentCallError {} + +#[cfg(test)] +#[path = "error_tests.rs"] +mod tests; diff --git a/crates/tinydocs-client/src/error_tests.rs b/crates/tinydocs-client/src/error_tests.rs new file mode 100644 index 0000000..7f9db52 --- /dev/null +++ b/crates/tinydocs-client/src/error_tests.rs @@ -0,0 +1,25 @@ +//! Formatting behavior for `TinyDocs` errors, including message redaction. + +use super::*; + +#[test] +fn debug_and_display_redact_module_messages_but_raw_message_is_explicitly_available() { + let errors = [ + DocumentCallError::Unavailable("module secret".to_string()), + DocumentCallError::InvalidInput("field must be present".to_string()), + DocumentCallError::Failed("remote detail".to_string()), + ]; + for error in errors { + let debug = format!("{error:?}"); + let display = error.to_string(); + assert!(debug.contains("")); + assert!(!debug.contains("secret")); + assert!(!debug.contains("present")); + assert!(!debug.contains("remote detail")); + assert!(!display.contains("secret")); + assert!(!display.contains("present")); + assert!(!display.contains("remote detail")); + } + let invalid = DocumentCallError::InvalidInput("field must be present".to_string()); + assert_eq!(invalid.raw_message(), "field must be present"); +} diff --git a/crates/tinydocs-client/src/lib.rs b/crates/tinydocs-client/src/lib.rs new file mode 100644 index 0000000..8737ed9 --- /dev/null +++ b/crates/tinydocs-client/src/lib.rs @@ -0,0 +1,23 @@ +//! Typed host client for the `tinydocs` `TinyBus` module. +//! +//! Module loading, readiness, host policy, and transport ownership stay with +//! the embedding host. The client accepts that host's `ModuleClient`, then +//! owns `TinyDocs` contract calls and held-output transfer. +//! +//! ```no_run +//! use tinydocs_bus::DocumentSpec; +//! use tinydocs_client::{DocumentCallError, TinyDocsClient}; +//! +//! async fn create_docx( +//! client: &TinyDocsClient, +//! spec: &DocumentSpec, +//! ) -> Result, DocumentCallError> { +//! client.generate_docx(spec).await +//! } +//! ``` + +mod client; +mod error; + +pub use client::TinyDocsClient; +pub use error::DocumentCallError; diff --git a/crates/tinydocs-client/tests/recording_broker.rs b/crates/tinydocs-client/tests/recording_broker.rs new file mode 100644 index 0000000..35ac4a8 --- /dev/null +++ b/crates/tinydocs-client/tests/recording_broker.rs @@ -0,0 +1,1475 @@ +#![expect( + clippy::expect_used, + reason = "fixture assertions should stop immediately on broken broker setup" +)] + +//! Product-facing `TinyDocs` client tests over an in-process recording broker. + +use std::collections::{BTreeMap, HashMap}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use async_trait::async_trait; +use base64::Engine as _; +use base64::engine::general_purpose::STANDARD as BASE64; +use serde_json::{Value, json}; +use sha2::{Digest, Sha256}; +use tinybus::broker::Broker; +use tinybus::module::{ + AccessFuture, FailureReport, FailureReporter, FailureSink, ModuleAccess, ModuleClient, + ProxyFailure, ReleaseRecord, +}; +use tinybus::service::Interface; +use tinybus::transport::memory::MemoryBus; +use tinybus::{Connection, Error, InterfaceName, MemberName, ObjectPath, Proxy}; +use tinydocs_bus::names::{BUS_NAME, METHODS, OBJECT_PATH, methods}; +use tinydocs_bus::{DocumentFormat, DocumentSpec, ExtractDocumentSpec, OutputRef, RenderPdfSpec}; +use tinydocs_client::{DocumentCallError, TinyDocsClient}; +use tokio::sync::Notify; + +const MODULE_ID: &str = "tinydocs"; +const CHUNK_BYTES: u64 = 1024 * 1024; +const CHUNK_BYTES_USIZE: usize = 1024 * 1024; +const MAX_DOCUMENT_BYTES: u64 = 64 * 1024 * 1024; + +static RECORD: ReleaseRecord = ReleaseRecord { + id: MODULE_ID, + bus_name: BUS_NAME, + object_path: OBJECT_PATH, + version: "0.2.5", + release_url: "https://github.com/tinyhumansai/tinydocs/releases/tag/v0.2.5", + assets: &[], +}; + +#[derive(Default)] +struct NoopSink; + +impl FailureSink for NoopSink { + fn capture(&self, _report: &FailureReport<'_>) {} +} + +struct FixtureAccess { + connection: Connection, + reporter: FailureReporter, + enabled: bool, +} + +impl ModuleAccess for FixtureAccess { + fn find(&self, id: &str) -> Option<&'static ReleaseRecord> { + (id == MODULE_ID).then_some(&RECORD) + } + + fn enabled(&self) -> bool { + self.enabled + } + + fn reporter(&self) -> &FailureReporter { + &self.reporter + } + + fn ensure_loaded<'a>(&'a self, _id: &'a str) -> AccessFuture<'a, Result<(), String>> { + Box::pin(async { Ok(()) }) + } + + fn proxy<'a>( + &'a self, + bus_name: &'a str, + object_path: &'a str, + ) -> AccessFuture<'a, Result> { + Box::pin(async move { + self.connection + .proxy(bus_name, object_path, bus_name) + .map_err(|_| ProxyFailure::Contract) + }) + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +struct Call { + member: String, + args: Value, +} + +#[derive(Default)] +struct State { + calls: Vec, + input_streams: BTreeMap>, + outputs: HashMap>, + output_overrides: HashMap>, + generated_bytes: Vec, + next_output: usize, + faults: HashMap, + malformed_replies: HashMap, + declared_size: Option, + declared_digest: Option, + read_mode: ReadMode, + block_reads: bool, + read_started: Arc, + release_seen: Arc, + resume_read: Arc, + release_started: Arc, + release_delay: Option, + handover_to: Option, + detach_on_read: bool, +} + +#[derive(Clone, Copy, Default)] +enum ReadMode { + #[default] + Normal, + Empty, + InvalidBase64, + Overlong, +} + +#[derive(Clone)] +struct RecordingService { + connection: Connection, + state: Arc>, +} + +impl RecordingService { + fn hold(&self, bytes: Vec) -> OutputRef { + let mut state = self.state.lock().expect("fixture state lock"); + state.next_output += 1; + let output_id = format!("output-{}", state.next_output); + let sha256 = state + .declared_digest + .clone() + .unwrap_or_else(|| digest(&bytes)); + let total_bytes = state.declared_size.unwrap_or(bytes.len() as u64); + state.outputs.insert(output_id.clone(), bytes); + OutputRef { + output_id, + total_bytes, + sha256, + } + } + + fn hold_for(&self, member: &str, fallback: &[u8]) -> OutputRef { + let bytes = self + .state + .lock() + .expect("fixture state lock") + .output_overrides + .remove(member) + .unwrap_or_else(|| fallback.to_vec()); + self.hold(bytes) + } + + async fn read_stream_arg( + &self, + args: &Value, + index: usize, + member: &str, + ) -> tinybus::Result<()> { + let stream: tinybus::stream::StreamRef = serde_json::from_value(args[index].clone())?; + let bytes = self.connection.read_stream(&stream).await?; + self.state + .lock() + .expect("fixture state lock") + .input_streams + .insert(member.to_string(), bytes); + Ok(()) + } + + fn record_call(&self, member: &str, args: &Value) -> Option<(String, String)> { + let mut state = self.state.lock().expect("fixture state lock"); + state.calls.push(Call { + member: member.to_string(), + args: args.clone(), + }); + state.faults.get(member).cloned() + } + + fn malformed_reply(&self, member: &str) -> Option { + self.state + .lock() + .expect("fixture state lock") + .malformed_replies + .get(member) + .cloned() + } + + async fn return_fault( + &self, + member: &str, + args: &Value, + (name, message): (String, String), + ) -> tinybus::Result { + let stream_index = match member { + methods::GENERATE_PPTX if !args[1].is_null() => Some(1), + methods::EXTRACT_TEXT | methods::INSPECT_IMAGE => Some(0), + methods::EXTRACT_DOCUMENT | methods::CONVERT_MARKDOWN | methods::RENDER_PDF => Some(1), + _ => None, + }; + if let Some(index) = stream_index { + self.read_stream_arg(args, index, member).await?; + } + Err(Error::MethodFailed { name, message }) + } + + async fn return_malformed_reply( + &self, + member: &str, + args: &Value, + reply: Value, + ) -> tinybus::Result { + if member == methods::EXTRACT_DOCUMENT { + self.read_stream_arg(args, 1, member).await?; + } + Ok(reply) + } + + async fn dispatch_member(&self, member: &str, args: Value) -> tinybus::Result { + match member { + methods::GENERATE_DOCX => { + let bytes = self + .state + .lock() + .expect("fixture state lock") + .generated_bytes + .clone(); + let output = self.hold(bytes); + self.handover_alias().await?; + Ok(serde_json::to_value(output)?) + } + methods::GENERATE_PPTX => { + if !args[1].is_null() { + self.read_stream_arg(&args, 1, member).await?; + } + let output = self.hold_for(member, b"pptx bytes"); + self.handover_alias().await?; + Ok(serde_json::to_value(output)?) + } + methods::EXTRACT_TEXT => { + self.read_stream_arg(&args, 0, member).await?; + let output = self.hold_for(member, b"extracted text"); + self.handover_alias().await?; + Ok(serde_json::to_value(output)?) + } + methods::EXTRACT_DOCUMENT => { + self.read_stream_arg(&args, 1, member).await?; + Ok(json!({ + "format": "pdf", + "section_count": 1, + "sections": [{ + "source": "page:1", + "index": 1, + "text": "section text", + "scanned_candidate": false + }], + "truncated": false + })) + } + methods::CONVERT_MARKDOWN => { + self.read_stream_arg(&args, 1, member).await?; + let output = self.hold_for(member, b"# converted\n"); + self.handover_alias().await?; + Ok(serde_json::to_value(output)?) + } + methods::INSPECT_IMAGE => { + self.read_stream_arg(&args, 0, member).await?; + Ok(json!({ "format": "PNG", "width_px": 2, "height_px": 3 })) + } + methods::RENDER_PDF => { + self.read_stream_arg(&args, 1, member).await?; + let first = self.hold(b"page one".to_vec()); + let second = self.hold(b"page two".to_vec()); + self.handover_alias().await?; + Ok(json!({ + "page_count": 3, + "pages": [ + { "page": 1, "width": 2, "height": 3, "output": first }, + { "page": 3, "width": 4, "height": 5, "output": second } + ] + })) + } + methods::READ_OUTPUT => self.read_output(&args).await, + methods::RELEASE_OUTPUT => { + let output_id = args[0].as_str().expect("output id string"); + let (release_delay, release_started) = { + let state = self.state.lock().expect("fixture state lock"); + (state.release_delay, state.release_started.clone()) + }; + if let Some(delay) = release_delay { + release_started.notify_one(); + tokio::time::sleep(delay).await; + } + self.state + .lock() + .expect("fixture state lock") + .outputs + .remove(output_id); + self.state + .lock() + .expect("fixture state lock") + .release_seen + .notify_one(); + Ok(Value::Null) + } + unknown => Err(Error::MethodFailed { + name: "ai.tinyhumans.tinybus.Error.UnknownMethod".into(), + message: format!("unexpected test member {unknown}"), + }), + } + } +} + +#[async_trait] +impl Interface for RecordingService { + fn name(&self) -> InterfaceName { + InterfaceName::new(BUS_NAME).expect("valid TinyDocs interface") + } + + fn members(&self) -> Vec { + METHODS + .iter() + .map(|member| MemberName::new(*member).expect("valid TinyDocs member")) + .collect() + } + + async fn call(&self, member: &MemberName, args: Value) -> tinybus::Result { + let member = member.as_str().to_string(); + let fault = self.record_call(&member, &args); + if let Some((name, message)) = fault { + return self.return_fault(&member, &args, (name, message)).await; + } + if let Some(reply) = self.malformed_reply(&member) { + return self.return_malformed_reply(&member, &args, reply).await; + } + self.dispatch_member(&member, args).await + } +} + +impl RecordingService { + async fn handover_alias(&self) -> tinybus::Result<()> { + let replacement = self + .state + .lock() + .expect("fixture state lock") + .handover_to + .take(); + if let Some(replacement) = replacement { + self.connection.release_name(BUS_NAME).await?; + replacement.request_name(BUS_NAME).await?; + } + Ok(()) + } + + async fn read_output(&self, args: &Value) -> tinybus::Result { + let output_id = args[0].as_str().expect("output id string").to_string(); + let offset = args[1].as_u64().expect("output offset"); + let requested = args[2].as_u64().expect("requested length"); + let (mode, block, started, resume, detach, bytes) = { + let state = self.state.lock().expect("fixture state lock"); + ( + state.read_mode, + state.block_reads, + state.read_started.clone(), + state.resume_read.clone(), + state.detach_on_read, + state.outputs.get(&output_id).cloned().unwrap_or_default(), + ) + }; + if detach { + self.connection.close().await?; + } + if block { + started.notify_one(); + resume.notified().await; + } + let value = match mode { + ReadMode::Empty => String::new(), + ReadMode::InvalidBase64 => "!".to_string(), + ReadMode::Normal | ReadMode::Overlong => { + let start = usize::try_from(offset).unwrap_or(usize::MAX); + let wanted = usize::try_from(requested).unwrap_or(usize::MAX); + let extra = usize::from(matches!(mode, ReadMode::Overlong)); + let end = start + .saturating_add(wanted) + .saturating_add(extra) + .min(bytes.len()); + BASE64.encode(bytes.get(start..end).unwrap_or_default()) + } + }; + Ok(Value::String(value)) + } +} + +struct Fixture { + client: TinyDocsClient, + state: Arc>, + _broker: tokio::task::JoinHandle>, +} + +async fn test_fixture() -> Fixture { + test_fixture_with_options(true, None).await +} + +async fn test_fixture_with_enabled(enabled: bool) -> Fixture { + test_fixture_with_options(enabled, None).await +} + +async fn test_fixture_with_timeout(timeout: Duration) -> Fixture { + test_fixture_with_options(true, Some(timeout)).await +} + +async fn test_fixture_with_options(enabled: bool, timeout: Option) -> Fixture { + let bus = MemoryBus::new(); + let broker = Broker::new().spawn(bus.clone()); + let server = Connection::connect(bus.connect().await.expect("server transport")) + .await + .expect("server connection"); + let initial_state = State { + generated_bytes: b"docx bytes".to_vec(), + ..State::default() + }; + let state = Arc::new(Mutex::new(initial_state)); + server + .serve_at( + ObjectPath::new(OBJECT_PATH).expect("valid TinyDocs path"), + RecordingService { + connection: server.clone(), + state: state.clone(), + }, + ) + .await + .expect("serve recording module"); + server + .request_name(BUS_NAME) + .await + .expect("claim module name"); + + let client_connection = Connection::connect(bus.connect().await.expect("client transport")) + .await + .expect("client connection"); + let proxy = client_connection + .proxy(BUS_NAME, OBJECT_PATH, BUS_NAME) + .expect("fixture proxy"); + let mut sdk_client = ModuleClient::new(FixtureAccess { + connection: client_connection.clone(), + reporter: FailureReporter::new(Arc::new(NoopSink)), + enabled, + }) + .with_fixture(proxy); + if let Some(timeout) = timeout { + sdk_client = sdk_client.with_timeout(timeout); + } + + Fixture { + client: TinyDocsClient::new(sdk_client), + state, + _broker: broker, + } +} + +struct PeerHandoverFixture { + client: TinyDocsClient, + original: Arc>, + replacement: Arc>, + _broker: tokio::task::JoinHandle>, +} + +async fn peer_handover_fixture() -> PeerHandoverFixture { + let bus = MemoryBus::new(); + let broker = Broker::new().spawn(bus.clone()); + let original_connection = Connection::connect(bus.connect().await.expect("original transport")) + .await + .expect("original connection"); + let replacement_connection = + Connection::connect(bus.connect().await.expect("replacement transport")) + .await + .expect("replacement connection"); + let mut original_state = State { + generated_bytes: b"original bytes".to_vec(), + handover_to: Some(replacement_connection.clone()), + ..State::default() + }; + original_state.output_overrides.insert( + methods::GENERATE_PPTX.into(), + b"original pptx bytes".to_vec(), + ); + let mut replacement_state = State { + generated_bytes: b"replacement bytes".to_vec(), + ..State::default() + }; + replacement_state.output_overrides.insert( + methods::GENERATE_PPTX.into(), + b"replacement pptx bytes".to_vec(), + ); + original_state + .output_overrides + .insert(methods::EXTRACT_TEXT.into(), b"original text".to_vec()); + replacement_state + .output_overrides + .insert(methods::EXTRACT_TEXT.into(), b"replacement text".to_vec()); + original_state.output_overrides.insert( + methods::CONVERT_MARKDOWN.into(), + b"original markdown\n".to_vec(), + ); + replacement_state.output_overrides.insert( + methods::CONVERT_MARKDOWN.into(), + b"replacement markdown\n".to_vec(), + ); + let original = Arc::new(Mutex::new(original_state)); + let replacement = Arc::new(Mutex::new(replacement_state)); + for (connection, state) in [ + (&original_connection, original.clone()), + (&replacement_connection, replacement.clone()), + ] { + connection + .serve_at( + ObjectPath::new(OBJECT_PATH).expect("valid TinyDocs path"), + RecordingService { + connection: connection.clone(), + state, + }, + ) + .await + .expect("serve peer"); + } + original_connection + .request_name(BUS_NAME) + .await + .expect("original owns alias"); + let client_connection = Connection::connect(bus.connect().await.expect("client transport")) + .await + .expect("client connection"); + let access = FixtureAccess { + connection: client_connection, + reporter: FailureReporter::new(Arc::new(NoopSink)), + enabled: true, + }; + PeerHandoverFixture { + client: TinyDocsClient::new(ModuleClient::new(access).with_timeout(Duration::from_secs(2))), + original, + replacement, + _broker: broker, + } +} + +fn digest(bytes: &[u8]) -> String { + let mut output = String::new(); + for byte in Sha256::digest(bytes) { + use std::fmt::Write as _; + write!(&mut output, "{byte:02x}").expect("writing into String"); + } + output +} + +fn document_spec() -> DocumentSpec { + DocumentSpec { + title: "Broker proof".into(), + author: None, + sections: vec![], + } +} + +#[tokio::test] +async fn held_docx_uses_the_peer_that_owns_its_generation() { + let fixture = peer_handover_fixture().await; + let result = fixture.client.generate_docx(&document_spec()).await; + let replacement_control = fixture.client.generate_docx(&document_spec()).await; + + // The producer transfers the well-known name only after retaining its + // output. The later read and release must still reach that same peer. + let original_bytes = result + .as_ref() + .is_ok_and(|bytes| bytes == b"original bytes"); + let replacement_bytes = replacement_control + .as_ref() + .is_ok_and(|bytes| bytes == b"replacement bytes"); + let original = fixture.original.lock().expect("original state lock"); + let replacement = fixture.replacement.lock().expect("replacement state lock"); + let original_counts = ( + original + .calls + .iter() + .filter(|call| call.member == methods::GENERATE_DOCX) + .count(), + original + .calls + .iter() + .filter(|call| call.member == methods::READ_OUTPUT) + .count(), + original + .calls + .iter() + .filter(|call| call.member == methods::RELEASE_OUTPUT) + .count(), + ); + let replacement_counts = ( + replacement + .calls + .iter() + .filter(|call| call.member == methods::GENERATE_DOCX) + .count(), + replacement + .calls + .iter() + .filter(|call| call.member == methods::READ_OUTPUT) + .count(), + replacement + .calls + .iter() + .filter(|call| call.member == methods::RELEASE_OUTPUT) + .count(), + ); + assert!( + original_bytes + && replacement_bytes + && original_counts == (1, 1, 1) + && replacement_counts == (1, 1, 1), + "peer-hop DOCX: original_bytes={original_bytes}, replacement_control_bytes={replacement_bytes}, original generate/read/release={original_counts:?}, replacement generate/read/release={replacement_counts:?}" + ); +} + +#[tokio::test] +async fn held_pptx_upload_and_output_use_the_peer_that_owns_generation() { + let fixture = peer_handover_fixture().await; + let deck = tinydocs_bus::WirePresentationSpec { + title: "Deck".into(), + author: None, + theme: None, + slides: vec![], + }; + let images = b"fresh image bytes"; + let result = fixture.client.generate_pptx(&deck, images).await; + let replacement_control = fixture.client.generate_pptx(&deck, images).await; + + let original_bytes = result + .as_ref() + .is_ok_and(|bytes| bytes == b"original pptx bytes"); + let replacement_bytes = replacement_control + .as_ref() + .is_ok_and(|bytes| bytes == b"replacement pptx bytes"); + let original = fixture.original.lock().expect("original state lock"); + let replacement = fixture.replacement.lock().expect("replacement state lock"); + let original_counts = ( + original + .calls + .iter() + .filter(|call| call.member == methods::GENERATE_PPTX) + .count(), + original + .calls + .iter() + .filter(|call| call.member == methods::READ_OUTPUT) + .count(), + original + .calls + .iter() + .filter(|call| call.member == methods::RELEASE_OUTPUT) + .count(), + ); + let replacement_counts = ( + replacement + .calls + .iter() + .filter(|call| call.member == methods::GENERATE_PPTX) + .count(), + replacement + .calls + .iter() + .filter(|call| call.member == methods::READ_OUTPUT) + .count(), + replacement + .calls + .iter() + .filter(|call| call.member == methods::RELEASE_OUTPUT) + .count(), + ); + let original_stream = original + .input_streams + .get(methods::GENERATE_PPTX) + .map(Vec::as_slice) + == Some(images.as_slice()); + let replacement_stream = replacement + .input_streams + .get(methods::GENERATE_PPTX) + .map(Vec::as_slice) + == Some(images.as_slice()); + assert!( + original_bytes + && replacement_bytes + && original_counts == (1, 1, 1) + && replacement_counts == (1, 1, 1) + && original_stream + && replacement_stream, + "peer-hop PPTX: original_bytes={original_bytes}, replacement_control_bytes={replacement_bytes}, original generate/read/release={original_counts:?}, replacement generate/read/release={replacement_counts:?}, original_stream={original_stream}, replacement_stream={replacement_stream}" + ); +} + +fn output_call_counts(state: &State, producer: &str) -> (usize, usize, usize) { + ( + state + .calls + .iter() + .filter(|call| call.member == producer) + .count(), + state + .calls + .iter() + .filter(|call| call.member == methods::READ_OUTPUT) + .count(), + state + .calls + .iter() + .filter(|call| call.member == methods::RELEASE_OUTPUT) + .count(), + ) +} + +#[tokio::test] +async fn streamed_text_outputs_stay_with_their_producer_after_alias_handover() { + for (member, original_text, replacement_text) in [ + (methods::EXTRACT_TEXT, "original text", "replacement text"), + ( + methods::CONVERT_MARKDOWN, + "original markdown\n", + "replacement markdown\n", + ), + ] { + let fixture = peer_handover_fixture().await; + let first = if member == methods::EXTRACT_TEXT { + fixture.client.extract_text(b"PDF input").await + } else { + fixture + .client + .convert_markdown(DocumentFormat::Pdf, b"PDF input") + .await + }; + let replacement_control = if member == methods::EXTRACT_TEXT { + fixture.client.extract_text(b"PDF input").await + } else { + fixture + .client + .convert_markdown(DocumentFormat::Pdf, b"PDF input") + .await + }; + let original_bytes = first.as_ref().is_ok_and(|text| text == original_text); + let replacement_bytes = replacement_control + .as_ref() + .is_ok_and(|text| text == replacement_text); + let original = fixture.original.lock().expect("original state lock"); + let replacement = fixture.replacement.lock().expect("replacement state lock"); + let original_counts = output_call_counts(&original, member); + let replacement_counts = output_call_counts(&replacement, member); + assert!( + original_bytes + && replacement_bytes + && original_counts == (1, 1, 1) + && replacement_counts == (1, 1, 1), + "peer-hop {member}: original_bytes={original_bytes}, replacement_control_bytes={replacement_bytes}, original producer/read/release={original_counts:?}, replacement producer/read/release={replacement_counts:?}" + ); + } +} + +#[tokio::test] +async fn multi_page_render_outputs_stay_with_their_producer_after_alias_handover() { + let fixture = peer_handover_fixture().await; + let spec = RenderPdfSpec { + pages: vec![1, 3], + max_dimension: 256, + max_total_pixels: 65_536, + max_output_bytes: 1_048_576, + }; + let first = fixture.client.render_pdf(b"PDF input", &spec).await; + let replacement_control = fixture.client.render_pdf(b"PDF input", &spec).await; + let expected = vec![(1, b"page one".to_vec()), (3, b"page two".to_vec())]; + let original_pages = first.as_ref().is_ok_and(|pages| pages == &expected); + let replacement_pages = replacement_control + .as_ref() + .is_ok_and(|pages| pages == &expected); + let original = fixture.original.lock().expect("original state lock"); + let replacement = fixture.replacement.lock().expect("replacement state lock"); + let original_counts = output_call_counts(&original, methods::RENDER_PDF); + let replacement_counts = output_call_counts(&replacement, methods::RENDER_PDF); + assert!( + original_pages + && replacement_pages + && original_counts == (1, 2, 2) + && replacement_counts == (1, 2, 2), + "peer-hop RenderPdf: original_pages={original_pages}, replacement_control_pages={replacement_pages}, original render/read/release={original_counts:?}, replacement render/read/release={replacement_counts:?}" + ); +} + +#[tokio::test] +async fn detached_output_owner_does_not_fall_through_to_alias_replacement() { + let fixture = peer_handover_fixture().await; + fixture + .original + .lock() + .expect("original state lock") + .detach_on_read = true; + // The replacement deliberately has the same opaque resource id and bytes. + // A mistaken alias retry could therefore appear successful without the + // per-peer call counters below. + fixture + .replacement + .lock() + .expect("replacement state lock") + .outputs + .insert("output-1".into(), b"original bytes".to_vec()); + + let result = fixture.client.generate_docx(&document_spec()).await; + assert!(matches!(result, Err(DocumentCallError::Failed(_)))); + let original = fixture.original.lock().expect("original state lock"); + let replacement = fixture.replacement.lock().expect("replacement state lock"); + assert_eq!( + output_call_counts(&original, methods::GENERATE_DOCX), + (1, 1, 0) + ); + assert!( + replacement.calls.is_empty(), + "detached output work must not reach the alias replacement" + ); +} + +#[tokio::test] +async fn typed_operations_use_contract_members_and_stream_document_bytes() { + let fixture = test_fixture().await; + let input = b"incoming PDF bytes"; + + let generated = fixture + .client + .generate_docx(&document_spec()) + .await + .expect("held DOCX should be assembled"); + assert_eq!(generated, b"docx bytes"); + + let deck = tinydocs_bus::WirePresentationSpec { + title: "Deck".into(), + author: None, + theme: None, + slides: vec![], + }; + let slide_images = b"slide image bytes"; + assert_eq!( + fixture + .client + .generate_pptx(&deck, slide_images) + .await + .expect("PPTX output is assembled"), + b"pptx bytes" + ); + assert_eq!( + fixture + .client + .generate_pptx(&deck, &[]) + .await + .expect("text-only PPTX avoids an upload stream"), + b"pptx bytes" + ); + assert_eq!( + fixture + .client + .extract_text(input) + .await + .expect("text is read"), + "extracted text" + ); + assert_eq!( + fixture + .client + .convert_markdown(DocumentFormat::Pdf, input) + .await + .expect("Markdown is read"), + "# converted\n" + ); + + let extracted = fixture + .client + .extract_document(input, &ExtractDocumentSpec::new(DocumentFormat::Pdf)) + .await + .expect("typed extraction succeeds"); + assert_eq!(extracted.sections[0].text, "section text"); + + let image = fixture + .client + .inspect_image(b"PNG stream") + .await + .expect("image facts are decoded"); + assert_eq!((image.width_px, image.height_px), (2, 3)); + + let state = fixture.state.lock().expect("fixture state lock"); + assert_eq!(state.input_streams[methods::EXTRACT_DOCUMENT], input); + assert_eq!(state.input_streams[methods::INSPECT_IMAGE], b"PNG stream"); + assert_eq!(state.input_streams[methods::GENERATE_PPTX], slide_images); + assert_eq!(state.input_streams[methods::EXTRACT_TEXT], input); + assert_eq!(state.input_streams[methods::CONVERT_MARKDOWN], input); + assert!(state.calls.iter().any(|call| { + call.member == methods::GENERATE_DOCX && call.args[0]["title"] == "Broker proof" + })); + assert!( + state + .calls + .iter() + .any(|call| { call.member == methods::GENERATE_PPTX && call.args[1].is_null() }) + ); +} + +#[tokio::test] +async fn disabled_modules_are_unavailable_and_local_image_bounds_send_no_call() { + let fixture = test_fixture_with_enabled(false).await; + assert!(matches!( + fixture.client.generate_docx(&document_spec()).await, + Err(DocumentCallError::Unavailable(_)) + )); + assert_eq!( + fixture + .state + .lock() + .expect("fixture state lock") + .calls + .len(), + 0 + ); + + let fixture = test_fixture().await; + assert!(matches!( + fixture.client.inspect_image(&[]).await, + Err(DocumentCallError::InvalidInput(_)) + )); + let oversized = vec![0; tinydocs_bus::spec::presentation::MAX_IMAGE_BYTES + 1]; + assert!(matches!( + fixture.client.inspect_image(&oversized).await, + Err(DocumentCallError::InvalidInput(_)) + )); + assert_eq!( + fixture + .state + .lock() + .expect("fixture state lock") + .calls + .len(), + 0 + ); +} + +#[tokio::test] +async fn text_outputs_must_be_utf8_and_release_errors_do_not_discard_good_bytes() { + for member in [methods::EXTRACT_TEXT, methods::CONVERT_MARKDOWN] { + let fixture = test_fixture().await; + fixture + .state + .lock() + .expect("fixture state lock") + .output_overrides + .insert(member.to_string(), vec![0xff]); + let result = if member == methods::EXTRACT_TEXT { + fixture.client.extract_text(b"PDF bytes").await.map(|_| ()) + } else { + fixture + .client + .convert_markdown(DocumentFormat::Pdf, b"PDF bytes") + .await + .map(|_| ()) + }; + assert!(matches!(result, Err(DocumentCallError::Failed(_)))); + { + let state = fixture.state.lock().expect("fixture state lock"); + assert!(state.outputs.is_empty()); + assert!( + state + .calls + .iter() + .any(|call| call.member == methods::RELEASE_OUTPUT) + ); + } + } + + let fixture = test_fixture().await; + fixture + .state + .lock() + .expect("fixture state lock") + .faults + .insert( + methods::RELEASE_OUTPUT.to_string(), + ( + "ai.tinyhumans.tinydocs.Error.UnknownOutput".into(), + "already expired".into(), + ), + ); + assert_eq!( + fixture + .client + .generate_docx(&document_spec()) + .await + .expect("a release failure does not invalidate collected bytes"), + b"docx bytes" + ); +} + +#[tokio::test(start_paused = true)] +async fn output_release_waits_for_the_module_clients_configured_deadline() { + let fixture = test_fixture_with_timeout(Duration::from_secs(10)).await; + let release_started = fixture + .state + .lock() + .expect("fixture state lock") + .release_started + .clone(); + fixture + .state + .lock() + .expect("fixture state lock") + .release_delay = Some(Duration::from_secs(6)); + + let client = fixture.client.clone(); + let call = tokio::spawn(async move { client.generate_docx(&document_spec()).await }); + release_started.notified().await; + tokio::time::advance(Duration::from_millis(5_100)).await; + tokio::task::yield_now().await; + assert!( + !call.is_finished(), + "ordinary output release must honor the configured ten-second module deadline" + ); + + tokio::time::advance(Duration::from_secs(1)).await; + assert_eq!( + call.await + .expect("client task joins") + .expect("output is collected"), + b"docx bytes" + ); + assert!( + fixture + .state + .lock() + .expect("fixture state lock") + .outputs + .is_empty() + ); +} + +#[tokio::test] +async fn credential_like_normal_invalid_input_is_redacted_from_formatted_errors() { + let fixture = test_fixture().await; + let remote_text = "credential-like-token-DoNotDisplay"; + fixture + .state + .lock() + .expect("fixture state lock") + .faults + .insert( + methods::GENERATE_DOCX.to_string(), + ( + "ai.tinyhumans.tinydocs.Error.InvalidInput".into(), + remote_text.into(), + ), + ); + let error = fixture + .client + .generate_docx(&document_spec()) + .await + .expect_err("invalid input is returned"); + assert!(matches!(error, DocumentCallError::InvalidInput(_))); + assert_eq!(error.raw_message(), remote_text); + assert!(!error.to_string().contains(remote_text)); + assert!(!format!("{error:?}").contains(remote_text)); +} + +#[tokio::test] +async fn credential_like_held_output_error_is_redacted_from_formatted_errors() { + let fixture = test_fixture().await; + let remote_text = "credential-like-output-token-DoNotDisplay"; + fixture + .state + .lock() + .expect("fixture state lock") + .faults + .insert( + methods::READ_OUTPUT.to_string(), + ( + "ai.tinyhumans.tinydocs.Error.InvalidInput".into(), + remote_text.into(), + ), + ); + let error = fixture + .client + .generate_docx(&document_spec()) + .await + .expect_err("held output error is returned"); + assert!(matches!(error, DocumentCallError::InvalidInput(_))); + assert_eq!(error.raw_message(), remote_text); + assert!(!error.to_string().contains(remote_text)); + assert!(!format!("{error:?}").contains(remote_text)); + assert!( + fixture + .state + .lock() + .expect("fixture state lock") + .outputs + .is_empty() + ); +} + +#[tokio::test] +async fn ordinary_module_wire_failures_keep_input_availability_and_execution_distinct() { + let fixture = test_fixture().await; + fixture + .state + .lock() + .expect("fixture state lock") + .faults + .insert( + methods::GENERATE_DOCX.to_string(), + ( + "ai.tinyhumans.tinydocs.Error.InvalidInput".into(), + "title is empty".into(), + ), + ); + assert!(matches!( + fixture.client.generate_docx(&document_spec()).await, + Err(DocumentCallError::InvalidInput(_)) + )); + + fixture + .state + .lock() + .expect("fixture state lock") + .faults + .insert( + methods::GENERATE_DOCX.to_string(), + ( + "ai.tinyhumans.tinydocs.Error.GenerationFailed".into(), + "private generator detail".into(), + ), + ); + assert!(matches!( + fixture + .client + .generate_docx(&document_spec()) + .await + .expect_err("generation failure is returned"), + DocumentCallError::Failed(_) + )); + + fixture + .state + .lock() + .expect("fixture state lock") + .faults + .insert( + methods::GENERATE_DOCX.to_string(), + ( + "ai.tinyhumans.tinybus.Error.ModuleUnavailable".into(), + "module is down".into(), + ), + ); + assert!(matches!( + fixture.client.generate_docx(&document_spec()).await, + Err(DocumentCallError::Unavailable(_)) + )); +} + +#[tokio::test] +async fn streamed_module_wire_failures_keep_input_and_execution_distinct() { + let fixture = test_fixture().await; + fixture + .state + .lock() + .expect("fixture state lock") + .faults + .insert( + methods::EXTRACT_DOCUMENT.to_string(), + ( + "ai.tinyhumans.tinydocs.Error.InvalidInput".into(), + "PDF header is invalid".into(), + ), + ); + assert!(matches!( + fixture + .client + .extract_document( + b"invalid PDF", + &ExtractDocumentSpec::new(DocumentFormat::Pdf) + ) + .await, + Err(DocumentCallError::InvalidInput(_)) + )); + + let fixture = test_fixture().await; + fixture + .state + .lock() + .expect("fixture state lock") + .faults + .insert( + methods::INSPECT_IMAGE.to_string(), + ( + "ai.tinyhumans.tinydocs.Error.ExtractionFailed".into(), + "image parser failed".into(), + ), + ); + assert!(matches!( + fixture.client.inspect_image(b"PNG bytes").await, + Err(DocumentCallError::Failed(_)) + )); +} + +#[tokio::test] +async fn held_output_invalid_input_is_classified_and_still_released() { + let fixture = test_fixture().await; + fixture + .state + .lock() + .expect("fixture state lock") + .faults + .insert( + methods::READ_OUTPUT.to_string(), + ( + "ai.tinyhumans.tinydocs.Error.InvalidInput".into(), + "output range is invalid".into(), + ), + ); + assert!(matches!( + fixture.client.generate_docx(&document_spec()).await, + Err(DocumentCallError::InvalidInput(_)) + )); + let released = { + let state = fixture.state.lock().expect("fixture state lock"); + state + .calls + .iter() + .any(|call| call.member == methods::RELEASE_OUTPUT) + && state.outputs.is_empty() + }; + assert!(released); +} + +#[tokio::test] +async fn malformed_typed_stream_reply_is_unavailable_for_contract_mismatch() { + let fixture = test_fixture().await; + fixture + .state + .lock() + .expect("fixture state lock") + .malformed_replies + .insert( + methods::EXTRACT_DOCUMENT.to_string(), + json!({"truncated": "wrong type"}), + ); + let error = fixture + .client + .extract_document(b"PDF bytes", &ExtractDocumentSpec::new(DocumentFormat::Pdf)) + .await + .expect_err("a malformed typed reply is rejected"); + assert!(matches!(error, DocumentCallError::Unavailable(_))); +} + +#[tokio::test] +async fn held_outputs_are_read_in_order_and_released_after_success() { + let fixture = test_fixture().await; + let bytes = vec![0x5a; CHUNK_BYTES_USIZE * 2 + 13]; + { + let mut state = fixture.state.lock().expect("fixture state lock"); + state.generated_bytes = bytes.clone(); + state.declared_digest = Some(digest(&bytes)); + } + let got = fixture + .client + .generate_docx(&document_spec()) + .await + .expect("output is assembled"); + assert_eq!(got, bytes); + let offsets = fixture + .state + .lock() + .expect("fixture state lock") + .calls + .iter() + .filter(|call| call.member == methods::READ_OUTPUT) + .map(|call| call.args[1].as_u64().expect("read offset")) + .collect::>(); + assert_eq!(offsets, [0, CHUNK_BYTES, CHUNK_BYTES * 2]); + assert!( + fixture + .state + .lock() + .expect("fixture state lock") + .outputs + .is_empty() + ); +} + +#[tokio::test] +async fn held_outputs_with_bad_digest_or_declared_size_are_released() { + let fixture = test_fixture().await; + { + let mut state = fixture.state.lock().expect("fixture state lock"); + state.generated_bytes = b"digest must reject this output".to_vec(); + state.declared_digest = Some("0".repeat(64)); + } + assert!(matches!( + fixture.client.generate_docx(&document_spec()).await, + Err(DocumentCallError::Failed(_)) + )); + { + let state = fixture.state.lock().expect("fixture state lock"); + assert!( + state + .calls + .iter() + .any(|call| call.member == methods::RELEASE_OUTPUT) + ); + assert!(state.outputs.is_empty()); + } + + let fixture = test_fixture().await; + fixture + .state + .lock() + .expect("fixture state lock") + .declared_size = Some(MAX_DOCUMENT_BYTES + 1); + assert!(matches!( + fixture.client.generate_docx(&document_spec()).await, + Err(DocumentCallError::Failed(_)) + )); + { + let state = fixture.state.lock().expect("fixture state lock"); + assert!( + !state + .calls + .iter() + .any(|call| call.member == methods::READ_OUTPUT) + ); + assert!( + state + .calls + .iter() + .any(|call| call.member == methods::RELEASE_OUTPUT) + ); + } +} + +#[tokio::test] +async fn truncated_invalid_and_overlong_chunks_are_released() { + for mode in [ReadMode::Empty, ReadMode::InvalidBase64, ReadMode::Overlong] { + let fixture = test_fixture().await; + { + let mut state = fixture.state.lock().expect("fixture state lock"); + state.read_mode = mode; + if matches!(mode, ReadMode::Overlong) { + state.generated_bytes = vec![0x33; CHUNK_BYTES_USIZE + 16]; + state.declared_size = Some(CHUNK_BYTES); + let declared_digest = digest(&state.generated_bytes); + state.declared_digest = Some(declared_digest); + } + } + assert!(matches!( + fixture.client.generate_docx(&document_spec()).await, + Err(DocumentCallError::Failed(_)) + )); + assert!( + fixture + .state + .lock() + .expect("fixture state lock") + .calls + .iter() + .any(|call| call.member == methods::RELEASE_OUTPUT) + ); + } +} + +#[tokio::test] +async fn cancellation_releases_held_output_and_render_releases_every_page() { + let fixture = test_fixture().await; + fixture + .state + .lock() + .expect("fixture state lock") + .block_reads = true; + let started = fixture + .state + .lock() + .expect("fixture state lock") + .read_started + .clone(); + let release_seen = fixture + .state + .lock() + .expect("fixture state lock") + .release_seen + .clone(); + let client = fixture.client.clone(); + let call = tokio::spawn(async move { client.generate_docx(&document_spec()).await }); + started.notified().await; + call.abort(); + let _ = call.await; + release_seen.notified().await; + let resume_read = fixture + .state + .lock() + .expect("fixture state lock") + .resume_read + .clone(); + resume_read.notify_one(); + assert!( + fixture + .state + .lock() + .expect("fixture state lock") + .outputs + .is_empty() + ); + + let fixture = test_fixture().await; + let rendered = fixture + .client + .render_pdf( + b"PDF input", + &RenderPdfSpec { + pages: vec![1, 3], + max_dimension: 256, + max_total_pixels: 65_536, + max_output_bytes: 1_048_576, + }, + ) + .await + .expect("selected pages are collected"); + assert_eq!( + rendered.iter().map(|(page, _)| *page).collect::>(), + vec![1, 3] + ); + assert!( + fixture + .state + .lock() + .expect("fixture state lock") + .outputs + .is_empty() + ); + + let fixture = test_fixture().await; + fixture.state.lock().expect("fixture state lock").read_mode = ReadMode::InvalidBase64; + assert!(matches!( + fixture + .client + .render_pdf( + b"PDF input", + &RenderPdfSpec { + pages: vec![1, 3], + max_dimension: 256, + max_total_pixels: 65_536, + max_output_bytes: 1_048_576, + }, + ) + .await, + Err(DocumentCallError::Failed(_)) + )); + let state = fixture.state.lock().expect("fixture state lock"); + assert!(state.outputs.is_empty()); + assert_eq!( + state + .calls + .iter() + .filter(|call| call.member == methods::RELEASE_OUTPUT) + .count(), + 2 + ); +} diff --git a/crates/tinydocs-module/Cargo.toml b/crates/tinydocs-module/Cargo.toml index 6daee69..233fc2e 100644 --- a/crates/tinydocs-module/Cargo.toml +++ b/crates/tinydocs-module/Cargo.toml @@ -21,9 +21,9 @@ tinydocs = { path = "../..", default-features = false, features = ["docx", "pptx # The transport-free vocabulary shared with every TinyBus host. tinydocs-bus = { path = "../tinydocs-bus" } # TinyBus provides the typed service interface and dynamic module host ABI. -tinybus = { version = ">=0.1.3, <0.2", default-features = false, features = ["macros", "modules"] } +tinybus = { version = "0.3.2", default-features = false, features = ["macros", "modules"] } # The module-side SDK owns its runtime and exports the stable C entrypoints. -tinybus-module = ">=0.1.3, <0.2" +tinybus-module = "0.3.2" # Module methods move CPU-bound document synthesis onto Tokio's blocking pool. tokio = { version = "1", features = ["macros", "rt-multi-thread"] } # Bus frames are JSON, where a byte array costs ~3.5 bytes per byte. Chunks are @@ -54,4 +54,4 @@ base64 = "0.23" serde_json = "1" zip = { version = "8", default-features = false, features = ["deflate"] } # Enable TinyBus loader helpers only in test builds, not downstream production builds. -tinybus = { version = ">=0.1.3, <0.2", default-features = false, features = ["test-support"] } +tinybus = { version = "0.3.2", default-features = false, features = ["test-support"] } diff --git a/vendor/tinybus b/vendor/tinybus index eb5cf72..cba2de0 160000 --- a/vendor/tinybus +++ b/vendor/tinybus @@ -1 +1 @@ -Subproject commit eb5cf727071a6eb12f9efa3db8ec59e54039f336 +Subproject commit cba2de0e573f99fe4c7e45ba14e2d6b5a2a7757f