fix(plugin): align tool-call payload contract

Author: Q00

src/ouroboros/plugin/firewall.py

@@ -41,6 +41,7 @@
 import re
 import shlex
 import subprocess
+import time
 from typing import Literal
 
 from ouroboros.plugin.digest import (
@@ -1107,11 +1108,14 @@ def _run_failed_invocation_observability_hooks() -> None:
     ).hexdigest()
     redacted_tool_argv, _ = _redact_argv([command_name] + list(argv))
     tool_args_preview = shlex.join(redacted_tool_argv)
-    tool_args_digest = hashlib.sha256(
-        json.dumps([command_name] + list(argv), separators=(",", ":")).encode(
-            "utf-8", errors="surrogateescape"
-        )
-    ).hexdigest()
+    tool_args_digest = (
+        "sha256:"
+        + hashlib.sha256(
+            json.dumps([command_name] + list(argv), separators=(",", ":")).encode(
+                "utf-8", errors="surrogateescape"
+            )
+        ).hexdigest()
+    )
     tool_name = f"{namespace}.{command_name}" if namespace else command_name
     command_permissions = tuple(getattr(command, "permissions", ()) or ())
     tool_permissions = command_permissions or tuple(_required_permissions(manifest))
@@ -1173,6 +1177,37 @@ def _run_failed_invocation_observability_hooks() -> None:
     if plugin_home is not None:
         run_kwargs["cwd"] = str(plugin_home)
 
+    tool_call_started_at = time.perf_counter()
+
+    def _tool_call_duration_ms() -> int:
+        return max(1, int(round((time.perf_counter() - tool_call_started_at) * 1000)))
+
+    def _output_digest(stdout_bytes: bytes, stderr_bytes: bytes) -> str:
+        return "sha256:" + hashlib.sha256(stdout_bytes + stderr_bytes).hexdigest()
+
+    def _dispatch_failed_after_tool_call(
+        *,
+        output_digest: str,
+        duration_ms: int,
+        exit_code: int | None,
+    ) -> None:
+        dispatch_after_tool_call(
+            manifest=manifest,
+            tool=tool_name,
+            status="failed",
+            output_digest=output_digest,
+            duration_ms=duration_ms,
+            correlation_id=correlation_id,
+            invocation_id=tool_call_invocation_id,
+            event_sink=_emit,
+            exit_code=exit_code,
+            namespace=namespace,
+            command_name=command_name,
+            trust_state=trust_state,
+            plugin_home=plugin_home,
+            subprocess_runner=runner,
+        )
+
     try:
         completed = runner(cmd_argv, **run_kwargs)
     except FileNotFoundError as exc:
@@ -1191,6 +1226,11 @@ def _run_failed_invocation_observability_hooks() -> None:
                 provenance={"correlation_id": correlation_id},
             )
         )
+        _dispatch_failed_after_tool_call(
+            output_digest=_output_digest(b"", message.encode("utf-8", errors="surrogateescape")),
+            duration_ms=_tool_call_duration_ms(),
+            exit_code=127,
+        )
         _run_failed_invocation_observability_hooks()
         return InvocationResult(
             status="failed",
@@ -1208,6 +1248,7 @@ def _run_failed_invocation_observability_hooks() -> None:
         stderr_bytes = _to_bytes(exc.stderr)
         stdout_hash = hashlib.sha256(stdout_bytes).hexdigest()
         stderr_hash = hashlib.sha256(stderr_bytes).hexdigest()
+        output_digest = _output_digest(stdout_bytes, stderr_bytes)
         message = (
             f"entrypoint timed out after "
             f"{DEFAULT_PLUGIN_INVOCATION_TIMEOUT_SECONDS:g}s: {cmd_argv[0]!r}"
@@ -1231,6 +1272,11 @@ def _run_failed_invocation_observability_hooks() -> None:
                 },
             )
         )
+        _dispatch_failed_after_tool_call(
+            output_digest=output_digest,
+            duration_ms=_tool_call_duration_ms(),
+            exit_code=124,
+        )
         _run_failed_invocation_observability_hooks()
         return InvocationResult(
             status="failed",
@@ -1266,6 +1312,11 @@ def _run_failed_invocation_observability_hooks() -> None:
                 },
             )
         )
