Repository navigation
Conversation
- delete `sensd_client.rs`, `sensor_guard.rs`, and related transport-era helpers from the SDK - start rebasing stage-as-you-go, acquisition, and the node crates on JetStream-backed runtime surfaces
…test reports Refresh `docs/vision/project-target-state.md` for the JetStream-first architecture and move the large testing-analysis artifacts under a dedicated docs namespace.
- rework `service.rs`, `material_assembler.rs`, and `validator.rs` around the JetStream-only flow - update checkpoint/repository tests so the new pipeline is exercised end to end
Harden validation, blob retrieval, and gateway service-container paths so the post-gRPC runtime does not keep stale assumptions around content lookup and event acceptance.
Clean up the default module imports, tighten the example modules, and refresh the VM scenarios and documentation around the revised option layout.
Collapse the nested NixOS service-module layering, simplify the core database and monitoring modules, and refresh the VM scenarios that depend on that structure.
Tighten gateway blob handling, ingestd JetStream and material assembly error paths, and the surrounding Nix and test wiring that exercises those integrations.
- remove `figment_config.rs` from ingestd and the SDK - keep config loading in the main config modules and add targeted config-loading tests instead
…ager Rework the document ingestor so retrieval and related code paths go through BlobManager rather than the older direct access helpers.
Relocate the CLI, runner, and stream runtime files under `src/runtime/` while keeping short-lived reexports so node ports can happen incrementally.
Introduce `handles.rs`, split out checkpoint/stats/time-horizon helpers, and give processors a stable handle bundle for runtime-derived builders.
…dles/ServiceInfo runtime state Migrate the terminal satellite processor off direct `StreamProcessorContext` dependency and onto handle-based runtime initialization. The processor now stores `ProcessorHandles` + `ServiceInfo`, bootstraps streams from transport handles, and builds `StageAsYouGoContext` from `EventEmitter` rather than raw sender channels. Runtime-aware init paths are added (`initialize_with_runtime`) and scan/watch logic is updated to use handle/service accessors for host identity, DB pool, and emitter usage. Supporting stage-as-you-go internals are adjusted to accept `EventEmitter` and emit via unified helpers, making terminal ingestion behavior consistent with the new runtime abstraction used by other processors.
- move the remaining node and automaton crates onto runtime-state-based startup - keep bridge compatibility just long enough to avoid a single giant flag day
Update processors, examples, and tests so initialization flows through `ProcessorInitContext`, paving the way for deleting `StreamProcessorContext` in the next step.
Remove the old context type once the runtime-state ports are in place, leaving runtime-state-only processor plumbing as the active path.
…s on ProcessorRuntimeState Extend `ProcessorRuntimeState` with high-frequency helpers that processors repeatedly needed after the `StreamProcessorContext` removal: typed config extraction (`config_value`), raw config lookups (`raw_config_value`), direct event emission (`emit_event`), and a std `Path` view of the working directory (`work_dir`). The commit also pulls in the required event/path imports so processors can operate through runtime state without rebuilding wrappers. In this band, it is an ergonomics consolidation that reduces boilerplate before the stage-context runtime binding landed.
Replace the older constructor trio with runtime- and sender-based StageAsYouGo helpers and update the processors and tests to use those builders.
…andles Teach AcquisitionManager to derive its NATS and database wiring from ProcessorHandles and update the file and terminal processors to use that entry point.
…faces Add convenience constructors that derive ReplayManager and StageAsYouGoContext from runtime state or handles and switch the affected processors to those helpers.
Add a ProcessorRuntimeState helper that builds AcquisitionManager instances from the active handles so processors stop reassembling that wiring by hand.
…nd add reusable integration-test runtime fixture Add runtime-native heartbeat construction so processors no longer hand-thread service names when emitting liveness signals. `HeartbeatEmitter::from_runtime` and `ProcessorRuntimeState::heartbeat_emitter` now build emitters from `ServiceInfo`, aligning heartbeat setup with other runtime helpers added in nearby commits. The same change introduces `tests/support/runtime.rs`, a reusable `TestRuntime` harness that assembles ephemeral NATS, checkpoint manager, event emitter channel, and `ProcessorRuntimeState` in one place for SDK integration tests. This commit tightens both production ergonomics and test setup consistency for runtime-handle migration work.
…oss nodes - add the new runtime crate for shared CLI and runner surfaces - migrate node crates and replay/runtime modules onto the extracted package while trimming old compatibility shims
Repair `types/events/payloads/system.rs` and the affected runtime imports so node crates keep the system event construction helpers they relied on before the extraction.
…ing warning cleanup Trim dead code in `JetStreamConsumer` by deleting the unused `env: SinexEnvironment` field and its local initialization in `new()`. No behavior changes: the consumer still builds JetStream context, DB pool, validator, topology, and stats exactly as before. In the local sequence this acts as a hygiene commit between larger runtime/payload and canonicalizer refactors, reducing warning noise so substantive structural changes are easier to review.
…ied processors only Remove the old terminal-command canonicalizer implementation path and leave the unified processor route as the only supported path.
Align ingestd and its tests with the `events.raw` subject layout, then add stream-name and validator-order coverage so the subject switch is locked down.
Thread the agenix input through the default module and point the example configs at the new secret-backed option layout.
- add `nixos/modules/secrets.nix` - introduce the encrypted `secret/` repository scaffold and wire the new flake input
Tighten repository behavior and add coverage for cascade cycles, event ordering, integrity, and provenance-sensitive insert/query paths after the runtime refactor.
Sinity
added
area:core
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
|
Important Review skippedToo many files! This PR contains 236 files, which is 86 over the limit of 150. ⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Pro Run ID: ⛔ Files ignored due to path filters (12)
📒 Files selected for processing (236)
You can disable this status message by setting the Use the checkbox below for a quick retry:
Note
|
Sinity
added a commit
that referenced
this pull request
Apr 9, 2026
#94) Finish the sensd-era cleanup by deleting the last transport helpers, move the shared processor runtime into its own crate, and rebase ingestd, node crates, and NixOS wiring on the JetStream-only runtime model. transport and runtime extraction: - delete `crate/lib/sinex-satellite-sdk/src/{sensd_client.rs,sensor_guard.rs}` and remove the last gRPC/protobuf workspace deps from `Cargo.toml`, which strips the SDK down to the JetStream-era transport model - add `crate/lib/sinex-processor-runtime/{Cargo.toml,src/cli.rs,src/runner.rs,src/lib.rs}` and move the shared CLI/runner plumbing out of `sinex-satellite-sdk`, so runtime concerns stop living in the same crate as sensor helpers and annex utilities - relocate `stream_processor.rs` under `crate/lib/sinex-satellite-sdk/src/runtime/stream/`, add `ProcessorHandles`, `ProcessorRuntimeState`, and `runtime_state.rs`, and then delete `StreamProcessorContext`, making runtime-state-only initialization the single processor path node and SDK adoption: - port the terminal, filesystem, desktop, system, document, analytics, search, content, PKM, and health crates to `ProcessorRuntimeState` while keeping short-lived compatibility reexports during the transition - make `ProcessorInitContext` the only `StatefulStreamProcessor` initialization API and add convenience builders for config lookup, emission, workdir access, lifecycle management, coordination, replay, stage-as-you-go, heartbeat, and acquisition - add `StageAsYouGoContext::from_handles` and `AcquisitionManager::from_handles`, so node crates can build staging and capture helpers directly from runtime handles without the deleted sensd abstractions ingestd, gateway, and core hardening: - rework `crate/core/sinex-ingestd/src/{service.rs,material_assembler.rs,jetstream_consumer.rs,validator.rs}` around JetStream-only subjects, checkpoint flow, blob handling, and stricter validation, and switch ingestd to `events.raw` subjects with hardened consumer tests - tighten `crate/core/sinex-gateway/src/{handlers.rs,rpc_server.rs,service_container.rs}` and route document retrieval through `BlobManager`, so gateway and document ingest stop depending on stale transport-era assumptions - restore system payload builders in `crate/lib/sinex-core/src/types/events/payloads/system.rs`, remove the legacy inline terminal canonicalizer path, and harden repository/provenance/cascade tests in `crate/lib/sinex-core/tests/*` nixos and secrets: - flatten `nixos/modules/default.nix`, inline service module composition, and rename lifecycle options such as `services.sinex.preflightVerification.enable` to `services.sinex.lifecycle.preflight.enable` - add `nixos/modules/secrets.nix`, inject the `agenix` flake input, and scaffold the encrypted `secret/` repository, so the default module can express secret-backed deployments without bespoke local wiring
There was a problem hiding this comment.
Pull request overview
Note
Copilot was unable to run its full agentic suite in this review.
Extracts processor runtime/CLI concerns into a dedicated sinex-processor-runtime crate, removes the remaining sensd-era helpers, and migrates satellites/ingestd to a JetStream-only, runtime-state-first model (with accompanying schema + test updates).
Changes:
- Introduces
sinex-processor-runtimeand updates satellites to useprocessor_main!from the new crate. - Replaces
StreamProcessorContextplumbing withProcessorInitContext+ProcessorRuntimeStateacross processors and helpers. - Tightens JetStream + schema/checkpoint behavior and refreshes tests/utilities accordingly.
Reviewed changes
Copilot reviewed 150 out of 248 changed files in this pull request and generated 9 comments.
Show a summary per file
| File | Description |
|---|---|
| crate/satellites/sinex-document-ingestor/Cargo.toml | Adds runtime crate + camino to support runtime-state model. |
| crate/satellites/sinex-desktop-satellite/src/main.rs | Switches processor_main! to sinex_processor_runtime. |
| crate/satellites/sinex-desktop-satellite/src/lib.rs | Re-exports runtime CLI types + migrates to runtime init/state types. |
| crate/satellites/sinex-desktop-satellite/Cargo.toml | Adds dependency on sinex-processor-runtime. |
| crate/satellites/sinex-content-automaton/src/lib.rs | Migrates automaton initialization/event sending to runtime state + unbounded channel. |
| crate/satellites/sinex-analytics-automaton/src/main.rs | Switches processor_main! to sinex_processor_runtime. |
| crate/satellites/sinex-analytics-automaton/src/lib.rs | Migrates automaton to runtime state, adds new StatefulStreamProcessor impl. |
| crate/satellites/sinex-analytics-automaton/Cargo.toml | Adds dependency on sinex-processor-runtime. |
| crate/lib/sinex-test-utils/tests/rstest_integration_example.rs | Gates example test behind rstest-preview feature. |
| crate/lib/sinex-test-utils/tests/modern_test_infrastructure_example.rs | Gates example test behind rstest-preview feature. |
| crate/lib/sinex-test-utils/tests/macro_conversion_examples.rs | Gates example test behind rstest-preview feature. |
| crate/lib/sinex-test-utils/tests/integration/stream_processing_test.rs | Updates JetStream consumer creation APIs in tests. |
| crate/lib/sinex-test-utils/src/nats.rs | Improves missing nats-server error message. |
| crate/lib/sinex-test-utils/src/lib.rs | Adds jetstream module + re-export helper. |
| crate/lib/sinex-test-utils/src/jetstream.rs | Adds helper to ensure material JetStream streams exist in tests. |
| crate/lib/sinex-test-utils/src/fixtures.rs | Fixes material registry setup + renames checkpoint payload field; adds dataset distribution test. |
| crate/lib/sinex-test-utils/src/db_common.rs | Updates processor manifest schema/table references. |
| crate/lib/sinex-test-utils/src/database_pool.rs | Strengthens schema inspection + improves DB connection error messages + installs extensions. |
| crate/lib/sinex-test-utils/src/builders.rs | Renames test checkpoint builder API to checkpoint_data. |
| crate/lib/sinex-test-utils/Cargo.toml | Adds rstest-preview feature; removes tonic dependency. |
| crate/lib/sinex-services/src/search.rs | Normalizes pagination bounds and adds unit test. |
| crate/lib/sinex-services/Cargo.toml | Adds parking_lot and bon dependencies. |
| crate/lib/sinex-schema/tests/serde_tests.rs | Updates checkpoint record serde test with new fields. |
| crate/lib/sinex-schema/src/schema/processors.rs | Adds checkpoint metadata columns (checkpoint_version, created_at). |
| crate/lib/sinex-schema/src/migrations/mod.rs | Removes outdated migration comment. |
| crate/lib/sinex-schema/doc/schema_design.md | Updates docs to core.processor_checkpoints. |
| crate/lib/sinex-satellite-sdk/tests/support/runtime.rs | Adds runtime test builder for fully wired processor runtimes. |
| crate/lib/sinex-satellite-sdk/tests/support/mod.rs | Adds test support module. |
| crate/lib/sinex-satellite-sdk/tests/sensor_guard.rs | Removes sensor guard test (sensor guard removed). |
| crate/lib/sinex-satellite-sdk/tests/replay_stage_flow.rs | Adds replay-to-emitter integration tests using runtime builder. |
| crate/lib/sinex-satellite-sdk/tests/replay_control.rs | Updates import path for replay controller. |
| crate/lib/sinex-satellite-sdk/tests/material_acquisition.rs | Updates ingestd test config usage; adds restart recovery test. |
| crate/lib/sinex-satellite-sdk/tests/integration/stage_as_you_go_integration_test.rs | Migrates stage-as-you-go integration away from gRPC client; uses runtime sender + processor. |
| crate/lib/sinex-satellite-sdk/tests/integration/satellite_architecture_test.rs | Removes gRPC client expectation; aligns to unified runtime model. |
| crate/lib/sinex-satellite-sdk/tests/figment_config.rs | Removes obsolete figment config tests. |
| crate/lib/sinex-satellite-sdk/tests/config_loading_tests.rs | Adds config loading tests for new Figment-based config loaders. |
| crate/lib/sinex-satellite-sdk/src/sensor_guard.rs | Removes sensd-era sensor guard plumbing. |
| crate/lib/sinex-satellite-sdk/src/runtime/stream/time_horizon.rs | Adds TimeHorizon under runtime module. |
| crate/lib/sinex-satellite-sdk/src/runtime/stream/stats.rs | Adds processing stats type for runtime/stream. |
| crate/lib/sinex-satellite-sdk/src/runtime/stream/runtime_state.rs | Introduces ProcessorRuntimeState runtime snapshot with capability accessors. |
| crate/lib/sinex-satellite-sdk/src/runtime/stream/handles.rs | Introduces ProcessorHandles, ServiceInfo, EventEmitter, ProcessorInitContext. |
| crate/lib/sinex-satellite-sdk/src/runtime/stream/checkpoint.rs | Adds Checkpoint type under runtime module. |
| crate/lib/sinex-satellite-sdk/src/runtime/mod.rs | Adds runtime module entrypoint and re-exports. |
| crate/lib/sinex-satellite-sdk/src/replay_control.rs | Removes legacy replay control module (moved under replay/). |
| crate/lib/sinex-satellite-sdk/src/replay/progress.rs | Adds replay progress tracking types. |
| crate/lib/sinex-satellite-sdk/src/replay/mod.rs | Adds replay module surface and re-exports. |
| crate/lib/sinex-satellite-sdk/src/replay/metrics.rs | Adds replay metrics tracking. |
| crate/lib/sinex-satellite-sdk/src/replay/control.rs | Adds replay pause/resume controller under new module. |
| crate/lib/sinex-satellite-sdk/src/prelude.rs | Updates prelude exports to new replay/runtime surfaces. |
| crate/lib/sinex-satellite-sdk/src/preflight/database.rs | Updates critical table names to core.processor_checkpoints. |
| crate/lib/sinex-satellite-sdk/src/lifecycle.rs | Adds lifecycle manager construction from runtime + heartbeat hydration. |
| crate/lib/sinex-satellite-sdk/src/lib.rs | Removes legacy modules; re-exports runtime stream module; adds replay exports. |
| crate/lib/sinex-satellite-sdk/src/job_manager.rs | Adds constructors from handles/runtime. |
| crate/lib/sinex-satellite-sdk/src/heartbeat.rs | Adds heartbeat emitter construction from runtime. |
| crate/lib/sinex-satellite-sdk/src/examples/filesystem_processor.rs | Migrates example processor to runtime init/state and removes CLI exploration impl. |
| crate/lib/sinex-satellite-sdk/src/error_helpers.rs | Generalizes config access to work with ProcessorRuntimeState or config maps. |
| crate/lib/sinex-satellite-sdk/src/coordination.rs | Adds coordination construction from runtime. |
| crate/lib/sinex-satellite-sdk/src/config.rs | Reworks config loading using Figment (files + env + defaults). |
| crate/lib/sinex-satellite-sdk/src/checkpoint.rs | Updates docs + record shape for new checkpoint fields/table. |
| crate/lib/sinex-satellite-sdk/src/annex/blob_manager.rs | Secures temp file creation for blob ingest. |
| crate/lib/sinex-satellite-sdk/src/acquisition_manager.rs | Adds JetStream stream bootstrap + from-handles constructor; adjusts ledger metadata. |
| crate/lib/sinex-satellite-sdk/doc/grpc_client.md | Removes obsolete gRPC client doc. |
| crate/lib/sinex-satellite-sdk/doc/README.md | Updates docs with key runtime entry points and test runtime builder. |
| crate/lib/sinex-satellite-sdk/Cargo.toml | Adds tempfile dependency for secure temp file handling. |
| crate/lib/sinex-processor-runtime/tests/processor_runner.rs | Updates runtime/runner test to use ProcessorInitContext + new crate exports. |
| crate/lib/sinex-processor-runtime/src/runner.rs | Adjusts imports after extraction into new crate. |
| crate/lib/sinex-processor-runtime/src/lib.rs | Introduces new crate public surface (CLI, runner, replay helpers). |
| crate/lib/sinex-processor-runtime/Cargo.toml | Adds new runtime crate manifest and dependencies. |
| crate/lib/sinex-macros/src/stream_processor.rs | Updates generated CLI integration to call new processor_main!. |
| crate/lib/sinex-core/tests/system/reliability_test.rs | Updates checkpoint column names (checkpoint_data). |
| crate/lib/sinex-core/tests/repositories_schema_management.rs | Adds tests for schema hashing/reactivation/failure invariants. |
| crate/lib/sinex-core/tests/repositories_events.rs | Adds guard test ensuring cleanup doesn’t delete production-like events. |
| crate/lib/sinex-core/tests/property/event_model_fuzzing_test.rs | Renames payload field state_data → checkpoint_data. |
| crate/lib/sinex-core/tests/performance/resource_exhaustion_test.rs | Updates JetStream stream/consumer APIs in perf tests. |
| crate/lib/sinex-core/tests/performance/large_payload_test.rs | Updates JetStream consumer APIs in perf tests. |
| crate/lib/sinex-core/tests/performance/jetstream_performance_test.rs | Updates JetStream consumer deletion/creation APIs. |
| crate/lib/sinex-core/tests/performance/checkpoint_performance_test.rs | Updates JetStream stream/consumer APIs in perf tests. |
| crate/lib/sinex-core/tests/performance/bottleneck_identification_test.rs | Updates JetStream consumer APIs and stream info calls. |
| crate/lib/sinex-core/tests/integration/work_queue_test.rs | Updates JetStream usage + publisher API calls. |
| crate/lib/sinex-core/tests/integration/timestamp_test.rs | Fixes ULID access patterns; reduces unwraps; updates formatting and imports. |
| crate/lib/sinex-core/tests/integration/subscription_service_test.rs | Normalizes pool access and result type aliases. |
| crate/lib/sinex-core/tests/integration/single_writer_enforcement_test.rs | Normalizes pool access and result type aliases; cleans imports. |
| crate/lib/sinex-core/tests/integration/provenance_test.rs | Removes duplicate import; normalizes result alias usage. |
| crate/lib/sinex-core/tests/integration/integrity_tester_test.rs | Adds integration test covering integrity tester windowing behavior. |
| crate/lib/sinex-core/tests/integration/ingest_service_test.rs | Normalizes result alias usage across tests. |
| crate/lib/sinex-core/tests/integration/distributed_locking_test.rs | Normalizes result alias usage and imports. |
| crate/lib/sinex-core/tests/adversarial/chaos_engineering_test.rs | Updates processor manifest table references to core. |
| crate/lib/sinex-core/src/types/events/payloads/mod.rs | Adds macro to define typed payload structs consistently. |
| crate/lib/sinex-core/src/environment.rs | Tightens environment namespacing checks for DB/workdir/socket paths. |
| crate/lib/sinex-core/src/db/repositories/state.rs | Increments checkpoint versions; fixes ULID/UUID parameter typing for tx updates. |
| crate/lib/sinex-core/src/db/repositories/events_extensions.rs | Propagates conversion failures via try_to_event() in repository method. |
| crate/lib/sinex-core/src/db/repositories/checkpoints.rs | Adds checkpoint metadata columns to record/repo queries/updates. |
| crate/lib/sinex-core/Cargo.toml | Adds async-nats for test/perf modules. |
| crate/core/sinex-rpc-dispatcher/src/main.rs | Switches processor_main! to sinex_processor_runtime. |
| crate/core/sinex-rpc-dispatcher/src/lib.rs | Migrates processor init to ProcessorInitContext; uses runtime CLI types from new crate. |
| crate/core/sinex-rpc-dispatcher/Cargo.toml | Adds dependency on sinex-processor-runtime. |
| crate/core/sinex-ingestd/tests/validator_schema_order_tests.rs | Adds validator tests ensuring newest semver selection. |
| crate/core/sinex-ingestd/tests/service_outbox_tests.rs | Removes ignored local-JetStream outbox test. |
| crate/core/sinex-ingestd/tests/jetstream_stream_name_tests.rs | Adds test around stream name/lookup conventions. |
| crate/core/sinex-ingestd/tests/jetstream_idempotency_property_test.rs | Converts proptest usage to ignored ingestd pipeline scenarios. |
| crate/core/sinex-ingestd/tests/jetstream_idempotency_property_test.proptest-regressions | Adds proptest regression seeds file (historical). |
| crate/core/sinex-ingestd/tests/jetstream_e2e_integration_test.rs | Updates to new topology/consumer wiring; marks full-pipeline tests ignored. |
| crate/core/sinex-ingestd/tests/figment_config_tests.rs | Migrates to IngestdConfig and adds config file load test. |
| crate/core/sinex-ingestd/tests/config_security_tests.rs | Uses default-config JSON base to reduce brittleness; keeps security assertions. |
| crate/core/sinex-ingestd/src/validator.rs | Adjusts schema ordering and adds validate_payload_for helper. |
| crate/core/sinex-ingestd/src/prelude.rs | Removes IngestdFigmentConfig re-export. |
| crate/core/sinex-ingestd/src/main.rs | Adds CLI args for annex_path and assembler_state_dir. |
| crate/core/sinex-ingestd/src/lib.rs | Removes figment_config module; re-exports JetStreamTopology. |
| crate/core/sinex-ingestd/doc/figment_config.md | Updates docs to point at IngestdConfig::load. |
| crate/core/sinex-ingestd/doc/config.md | Updates docs to reflect single config module approach. |
| crate/core/sinex-ingestd/Cargo.toml | Adds libc and satellite-sdk dep tweaks; adds tempfile for tests. |
| crate/core/sinex-gateway/tests/cascade_analyzer_cycle_test.rs | Adds cycle detection test for deeper cascade graphs. |
| crate/core/sinex-gateway/src/service_container.rs | Uses env-aware default annex path; drains blob events to avoid backpressure. |
| crate/core/sinex-gateway/src/rpc_server.rs | Improves unknown-method handling; sets Unix socket permissions to 0600. |
| crate/core/sinex-gateway/src/native_messaging.rs | Moves blocking stdin/stdout operations onto spawn_blocking. |
| crate/core/sinex-gateway/src/main.rs | Centralizes default socket path constant. |
| crate/core/sinex-gateway/src/handlers.rs | Switches blob/note content to base64; returns blob metadata; adds unit test. |
| crate/core/sinex-gateway/src/cascade_analyzer.rs | Uses ULID session ids; adjusts cycle detection SQL; updates row-size estimate calc. |
| crate/core/sinex-gateway/Cargo.toml | Adds base64 + thiserror deps. |
| cli/completion.py | Updates manifest query schema from sinex_schemas to core. |
| TESTING.md | Documents nats-server dependency and recommends using runtime test builder. |
| Cargo.toml | Adds new crate to workspace; removes tonic/prost workspace deps; adds base64. |
Comments suppressed due to low confidence (1)
crate/core/sinex-ingestd/src/validator.rs:1
- Selecting the 'latest' schema via
updated_at DESC, schema_version DESCis not a reliable semver ordering strategy:schema_versionordering is lexicographic, so versions like1.9.9can incorrectly sort after1.10.0whenupdated_atties (andCURRENT_TIMESTAMPdefaults can be identical within a transaction). Since the PR adds tests asserting newest semver wins, prefer computing 'latest schema' using proper semver comparison (e.g., parse with a semver parser in Rust and choose max per (source,event_type), or implement explicit numeric ordering in SQL via split/CAST).
#![doc = include_str!("../doc/validator.md")]
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+9
to
+11
| js.get_or_create_stream(jetstream::stream::Config { | ||
| name: env.nats_subject("source_material_begin"), | ||
| subjects: vec![env.nats_subject("source_material.begin")], |
There was a problem hiding this comment.
The stream name is being set to a subject (env.nats_subject(\"source_material_begin\")) rather than a stream name. Elsewhere (e.g., AcquisitionManager::ensure_streams_once) stream names are derived via env.nats_stream_name(\"SOURCE_MATERIAL_BEGIN\"), so this helper will create a differently named stream and later get_stream(...) calls by stream name will fail. Use env.nats_stream_name(\"SOURCE_MATERIAL_BEGIN\")/...SLICES/...END for name, and keep subjects as env.nats_subject(...).
Comment on lines
+27
to
+29
| js.get_stream(&env.nats_subject("source_material_begin")) | ||
| .await | ||
| .expect("stream lookup by subject should succeed"); |
There was a problem hiding this comment.
async_nats::jetstream::Context::get_stream expects a stream name, not a subject. This test currently creates a stream named env.nats_stream_name(\"SOURCE_MATERIAL_BEGIN\"), but then attempts to fetch it using env.nats_subject(\"source_material_begin\"), which is a subject-like string. If the intent is to validate lookup by stream name, pass stream_name; if the intent is lookup-by-subject, the test should be rewritten to use the JetStream API that resolves streams by subject (or explicitly search stream infos) rather than get_stream.
Comment on lines
+85
to
+97
| pub async fn update(&mut self, processed: u64) { | ||
| let mut guard = self.state.write().await; | ||
| guard.processed_events += processed; | ||
| guard.batches_processed += 1; | ||
| guard.last_update = Utc::now(); | ||
|
|
||
| let elapsed = guard | ||
| .last_update | ||
| .signed_duration_since(guard.start_time) | ||
| .to_std() | ||
| .unwrap_or_default(); | ||
| guard.current_rate = processed as f64 / elapsed.as_secs_f64().max(1.0); | ||
| guard.average_rate = guard.processed_events as f64 / elapsed.as_secs_f64().max(1.0); |
There was a problem hiding this comment.
current_rate is computed using total elapsed time since start_time, which makes it converge toward average_rate and shrink over time (it is not a 'current' or per-batch rate). If you want an instantaneous/current rate, track the previous last_update (or a separate last_batch_time) and divide processed by the delta since the previous update; keep average_rate as total processed over total elapsed.
Comment on lines
+280
to
293
| let mut temp_file = tempfile::Builder::new() | ||
| .prefix("sinex_blob_") | ||
| .suffix(".tmp") | ||
| .tempfile_in(std::env::temp_dir()) | ||
| .wrap_err("Failed to create secure temporary file")?; | ||
|
|
||
| use std::io::Write; | ||
| temp_file | ||
| .write_all(content) | ||
| .wrap_err("Failed to write blob bytes to temporary file")?; | ||
| temp_file | ||
| .flush() | ||
| .wrap_err("Failed to flush temporary blob file")?; | ||
|
|
There was a problem hiding this comment.
This is inside an async function, but it performs blocking I/O (std::io::Write::write_all + flush) on the Tokio runtime thread. That can stall unrelated async tasks under load. Prefer either using tokio::fs::File/tokio::io::AsyncWriteExt for writes, or wrapping the blocking write/flush in tokio::task::spawn_blocking before proceeding with async git-annex operations.
Suggested change
| let mut temp_file = tempfile::Builder::new() | |
| .prefix("sinex_blob_") | |
| .suffix(".tmp") | |
| .tempfile_in(std::env::temp_dir()) | |
| .wrap_err("Failed to create secure temporary file")?; | |
| use std::io::Write; | |
| temp_file | |
| .write_all(content) | |
| .wrap_err("Failed to write blob bytes to temporary file")?; | |
| temp_file | |
| .flush() | |
| .wrap_err("Failed to flush temporary blob file")?; | |
| let content = content.to_vec(); | |
| let temp_file = tokio::task::spawn_blocking(move || -> Result<tempfile::NamedTempFile> { | |
| use std::io::Write; | |
| let mut temp_file = tempfile::Builder::new() | |
| .prefix("sinex_blob_") | |
| .suffix(".tmp") | |
| .tempfile_in(std::env::temp_dir()) | |
| .wrap_err("Failed to create secure temporary file")?; | |
| temp_file | |
| .write_all(&content) | |
| .wrap_err("Failed to write blob bytes to temporary file")?; | |
| temp_file | |
| .flush() | |
| .wrap_err("Failed to flush temporary blob file")?; | |
| Ok(temp_file) | |
| }) | |
| .await | |
| .wrap_err("Failed to join blocking task for temporary blob file creation")??; |
Comment on lines
+62
to
+64
| while let Some(event) = event_receiver.recv().await { | ||
| // In lieu of a dedicated processing pipeline, drain events to avoid backpressure. | ||
| debug!(?event, "Blob manager emitted event"); |
There was a problem hiding this comment.
Logging full event payloads at debug can leak sensitive content/metadata and may be very large (especially for blob-related events), creating both data exposure risk and log-volume operational issues. Consider logging only stable identifiers (event id, type, blob/annex key) and omit/redact payload fields, or gate verbose logging behind a dedicated feature/flag.
Suggested change
| while let Some(event) = event_receiver.recv().await { | |
| // In lieu of a dedicated processing pipeline, drain events to avoid backpressure. | |
| debug!(?event, "Blob manager emitted event"); | |
| while let Some(_event) = event_receiver.recv().await { | |
| // In lieu of a dedicated processing pipeline, drain events to avoid backpressure. | |
| // Do not log the full event payload here: events may contain sensitive metadata | |
| // and can be large, especially for blob-related operations. | |
| debug!("Blob manager emitted event"); |
Comment on lines
+366
to
368
| "UPDATE core.processor_manifests | ||
| SET last_heartbeat_ts = $1, updated_at = $2 | ||
| WHERE processor_name = $3 AND processor_type = 'automaton", |
There was a problem hiding this comment.
These SQL strings appear to be missing a closing quote after 'automaton' (and potentially the closing quote of the Rust string literal depending on the unseen continuation). If compiled as shown, the SQL sent to Postgres will be invalid and tests will fail at runtime. Update the SQL to correctly quote the literal (processor_type = 'automaton') and ensure the Rust string literal is properly terminated.
| // Deregister agent | ||
| match sqlx::query!( | ||
| "DELETE FROM sinex_schemas.processor_manifests WHERE processor_name = $1 AND processor_type = 'automaton", | ||
| "DELETE FROM core.processor_manifests WHERE processor_name = $1 AND processor_type = 'automaton", |
There was a problem hiding this comment.
These SQL strings appear to be missing a closing quote after 'automaton' (and potentially the closing quote of the Rust string literal depending on the unseen continuation). If compiled as shown, the SQL sent to Postgres will be invalid and tests will fail at runtime. Update the SQL to correctly quote the literal (processor_type = 'automaton') and ensure the Rust string literal is properly terminated.
| // Verify final database state | ||
| let remaining_agents = sqlx::query!( | ||
| "SELECT COUNT(*) as count FROM sinex_schemas.processor_manifests WHERE processor_name LIKE $1 AND processor_type = 'automaton", | ||
| "SELECT COUNT(*) as count FROM core.processor_manifests WHERE processor_name LIKE $1 AND processor_type = 'automaton", |
There was a problem hiding this comment.
These SQL strings appear to be missing a closing quote after 'automaton' (and potentially the closing quote of the Rust string literal depending on the unseen continuation). If compiled as shown, the SQL sent to Postgres will be invalid and tests will fail at runtime. Update the SQL to correctly quote the literal (processor_type = 'automaton') and ensure the Rust string literal is properly terminated.
Comment on lines
+109
to
+114
| let earlier_id = earlier.id.as_ref().expect("ingestion id"); | ||
| let later_id = later.id.as_ref().expect("ingestion id"); | ||
| let earlier_str = earlier_id.to_string(); | ||
| let later_str = later_id.to_string(); | ||
| assert!( | ||
| earlier.id.unwrap() < later.id.unwrap(), | ||
| earlier_str < later_str, |
There was a problem hiding this comment.
This converts ULIDs to strings just to compare ordering, which adds allocations and obscures intent. Since ULIDs are orderable, compare the ULID values directly (e.g., compare earlier_id and later_id by value/clone) to preserve semantics and avoid string conversions.
Sinity
added
the
faux-pull-request
4 tasks
Sinity
added a commit
that referenced
this pull request
Jul 13, 2026
#94) Finish the sensd-era cleanup by deleting the last transport helpers, move the shared processor runtime into its own crate, and rebase ingestd, node crates, and NixOS wiring on the JetStream-only runtime model. transport and runtime extraction: - delete `crate/lib/sinex-satellite-sdk/src/{sensd_client.rs,sensor_guard.rs}` and remove the last gRPC/protobuf workspace deps from `Cargo.toml`, which strips the SDK down to the JetStream-era transport model - add `crate/lib/sinex-processor-runtime/{Cargo.toml,src/cli.rs,src/runner.rs,src/lib.rs}` and move the shared CLI/runner plumbing out of `sinex-satellite-sdk`, so runtime concerns stop living in the same crate as sensor helpers and annex utilities - relocate `stream_processor.rs` under `crate/lib/sinex-satellite-sdk/src/runtime/stream/`, add `ProcessorHandles`, `ProcessorRuntimeState`, and `runtime_state.rs`, and then delete `StreamProcessorContext`, making runtime-state-only initialization the single processor path node and SDK adoption: - port the terminal, filesystem, desktop, system, document, analytics, search, content, PKM, and health crates to `ProcessorRuntimeState` while keeping short-lived compatibility reexports during the transition - make `ProcessorInitContext` the only `StatefulStreamProcessor` initialization API and add convenience builders for config lookup, emission, workdir access, lifecycle management, coordination, replay, stage-as-you-go, heartbeat, and acquisition - add `StageAsYouGoContext::from_handles` and `AcquisitionManager::from_handles`, so node crates can build staging and capture helpers directly from runtime handles without the deleted sensd abstractions ingestd, gateway, and core hardening: - rework `crate/core/sinex-ingestd/src/{service.rs,material_assembler.rs,jetstream_consumer.rs,validator.rs}` around JetStream-only subjects, checkpoint flow, blob handling, and stricter validation, and switch ingestd to `events.raw` subjects with hardened consumer tests - tighten `crate/core/sinex-gateway/src/{handlers.rs,rpc_server.rs,service_container.rs}` and route document retrieval through `BlobManager`, so gateway and document ingest stop depending on stale transport-era assumptions - restore system payload builders in `crate/lib/sinex-core/src/types/events/payloads/system.rs`, remove the legacy inline terminal canonicalizer path, and harden repository/provenance/cascade tests in `crate/lib/sinex-core/tests/*` nixos and secrets: - flatten `nixos/modules/default.nix`, inline service module composition, and rename lifecycle options such as `services.sinex.preflightVerification.enable` to `services.sinex.lifecycle.preflight.enable` - add `nixos/modules/secrets.nix`, inject the `agenix` flake input, and scaffold the encrypted `secret/` repository, so the default module can express secret-backed deployments without bespoke local wiring
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
I finished the sensd-era cleanup by deleting the last transport-only helpers, extracted the shared processor runtime into a dedicated
sinex-processor-runtimecrate, and migrated ingestd plus the node crates onto a runtime-state-only model. The branch also tightens ingestd around JetStream-only subjects and blob handling, updates gateway/document retrieval paths that were still carrying old assumptions, and flattens the NixOS module story enough to add an agenix-backed secrets module cleanly. The through-line is not "move files around." It is "make the JetStream-only architecture visible in the runtime model, not just in individual transport helpers." I kept compatibility reexports and bridge phases just long enough to move the crates, then deletedStreamProcessorContextand the last sensd plumbing once the nodes were onProcessorRuntimeState. The final tail of the branch hardens repositories, provenance-sensitive tests, and cascade behavior so the refactor does not leave the data model wobblier than before.Motivation
After the earlier transport removal work, two large pieces of debt remained.
The first was architectural:
sinex-satellite-sdkstill mixed together transport/runtime machinery, sensor helpers, annex behavior, and initialization types that had clearly become runtime concerns.StreamProcessorContextwas especially revealing here. Even though the transport model had already moved to JetStream, processor initialization still looked like a legacy context object assembled from older assumptions.The second was operational: the codebase was still carrying sensd-era convenience layers and NixOS/module composition that made the new runtime harder to reason about. There were also a few JetStream integration gaps — subject naming, blob flow, checkpoint/state behavior, and validation rough edges — that were easier to fix once the runtime surface had a cleaner center of gravity.
I deliberately staged the runtime work. The branch first removes the last obvious sensd helpers, then relocates runtime files under
runtime/with compatibility reexports, then introducesProcessorHandlesandProcessorRuntimeState, then ports the node crates, and only after that deletesStreamProcessorContext. That sequence matters: it let me move a large number of crates without forcing a single giant flag day.Changes grouped by concern
1. Extract the runtime so it can be reasoned about as a runtime
The core structural change is the new crate:
I moved the shared CLI/runner surfaces there and stopped pretending they were just another piece of the SDK. The extraction is paired with a second move inside
sinex-satellite-sdk: the stream runtime files now live undersrc/runtime/stream/instead of a single sprawlingstream_processor.rs.That layout change becomes meaningful once
ProcessorHandlesandProcessorRuntimeStateland. They give processors a stable way to reach:The important part is that these are runtime capabilities, not transport-era conveniences.
2. Delete the last sensd-shaped helpers and bind staging/capture to runtime handles
Two deletions define the "remove sensd plumbing" part of the branch:
crate/lib/sinex-satellite-sdk/src/sensd_client.rscrate/lib/sinex-satellite-sdk/src/sensor_guard.rsThose files kept a sensd-shaped mental model alive even after the transport direction had changed. Their removal matters more once the runtime surfaces are in place, because the replacement is not "nothing"; it is direct runtime-backed construction:
That makes stage-as-you-go and acquisition helpers runtime-derived instead of being routed through deleted abstractions.
3. Port the node crates in stages, then delete the old init path
The node migration happens in two phases.
First, I port the node and automaton crates to
ProcessorRuntimeStatewhile keeping bridge compatibility. That touches the terminal, filesystem, desktop, system, document, analytics, content, PKM, search, and health-related crates. The point of this phase is to get the call sites over without immediately deleting every compatibility layer.Second, I make
ProcessorInitContextthe only initialization API forStatefulStreamProcessor, then deleteStreamProcessorContextafter the ports are done. That is the moment the runtime refactor becomes real rather than incremental.The runtime state picks up additional convenience accessors as part of that process: config access, emitters, workdir access,
JobManager,LifecycleManager, coordination, heartbeat, replay, stage-as-you-go, and acquisition builders.4. Tighten ingestd and gateway around the JetStream-only world
The runtime extraction would have felt incomplete if ingestd still had rough edges around the bus and blob path, so I used the branch to rework the ingest side too.
The most important ingestd changes are in:
service.rsmaterial_assembler.rsjetstream_consumer.rsvalidator.rsThe branch aligns checkpoint flow with the JetStream-only model, improves material assembly and blob handling, switches ingestd over to
events.rawsubjects, and hardens the consumer tests that prove that behavior. On the gateway side, I tightened handler/service-container behavior and pushed document retrieval throughBlobManager, which removes another stale assumption about how content should be fetched.I also removed the legacy inline canonicalizer path so the unified processor route is the only terminal canonicalization path left.
5. Flatten NixOS module composition and add secret-backed wiring
The NixOS side is a real part of the landing, not a side quest. I flattened the module stack enough that the default module can own more of the composition directly, and I renamed the lifecycle options to a nested form:
That cleanup made it much easier to add
nixos/modules/secrets.nixand wireagenixin cleanly. The branch adds the flake input, the encryptedsecret/scaffold, and the default-module hooks needed to make secret-backed deployments part of the supported configuration story instead of a local afterthought.6. Harden provenance-sensitive repository behavior on the way out
The last part of the branch is not cosmetic cleanup. I used it to harden repository and test behavior around:
That is especially visible in
crate/lib/sinex-core/tests/*and the repository implementations they exercise. I wanted the runtime/transport refactor to land with sharper invariants, not just a new file layout.Testing
The branch leans on a broad but targeted set of tests:
crate/lib/sinex-core/tests/*crate/lib/sinex-satellite-sdk/tests/support/runtime.rstests/e2e/tests/pipeline_end_to_end.rsNotes
I kept compatibility reexports and bridge phases on purpose while the node crates were being ported. That is why some runtime files appear to move twice in the branch log: the first move makes the extraction possible, the second deletes the old compatibility layer once the ports are complete. I also consider the agenix work part of the same architectural cleanup because the flattened module stack and new runtime model would still have left deployment configuration feeling unfinished without a real secret path. Finally,
events.rawsubject alignment is important enough to call out explicitly — that subject switch is part of what makes the JetStream-only runtime model feel settled rather than halfway migrated.Ref #93