diff --git a/src/spark_intelligence/cli.py b/src/spark_intelligence/cli.py index 2bdaab31..ee94f665 100644 --- a/src/spark_intelligence/cli.py +++ b/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": diff --git a/tests/test_gateway_tool_ledger_cli.py b/tests/test_gateway_tool_ledger_cli.py new file mode 100644 index 00000000..ce43d3be --- /dev/null +++ b/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)