+        _dispatch_failed_after_tool_call(
+            output_digest=_output_digest(b"", message.encode("utf-8", errors="surrogateescape")),
+            duration_ms=_tool_call_duration_ms(),
+            exit_code=126,
+        )
         _run_failed_invocation_observability_hooks()
         return InvocationResult(
             status="failed",
@@ -1278,7 +1329,8 @@ def _run_failed_invocation_observability_hooks() -> None:
     stderr_bytes = _to_bytes(completed.stderr)
     stdout_hash = hashlib.sha256(stdout_bytes).hexdigest()
     stderr_hash = hashlib.sha256(stderr_bytes).hexdigest()
-    output_digest = hashlib.sha256(stdout_bytes + stderr_bytes).hexdigest()
+    output_digest = _output_digest(stdout_bytes, stderr_bytes)
+    duration_ms = _tool_call_duration_ms()
 
     terminal_provenance = {
         "correlation_id": correlation_id,
@@ -1307,7 +1359,7 @@ def _run_failed_invocation_observability_hooks() -> None:
             tool=tool_name,
             status="success",
             output_digest=output_digest,
-            duration_ms=0,
+            duration_ms=duration_ms,
             correlation_id=correlation_id,
             invocation_id=tool_call_invocation_id,
             event_sink=_emit,
@@ -1347,7 +1399,7 @@ def _run_failed_invocation_observability_hooks() -> None:
         tool=tool_name,
         status="failed",
         output_digest=output_digest,
-        duration_ms=0,
+        duration_ms=duration_ms,
         correlation_id=correlation_id,
         invocation_id=tool_call_invocation_id,
         event_sink=_emit,

tests/integration/plugin/test_e2e.py

@@ -677,7 +677,13 @@ def test_path_5_before_tool_call_can_block_production_invocation(tmp_path: Path)
     _enable_v04_tool_call_hooks(
         plugin_home,
         before_policy="fail_closed",
-        before_script="import sys\nsys.exit(7)\n",
+        before_script=(
+            "import json, os, sys\n"
+            "from pathlib import Path\n"
+            "payload = json.loads(os.environ['OUROBOROS_PLUGIN_TOOL_CALL_PAYLOAD'])\n"
+            "Path('before-tool-payload.json').write_text(json.dumps(payload, sort_keys=True))\n"
+            "sys.exit(7)\n"
+        ),
     )
     _grant_trust(paths)
     _grant_tool_hook_trust(paths)
@@ -702,6 +708,8 @@ def test_path_5_before_tool_call_can_block_production_invocation(tmp_path: Path)
     types = [p["event_type"] for p in payloads]
     assert "plugin.tool.intercept.blocked" in types
     assert "plugin.completed" not in types
+    observed = json.loads((plugin_home / "before-tool-payload.json").read_text())
+    assert observed["args_digest"].startswith("sha256:")
     failed = next(p for p in payloads if p["event_type"] == "plugin.failed")
     assert failed["result"]["status"] == "blocked"
     assert failed["provenance"]["reason"] == "tool_call_blocked"
@@ -772,10 +780,12 @@ def test_path_5_after_tool_call_observes_completed_result(tmp_path: Path) -> Non
     observed = json.loads((plugin_home / "after-tool-payload.json").read_text())
     assert observed["status"] == "success"
     assert observed["tool"] == "github-pr.review"
-    expected_digest = hashlib.sha256(
-        (result.stdout_bytes or b"") + (result.stderr_bytes or b"")
-    ).hexdigest()
+    expected_digest = (
+        "sha256:"
+        + hashlib.sha256((result.stdout_bytes or b"") + (result.stderr_bytes or b"")).hexdigest()
+    )
     assert observed["output_digest"] == expected_digest
+    assert observed["duration_ms"] >= 1
     payloads = [unwrap_plugin_event(env) for env in envelopes]
     assert "plugin.tool.observe.recorded" in [p["event_type"] for p in payloads]
 
@@ -814,11 +824,13 @@ def test_path_5_after_tool_call_hashes_combined_failed_output(tmp_path: Path) ->
     assert result.stderr_bytes == b"synthetic failure\n"
     observed = json.loads((plugin_home / "after-tool-failed-payload.json").read_text())
     assert observed["status"] == "failed"
-    expected_digest = hashlib.sha256(
-        (result.stdout_bytes or b"") + (result.stderr_bytes or b"")
-    ).hexdigest()
+    expected_digest = (
+        "sha256:"
+        + hashlib.sha256((result.stdout_bytes or b"") + (result.stderr_bytes or b"")).hexdigest()
+    )
     assert observed["output_digest"] == expected_digest
     assert observed["output_digest"] != result.stdout_sha256
+    assert observed["duration_ms"] >= 1
     payloads = [unwrap_plugin_event(env) for env in envelopes]
     assert "plugin.tool.observe.recorded" in [p["event_type"] for p in payloads]
 

tests/unit/plugin/test_firewall.py

@@ -17,6 +17,7 @@
     HOOK_EVENT_TYPES,
     HOOK_LIFECYCLE_POLICY_SCOPE,
     HOOK_LIFECYCLE_READ_SCOPE,
+    HOOK_TOOL_OBSERVE_SCOPE,
 )
 from ouroboros.plugin.manifest import load_manifest
 from ouroboros.plugin.trust_store import TrustRecord, TrustStore
@@ -2028,6 +2029,72 @@ def test_entrypoint_missing_emits_failed_127(tmp_path: Path) -> None:
     assert "not found" in result.message.lower()
 
 
+def test_v04_after_tool_call_observes_launch_failure_paths(tmp_path: Path) -> None:
+    payload = json.loads(json.dumps(REFERENCE_MANIFEST))
+    payload["schema_version"] = "0.4"
+    payload["permissions"].append(
+        {
+            "scope": HOOK_TOOL_OBSERVE_SCOPE,
+            "risk": "read_only",
+            "required": True,
+            "reason": "Allow tool-call observation.",
+        }
+    )
+    payload["hooks"] = [
+        {
+            "name": "after_tool_call",
+            "entrypoint": {"type": "command", "command": "python -m hook_after_tool"},
+            "permissions": [HOOK_TOOL_OBSERVE_SCOPE],
+            "failure_policy": "fail_open",
+        }
+    ]
+    program = _make_program(tmp_path, payload)
+    trust = _grant_trust_scopes(tmp_path, "github:read", HOOK_TOOL_OBSERVE_SCOPE)
+    observed_payloads: list[dict] = []
+
+    def _runner(argv, *args, **kwargs) -> subprocess.CompletedProcess:
+        if argv[:3] == ["python", "-m", "hook_after_tool"]:
+            observed_payloads.append(
+                json.loads(kwargs["env"]["OUROBOROS_PLUGIN_TOOL_CALL_PAYLOAD"])
+            )
+            return subprocess.CompletedProcess(args=argv, returncode=0, stdout="", stderr="")
+        raise FileNotFoundError("synthetic missing entrypoint")
+
+    events: list[dict] = []
+    result = invoke_plugin(
+        program,
+        command_name="review",
+        argv=["url"],
+        trust_record=trust,
+        event_sink=events.append,
+        correlation_id="corr-v04-launch-failure-after-tool",
+        subprocess_runner=_runner,
+    )
+
+    assert result.status == "failed"
+    assert result.exit_code == 127
+    assert observed_payloads == [
+        {
+            "correlation_id": "corr-v04-launch-failure-after-tool",
+            "duration_ms": observed_payloads[0]["duration_ms"],
+            "exit_code": 127,
+            "invocation_id": observed_payloads[0]["invocation_id"],
+            "output_digest": observed_payloads[0]["output_digest"],
+            "status": "failed",
+            "tool": "github-pr.review",
+        }
+    ]
+    assert observed_payloads[0]["duration_ms"] >= 1
+    assert observed_payloads[0]["output_digest"].startswith("sha256:")
+    assert [event["event_type"] for event in events] == [
+        "plugin.invoked",
+        "plugin.permission_used",
+        "plugin.permission_used",
+        "plugin.failed",
+        "plugin.tool.observe.recorded",
+    ]
+
+
 def test_subprocess_invocation_uses_default_timeout(tmp_path: Path) -> None:
     """The firewall owns the external plugin process boundary, so the
     subprocess launch must always carry a finite timeout."""