refactor(runtime): extract processor runtime and remove sensd plumbing

Cargo.toml

@@ -5,6 +5,7 @@ members = [
   "crate/lib/sinex-schema",
   "crate/lib/sinex-macros",
   "crate/lib/sinex-satellite-sdk",
+  "crate/lib/sinex-processor-runtime",
   "crate/lib/sinex-services",
   # Re-enabled to ensure tests compile
   "crate/lib/sinex-test-utils",
@@ -134,11 +135,6 @@ indexmap = "2.0"
 # Retry logic
 tokio-retry = "0.3"
 
-# gRPC and protocol buffers
-tonic = "0.11"
-prost = "0.12"
-tonic-build = "0.11"
-
 # Message bus
 async-nats = "0.33"
 
@@ -150,6 +146,7 @@ tokio-stream = "0.1"
 futures = "0.3"
 dirs = "5.0"
 lazy_static = "1.5"
+base64 = "0.22"
 
 # File system watching
 notify = "6.1"

TESTING.md

@@ -22,8 +22,30 @@ just pre-commit
 ```
 
 **Prerequisites:** run `nix develop` (or source your local toolchain) so that
-TimescaleDB, NATS, and the Cargo toolchain are available. Test helpers rely on
-environment variables such as `DATABASE_URL` that the dev shell provides.
+TimescaleDB/PostgreSQL, the `nats-server` binary, and the Cargo toolchain are
+available. `TestContext` connects to the local Postgres socket at
+`/run/postgresql` and spins up an ephemeral JetStream instance by shelling out
+to `nats-server`, so both services must be reachable before running `just test`.
+
+If you are inside the Nix dev shell (`nix develop`), `nats-server` is already
+on `PATH` (see `flake.nix` exporting `NATS_SERVER_BIN`). For other toolchains,
+install NATS manually:
+
+```bash
+# macOS (Homebrew)
+brew install nats-server
+
+# Linux (Debian/Ubuntu)
+sudo apt-get install -y nats-server
+
+# Or download a release binary
+curl -L https://github.com/nats-io/nats-server/releases/latest/download/nats-server-amd64.zip -o /tmp/nats.zip
+unzip /tmp/nats.zip -d /tmp/nats
+sudo mv /tmp/nats/nats-server /usr/local/bin/
+```
+
+You can override the binary path with `NATS_SERVER_BIN=/custom/path/nats-server`
+if you prefer to keep it outside `$PATH`.
 
 ## Test Layout at a Glance
 
@@ -51,7 +73,22 @@ tests are reserved for scenarios that truly span multiple crates or binaries.
 - **Async hygiene:** use bounded concurrency (`buffer_unordered`, semaphores),
   propagate errors rather than ignoring `JoinHandle`s, and avoid `std::thread::sleep`.
 - **Fixtures:** prefer the fixture namespaces under `sinex_test_utils::fixtures`
-  rather than re-creating bespoke data builders.
+  rather than re-creating bespoke data builders. When you need a satellite
+  runtime for integration coverage, include the shared builder from
+  `sinex-satellite-sdk/tests/support/runtime.rs` instead of wiring pools and
+  emitters manually:
+
+  ```rust
+  #[path = "support/runtime.rs"]
+  mod runtime;
+
+  use runtime::TestRuntimeBuilder;
+
+  let test_runtime = TestRuntimeBuilder::new(&ctx, "my-service")
+      .with_dry_run(true)
+      .build()
+      .await?;
+  ```
 - **Property tests:** place proptest suites alongside the crate they fuzz (see
   below). Capture any new failing seeds in that crate’s
   `tests/property/*.proptest-regressions` files.

cli/completion.py

@@ -68,7 +68,7 @@ def get_automata() -> List[str]:
     try:
         with get_db_connection() as conn:
             with conn.cursor() as cur:
-                cur.execute("SELECT DISTINCT processor_name FROM sinex_schemas.processor_manifests WHERE processor_type = 'automaton' ORDER BY processor_name")
+                cur.execute("SELECT DISTINCT processor_name FROM core.processor_manifests WHERE processor_type = 'automaton' ORDER BY processor_name")
                 return [row['processor_name'] for row in cur.fetchall()]
     except Exception:
         return []
@@ -1157,4 +1157,4 @@ def install_completion(shell: str, completion_dir: Optional[str] = None) -> bool
         print(' '.join(get_satellites()))
     else:
         print(f"Unknown command: {shell}")
-        sys.exit(1)
+        sys.exit(1)

cli/exo.py

@@ -14,7 +14,7 @@
 
 import click
 import psycopg2
-from psycopg2.extras import RealDictCursor
+from psycopg2.extras import RealDictCursor, Json
 from rich.console import Console
 from rich.table import Table
 from rich.json import JSON
@@ -581,7 +581,7 @@ def processor_list(type: Optional[str], status: Optional[str]):
             query_parts = [
                 "SELECT processor_name, processor_type, description, version, status, "
                 "produces_event_types, last_heartbeat_ts, registered_at "
-                "FROM sinex_schemas.processor_manifests"
+                "FROM core.processor_manifests"
             ]
             params = []
             where_conditions = []
@@ -654,7 +654,7 @@ def processor_status(processor_name: str):
         with conn.cursor() as cur:
             # Get processor manifest
             cur.execute("""
-                SELECT * FROM sinex_schemas.processor_manifests
+                SELECT * FROM core.processor_manifests
                 WHERE processor_name = %s
             """, (processor_name,))
             processor = cur.fetchone()
@@ -667,10 +667,10 @@ def processor_status(processor_name: str):
             checkpoints = []
             if processor['processor_type'] == 'automaton':
                 cur.execute("""
-                    SELECT automaton_name, last_processed_id, processed_count, 
-                           last_activity, state_data
-                    FROM core.automaton_checkpoints
-                    WHERE automaton_name = %s
+                    SELECT processor_name, consumer_group, consumer_name,
+                           last_processed_id, processed_count, last_activity, checkpoint_data
+                    FROM core.processor_checkpoints
+                    WHERE processor_name = %s
                     ORDER BY last_activity DESC
                 """, (processor_name,))
                 checkpoints = cur.fetchall()
@@ -692,11 +692,14 @@ def processor_status(processor_name: str):
             if checkpoints:
                 console.print("\n[bold]Checkpoints:[/bold]")
                 for checkpoint in checkpoints:
-                    console.print(f"  Last processed: {checkpoint['last_processed_id'] or 'None'}")
-                    console.print(f"  Processed count: {checkpoint['processed_count']}")
-                    console.print(f"  Last activity: {checkpoint['last_activity']}")
-                    if checkpoint['state_data']:
-                        console.print(f"  State: {checkpoint['state_data']}")
+                    console.print(
+                        f"  Consumer: {checkpoint['consumer_group']}/{checkpoint['consumer_name']}"
+                    )
+                    console.print(f"    Last processed: {checkpoint['last_processed_id'] or 'None'}")
+                    console.print(f"    Processed count: {checkpoint['processed_count']}")
+                    console.print(f"    Last activity: {checkpoint['last_activity']}")
+                    if checkpoint['checkpoint_data']:
+                        console.print(f"    State: {checkpoint['checkpoint_data']}")
 
             console.print(f"\nRegistered: {processor['registered_at']}")
             console.print(f"Last seen: {processor['last_seen']}")
@@ -713,7 +716,7 @@ def automaton_list(status: Optional[str]):
             query_parts = [
                 "SELECT processor_name, description, version, status, "
                 "produces_event_types, last_heartbeat_ts, registered_at "
-                "FROM sinex_schemas.processor_manifests "
+                "FROM core.processor_manifests "
                 "WHERE processor_type = 'automaton'"
             ]
             params = []
@@ -774,7 +777,7 @@ def automaton_status(automaton_name: str):
         with conn.cursor() as cur:
             # Get automaton manifest
             cur.execute("""
-                SELECT * FROM sinex_schemas.processor_manifests
+                SELECT * FROM core.processor_manifests
                 WHERE processor_name = %s AND processor_type = 'automaton'
             """, (automaton_name,))
             automaton = cur.fetchone()
@@ -3141,16 +3144,18 @@ def explore_source_state(ctx, satellite: str, verbose: bool):
 
         with get_db_connection() as conn:
             with conn.cursor() as cur:
-                # Get automaton checkpoint for this satellite
+                # Get checkpoints for this satellite's processor(s)
                 cur.execute("""
                     SELECT 
-                        automaton_name,
+                        processor_name,
+                        consumer_group,
+                        consumer_name,
                         last_processed_id,
                         processed_count,
                         last_activity,
-                        state_data
-                    FROM core.automaton_checkpoints
-                    WHERE automaton_name LIKE %s
+                        checkpoint_data
+                    FROM core.processor_checkpoints
+                    WHERE processor_name LIKE %s
                     ORDER BY last_activity DESC
                 """, [f"%{satellite}%"])
 
@@ -3163,13 +3168,16 @@ def explore_source_state(ctx, satellite: str, verbose: bool):
                 console.print(f"\n[bold]Checkpoint Status:[/bold]")
 
                 for checkpoint in checkpoints:
-                    console.print(f"  [cyan]{checkpoint['automaton_name']}[/cyan]")
+                    console.print(
+                        f"  [cyan]{checkpoint['processor_name']}[/cyan] :: "
+                        f"{checkpoint['consumer_group']}/{checkpoint['consumer_name']}"
+                    )
                     console.print(f"    Last processed: {checkpoint['last_processed_id'] or 'None'}")
                     console.print(f"    Processed count: {checkpoint['processed_count']}")
                     console.print(f"    Last activity: {checkpoint['last_activity']}")
-                    
-                    if verbose and checkpoint['state_data']:
-                        console.print(f"    State data: {JSON(checkpoint['state_data'])}")
+
+                    if verbose and checkpoint['checkpoint_data']:
+                        console.print(f"    State: {JSON(checkpoint['checkpoint_data'])}")
 
                     console.print()
 
@@ -3606,44 +3614,44 @@ def find_satellite_binary(name):
             sys.exit(1)
         satellites_to_scan = [satellite]
 
-    # Log the operation (disabled for now - satellites don't have unified CLI yet)
     operation_id = None
-    console.print(f"[blue]Note: Operations logging temporarily disabled pending satellite CLI unification[/blue]")
-
-    # TODO: Re-enable when satellites implement StatefulStreamProcessor with scan subcommand
-    # try:
-    #     with get_db_connection() as conn:
-    #         with conn.cursor() as cur:
-    #             # Use ULID() PostgreSQL function to generate a proper ULID
-    #             cur.execute("""
-    #                 INSERT INTO core.operations_log (
-    #                     operation_id, operation_type, status, invoked_by_user, parameters
-    #                 ) VALUES (ulid(), %s, %s, %s, %s)
-    #                 RETURNING operation_id
-    #             """, (
-    #                 'scan',
-    #                 'started',
-    #                 os.getenv('USER', 'unknown'),
-    #                 json.dumps({
-    #                     'satellites': satellites_to_scan,
-    #                     'since': since,
-    #                     'until': until,
-    #                     'targets': list(targets),
-    #                     'dry_run': dry_run,
-    #                     'estimate': estimate,
-    #                     'interactive': interactive,
-    #                     'max_events': max_events,
-    #                     'timeout': timeout,
-    #                     'parallel': parallel
-    #                 })
-    #             ))
-    #             
-    #             # Get the generated operation_id
-    #             row = cur.fetchone()
-    #             operation_id = str(row[0]) if row else None
-    #             conn.commit()
-    # except Exception as e:
-    #     console.print(f"[yellow]Warning: Could not log operation: {e}[/yellow]")
+    try:
+        with get_db_connection() as conn:
+            with conn.cursor() as cur:
+                scope = {
+                    'satellites': satellites_to_scan,
+                    'since': since,
+                    'until': until,
+                    'targets': list(targets),
+                    'dry_run': dry_run,
+                    'estimate': estimate,
+                    'interactive': interactive,
+                    'max_events': max_events,
+                    'timeout': timeout,
+                    'parallel': parallel,
+                }
+                cur.execute(
+                    """
+                    INSERT INTO core.operations_log (
+                        operation_type,
+                        operator,
+                        scope,
+                        result_status
+                    ) VALUES (%s, %s, %s::jsonb, %s)
+                    RETURNING id
+                    """,
+                    (
+                        'scan',
+                        os.getenv('USER', 'unknown'),
+                        Json(scope),
+                        'running',
+                    ),
+                )
+                row = cur.fetchone()
+                operation_id = str(row['id']) if row and 'id' in row else None
+                conn.commit()
+    except Exception as e:
+        console.print(f"[yellow]Warning: Could not log scan operation: {e}[/yellow]")
 
     start_time = time.time()
 
@@ -3783,31 +3791,35 @@ def run_satellite_scan(satellite_name: str) -> Dict[str, Any]:
                 console.print("[dim]Output:[/dim]")
                 console.print(result['stdout'])
 
-    # Update operation log (disabled for now)
-    # TODO: Re-enable when satellites implement unified CLI
-    # if operation_id:
-    #     try:
-    #         with get_db_connection() as conn:
-    #             with conn.cursor() as cur:
-    #                 cur.execute("""
-    #                     UPDATE core.operations_log 
-    #                     SET status = %s, completed_at = NOW(), 
-    #                         duration_ms = %s, summary = %s
-    #                     WHERE operation_id = %s
-    #                 """, (
-    #                     'completed' if failed_scans == 0 else 'failed',
-    #                     int(total_duration * 1000),
-    #                     json.dumps({
-    #                         'satellites_scanned': len(satellites_to_scan),
-    #                         'successful_scans': successful_scans,
-    #                         'failed_scans': failed_scans,
-    #                         'results': results
-    #                     }),
-    #                     operation_id
-    #                 ))
-    #                 conn.commit()
-    #     except Exception as e:
-    #         console.print(f"[yellow]Warning: Could not update operation log: {e}[/yellow]")
+    if operation_id:
+        try:
+            with get_db_connection() as conn:
+                with conn.cursor() as cur:
+                    cur.execute(
+                        """
+                        UPDATE core.operations_log 
+                        SET result_status = %s,
+                            result_message = %s,
+                            preview_summary = %s::jsonb,
+                            duration_ms = %s
+                        WHERE id = %s::uuid::ulid
+                        """,
+                        (
+                            'success' if failed_scans == 0 else 'failure',
+                            None,
+                            Json({
+                                'satellites_scanned': len(satellites_to_scan),
+                                'successful_scans': successful_scans,
+                                'failed_scans': failed_scans,
+                                'results': results,
+                            }),
+                            int(total_duration * 1000),
+                            operation_id,
+                        ),
+                    )
+                    conn.commit()
+        except Exception as e:
+            console.print(f"[yellow]Warning: Could not update operation log: {e}[/yellow]")
 
     # Exit with error if any scans failed
     if failed_scans > 0:

crate/core/sinex-gateway/Cargo.toml

@@ -33,6 +33,7 @@ tokio-stream = { workspace = true, features = ["net"] }
 # JSON-RPC
 serde = { workspace = true }
 serde_json = { workspace = true }
+base64 = { workspace = true }
 
 # Error handling
 color-eyre = { workspace = true }
@@ -43,6 +44,7 @@ camino = { workspace = true }
 # Logging
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
+thiserror = { workspace = true }
 
 # CLI
 clap = { workspace = true, features = ["derive"] }