fix(gateway): expose canonical tool-ledger ingest CLI

src/spark_intelligence/cli.py

@@ -68,6 +68,7 @@
     gateway_status,
     gateway_trace_view,
 )
+from spark_intelligence.gateway.tool_ledger import ingest_tool_ledger_payload
 from spark_intelligence.gateway.tracing import read_gateway_traces, redact_gateway_trace_log, repair_gateway_trace_proof_continuity
 from spark_intelligence.gateway.oauth_callback import pending_oauth_redirect_uri, serve_gateway_oauth_callback
 from spark_intelligence.harness_contract import build_vnext_action_intent_envelope
@@ -2206,6 +2207,16 @@ def build_parser() -> argparse.ArgumentParser:
         default="simulation",
         help="Fix the trace origin for the lifetime of this parent-owned session",
     )
+    gateway_ingest_tool_ledger_parser = gateway_subparsers.add_parser(
+        "ingest-tool-ledger",
+        help="Persist one schema-valid bound tool-ledger row from JSON",
+    )
+    gateway_ingest_tool_ledger_parser.add_argument(
+        "payload_file",
+        help="JSON payload path, or - to read one payload from stdin",
+    )
+    gateway_ingest_tool_ledger_parser.add_argument("--home", help="Override Spark Intelligence home directory")
+    gateway_ingest_tool_ledger_parser.add_argument("--json", action="store_true", help="Emit machine-readable output")
     gateway_ask_telegram_parser = gateway_subparsers.add_parser(
         "ask-telegram",
         help="Send one synthetic DM through the Telegram runtime path and print Spark's reply",
@@ -5680,6 +5691,38 @@ def handle_gateway_serve_stdio(args: argparse.Namespace) -> int:
         return 2
 
 
+def handle_gateway_ingest_tool_ledger(args: argparse.Namespace) -> int:
+    max_chars = 1_048_576
+    try:
+        if args.payload_file == "-":
+            payload_text = sys.stdin.read(max_chars + 1)
+        else:
+            payload_path = Path(args.payload_file)
+            if payload_path.stat().st_size > max_chars:
+                raise ValueError("tool ledger ingest payload exceeds 1 MiB")
+            payload_text = payload_path.read_text(encoding="utf-8-sig")
+        if len(payload_text) > max_chars:
+            raise ValueError("tool ledger ingest payload exceeds 1 MiB")
+        payload = json.loads(payload_text)
+        if not isinstance(payload, dict):
+            raise ValueError("tool ledger ingest payload must be a JSON object")
+    except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as exc:
+        print(f"Tool ledger ingest input rejected: {exc}", file=sys.stderr)
+        return 2
+
+    config_manager = ConfigManager.from_home(args.home)
+    state_db = StateDB(config_manager.paths.state_db)
+    config_manager.bootstrap()
+    state_db.initialize()
+    try:
+        result = ingest_tool_ledger_payload(state_db, payload)
+    except ValueError as exc:
+        print(f"Tool ledger ingest rejected: {exc}", file=sys.stderr)
+        return 2
+    print(result.to_json() if args.json else result.to_text())
+    return 0
+
+
 def handle_gateway_ask_telegram(args: argparse.Namespace) -> int:
     config_manager = ConfigManager.from_home(args.home)
     state_db = StateDB(config_manager.paths.state_db)
@@ -9866,6 +9909,8 @@ def main(argv: list[str] | None = None) -> int:
         return handle_gateway_simulate_telegram_update(args)
     if args.command == "gateway" and args.gateway_command == "serve-stdio":
         return handle_gateway_serve_stdio(args)
+    if args.command == "gateway" and args.gateway_command == "ingest-tool-ledger":
+        return handle_gateway_ingest_tool_ledger(args)
     if args.command == "gateway" and args.gateway_command == "ask-telegram":
         return handle_gateway_ask_telegram(args)
     if args.command == "gateway" and args.gateway_command == "shadow-telegram":

tests/test_gateway_tool_ledger_cli.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+import io
+import json
+from unittest.mock import patch
+
+from spark_intelligence.bridge_authority import authorize_builder_bridge_action
+from spark_intelligence.harness_contract import build_vnext_tool_intent_envelope
+
+from tests.test_support import SparkTestCase
+
+
+class GatewayToolLedgerCliTests(SparkTestCase):
+    def _bound_row(self) -> dict[str, object]:
+        envelope = build_vnext_tool_intent_envelope(
+            surface="spawner",
+            actor_id_ref="human:r30-test",
+            request_id="request-r30-spawner-ledger-cli",
+            source_kind="spawner_canonical_ledger_ingest_test",
+            tool_name="spawner.dispatch",
+            owner_system="spawner-ui",
+            mutation_class="launches_mission",
+            intent_summary="User explicitly requested a governed Spawner dispatch.",
+            raw_turn_summary="Raw test turn remains offloaded.",
+            confidence=1.0,
+        )
+        self.assertIsNotNone(envelope)
+        verdict = authorize_builder_bridge_action(
+            {"turn_intent_envelope_vnext": envelope},
+            tool_name="spawner.dispatch",
+            owner_system="spawner-ui",
+            mutation_class="launches_mission",
+            state_db=self.state_db,
+            request_id="request-r30-spawner-ledger-cli",
+            actor_id="spawner-ui",
+            component="spawner_canonical_ledger_ingest_test",
+        )
+        self.assertTrue(verdict.allowed, verdict.reason_codes)
+        self.assertIsNotNone(verdict.tool_call_ledger)
+        ledger = verdict.tool_call_ledger
+        assert ledger is not None
+        authorization = ledger["authorization"]
+        result = ledger["result"]
+        return {
+            "ledger_id": ledger["ledger_id"],
+            "turn_id": ledger["turn_id"],
+            "action_id": ledger["action_id"],
+            "capability_id": ledger["capability_id"],
+            "authorization_decision_id": authorization["decision_id"],
+            "tool_name": ledger["tool_name"],
+            "owner_system": "spawner-ui",
+            "mutation_class": "launches_mission",
+            "surface": "spawner",
+            "request_id": "request-r30-spawner-ledger-cli",
+            "trace_ref": "trace:r30-spawner-ledger-cli",
+            "status": result["status"],
+            "ledger_json": ledger,
+        }
+
+    def test_ingest_tool_ledger_reads_stdin_and_persists_validated_row(self) -> None:
+        row = self._bound_row()
+
+        with patch("sys.stdin", io.StringIO(json.dumps({"row": row}))):
+            exit_code, stdout, stderr = self.run_cli(
+                "gateway",
+                "ingest-tool-ledger",
+                "-",
+                "--home",
+                str(self.home),
+                "--json",
+            )
+
+        self.assertEqual(exit_code, 0, stderr)
+        self.assertEqual(stderr, "")
+        result = self.read_json(stdout)
+        self.assertEqual(result["ledger_id"], row["ledger_id"])
+        self.assertEqual(result["turn_id"], row["turn_id"])
+        self.assertEqual(result["surface"], "spawner")
+
+    def test_ingest_tool_ledger_rejects_malformed_stdin_without_traceback(self) -> None:
+        with patch("sys.stdin", io.StringIO("{not-json")):
+            exit_code, stdout, stderr = self.run_cli(
+                "gateway",
+                "ingest-tool-ledger",
+                "-",
+                "--home",
+                str(self.home),
+                "--json",
+            )
+
+        self.assertEqual(exit_code, 2)
+        self.assertEqual(stdout, "")
+        self.assertIn("Tool ledger ingest input rejected:", stderr)
+        self.assertNotIn("Traceback", stderr)