diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json new file mode 100644 index 0000000..24f1916 --- /dev/null +++ b/.claude-plugin/marketplace.json @@ -0,0 +1,19 @@ +{ + "name": "marginal", + "owner": { + "name": "SignalLayer Labs", + "url": "https://github.com/SignalLayerLabs/Marginal" + }, + "metadata": { + "description": "Compute governance that has to justify its own cost.", + "version": "0.3.3" + }, + "plugins": [ + { + "name": "marginal-claude-code", + "source": "./plugins/marginal-claude-code", + "description": "Local-first compute governance evidence for Claude Code, in Shadow Mode.", + "category": "productivity" + } + ] +} diff --git a/CHANGELOG.md b/CHANGELOG.md index 8db83a8..036d0d7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,22 @@ All notable changes to MARGINAL are documented here. The project follows Semanti ## [Unreleased] +### Added + +- a Claude Code plugin labeled **Observe**: it records normalized tool-call evidence and + repeated-work recommendations in a local Decision Ledger, declares no control capability, and never + blocks a tool call or returns hook output; +- `marginal.integrations.hookkit`, the engine-independent parts of a hook integration: normalized + events, privacy-safe action normalization, conservative structured outcome classification, + workspace state evidence that fails open, and session correlation; +- `marginal install claude-code` and `marginal uninstall claude-code`. + +### Changed + +- the authenticated loopback session transport moved from `marginal.integrations.codex.transport` to + `marginal.integrations.transport` so every hook adapter shares it. The old import path re-exports + it unchanged. + ## [0.3.3] - 2026-08-13 ### Fixed diff --git a/docs/integrations/claude-code.md b/docs/integrations/claude-code.md new file mode 100644 index 0000000..29cd01e --- /dev/null +++ b/docs/integrations/claude-code.md @@ -0,0 +1,97 @@ +# Claude Code Plugin + +Capability label: **Observe**. + +The plugin records normalized tool-call evidence and repeated-work recommendations in a local +Decision Ledger. It never blocks a tool call, never rewrites tool arguments, and never returns +output to Claude Code. Shadow Mode is the only mode this integration supports today. + +Validated against Claude Code 2.1.233 on Linux. + +## Install + +```bash +claude plugin marketplace add SignalLayerLabs/Marginal +claude plugin install marginal-claude-code@marginal +``` + +`marginal install claude-code` runs the same two commands through the Claude Code CLI. + +The plugin needs the `marginal-ai` package importable by the `python3` on your `PATH`: + +```bash +python3 -m pip install --user marginal-ai +``` + +If `marginal` cannot be imported, every hook exits 0 without output and Claude Code behaves exactly +as if the plugin were absent. The plugin never installs anything on your behalf. + +Point `MARGINAL_RUNTIME` at a directory or zipapp to load MARGINAL from somewhere else. + +## Remove + +```bash +claude plugin uninstall marginal-claude-code@marginal +``` + +Removal leaves the ledger in place. Delete the plugin data directory to discard the evidence. + +## What it observes + +| Hook | What MARGINAL records | +|---|---| +| `SessionStart` | starts one authenticated loopback service for the session | +| `PreToolUse` | a normalized action, its semantic key, workspace state hash, and the repetition signal | +| `PostToolUse` | a **proven success** with engine-measured `duration_ms` | +| `PostToolUseFailure` | a **proven failure**, or `unknown` when the call was interrupted | +| `SessionEnd` | a session summary, then closes the service | + +Claude Code separates success from failure at the event level, so outcome is an engine-declared fact +rather than something inferred from response text. That is a stronger evidence surface than a single +completion hook provides. Two limits still apply: + +- per-tool token usage is not exposed, so token cost is reported as unavailable rather than as zero; +- a hook that Claude Code does not deliver leaves a proposal unmatched, and the session records it as + `unknown` instead of assuming success. + +## Repeated work + +A repeat is only interesting when the same semantic action runs against the same workspace state and +produces the same completion evidence. The ledger escalates through +`NO_PROGRESS_OBSERVED` to `NO_PROGRESS_ENFORCEMENT_ELIGIBLE`, and Shadow Mode still allows the +action, recording `SHADOW_OVERRIDE`. Nothing is blocked. + +Read the recommendations back with the standard tooling: + +```bash +marginal ledger-report "$CLAUDE_PLUGIN_DATA/ledger"/*/*.jsonl +``` + +## Enforcement + +This integration does not enforce. The adapter declares no control capability, so the core refuses to +run it in a blocking mode. The documented deny transport is implemented and tested +(`marginal.integrations.claude_code.decisions`) but nothing calls it. + +Turning it on requires an evidence gate equivalent to the Codex Earned Enforcement window: proven hook +coverage, reviewed stop candidates, zero false stops, and bounded governance latency. Until that +evidence exists for Claude Code, a stop recommendation stays a recommendation. + +## Privacy and limits + +The ledger is written under the plugin's own data directory with owner-only permissions. + +- tool arguments, command text, tool output, prompts, and transcripts are never persisted; only + digests of them are; +- error text from a failed tool contributes to an evidence digest and is not written to the ledger; +- `LOCAL_FULL` is the default profile because the ledger stays on the user's machine. It retains + session identifiers and tool names; +- set `MARGINAL_PRIVACY_PROFILE=safe_telemetry` to write keyed pseudonyms instead. Pseudonymization is + not anonymization; +- Claude Code's own transcripts, shell history, and telemetry are outside this boundary. + +## Workspace state + +State evidence comes from Git. A session directory inside a work tree is observable, including a +repository with no commits yet. Outside a repository the state hash is empty, which makes every +repetition control fail open rather than invent certainty. diff --git a/docs/integrations/overview.md b/docs/integrations/overview.md index c187888..7f2fdfd 100644 --- a/docs/integrations/overview.md +++ b/docs/integrations/overview.md @@ -75,6 +75,16 @@ It provides lifecycle correlation, privacy-safe normalization, outcome classific authenticated local service, Shadow Mode, Earned Enforcement receipts, and reversible native installation. See [Codex plugin](codex.md). -OpenCode, Claude Code, and GitHub Copilot remain roadmap work. Codex is labeled Tool Enforcement, +The Claude Code plugin is labeled **Observe**. It records normalized evidence and repeated-work +recommendations in a local Decision Ledger, declares no control capability, and never blocks a tool +call. Its outcome evidence is engine-declared, because Claude Code reports success and failure as +separate hook events. See [Claude Code plugin](claude-code.md). + +OpenCode and GitHub Copilot remain roadmap work. Codex is labeled Tool Enforcement, not Full Compute Enforcement, because specialized and hosted tool paths can fall outside local hook coverage. + +Engine-independent parts of a hook integration live in `marginal.integrations.hookkit`: normalized +events, action normalization, structured outcome classification, workspace state evidence, and +session correlation. The Codex integration predates that module and still carries its own copies; +migrating it is separate work so a new adapter never destabilizes the validated Codex path. diff --git a/plugins/marginal-claude-code/.claude-plugin/plugin.json b/plugins/marginal-claude-code/.claude-plugin/plugin.json new file mode 100644 index 0000000..f7a8dc5 --- /dev/null +++ b/plugins/marginal-claude-code/.claude-plugin/plugin.json @@ -0,0 +1,18 @@ +{ + "name": "marginal-claude-code", + "version": "0.3.3", + "description": "Local-first compute governance evidence for Claude Code. Shadow Mode only: it observes repeated work and never blocks a tool call.", + "author": { + "name": "SignalLayer Labs", + "url": "https://github.com/SignalLayerLabs/Marginal" + }, + "homepage": "https://signallayerlabs.github.io/Marginal/", + "repository": "https://github.com/SignalLayerLabs/Marginal", + "license": "Apache-2.0", + "keywords": [ + "agent-compute", + "compute-governance", + "claude-code", + "token-efficiency" + ] +} diff --git a/plugins/marginal-claude-code/hooks/hooks.json b/plugins/marginal-claude-code/hooks/hooks.json new file mode 100644 index 0000000..402e62a --- /dev/null +++ b/plugins/marginal-claude-code/hooks/hooks.json @@ -0,0 +1,68 @@ +{ + "description": "Local-only MARGINAL Shadow Mode observation for Claude Code.", + "hooks": { + "SessionStart": [ + { + "hooks": [ + { + "type": "command", + "command": "python3 \"${CLAUDE_PLUGIN_ROOT}/scripts/marginal_hook.py\"", + "timeout": 10, + "statusMessage": "Starting MARGINAL Shadow Mode" + } + ] + } + ], + "PreToolUse": [ + { + "matcher": "*", + "hooks": [ + { + "type": "command", + "command": "python3 \"${CLAUDE_PLUGIN_ROOT}/scripts/marginal_hook.py\"", + "timeout": 5, + "statusMessage": "Measuring marginal value" + } + ] + } + ], + "PostToolUse": [ + { + "matcher": "*", + "hooks": [ + { + "type": "command", + "command": "python3 \"${CLAUDE_PLUGIN_ROOT}/scripts/marginal_hook.py\"", + "timeout": 5, + "statusMessage": "Recording redacted completion evidence" + } + ] + } + ], + "PostToolUseFailure": [ + { + "matcher": "*", + "hooks": [ + { + "type": "command", + "command": "python3 \"${CLAUDE_PLUGIN_ROOT}/scripts/marginal_hook.py\"", + "timeout": 5, + "statusMessage": "Recording redacted failure evidence" + } + ] + } + ], + "SessionEnd": [ + { + "hooks": [ + { + "type": "command", + "command": "python3 \"${CLAUDE_PLUGIN_ROOT}/scripts/marginal_hook.py\"", + "timeout": 5, + "statusMessage": "Closing MARGINAL session" + } + ] + } + ] + } +} diff --git a/plugins/marginal-claude-code/scripts/marginal_hook.py b/plugins/marginal-claude-code/scripts/marginal_hook.py new file mode 100644 index 0000000..d3b3030 --- /dev/null +++ b/plugins/marginal-claude-code/scripts/marginal_hook.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 +"""Dependency-free launcher for the MARGINAL Claude Code hook. + +The launcher never installs anything and never fails a hook. If the ``marginal`` +package cannot be imported, it exits 0 with no output and Claude Code proceeds +exactly as if MARGINAL were not present. + +Resolution order: + +1. ``MARGINAL_RUNTIME`` — a directory or zipapp added to ``sys.path``; +2. ``runtime/marginal_runtime.pyz`` bundled next to this script, when present; +3. the ambient interpreter's own ``marginal`` installation. +""" + +from __future__ import annotations + +import os +import sys +from pathlib import Path + + +def _candidate_paths() -> list[str]: + candidates: list[str] = [] + override = os.environ.get("MARGINAL_RUNTIME") + if override: + candidates.append(override) + plugin_root = os.environ.get("CLAUDE_PLUGIN_ROOT") + root = Path(plugin_root) if plugin_root else Path(__file__).resolve().parent.parent + bundled = root / "runtime" / "marginal_runtime.pyz" + if bundled.is_file(): + candidates.append(str(bundled)) + return candidates + + +def main() -> int: + if sys.version_info < (3, 10): # noqa: UP036 - plugin bootstrap must fail open on unsupported Python + return 0 + for candidate in _candidate_paths(): + if candidate not in sys.path: + sys.path.insert(0, candidate) + try: + from marginal.integrations.claude_code.service import hook_main + except Exception: + return 0 + try: + return hook_main([]) + except Exception: + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/plugins/marginal/runtime/marginal_runtime.pyz b/plugins/marginal/runtime/marginal_runtime.pyz index 750ced9..3f44428 100644 Binary files a/plugins/marginal/runtime/marginal_runtime.pyz and b/plugins/marginal/runtime/marginal_runtime.pyz differ diff --git a/plugins/marginal/runtime/provenance.json b/plugins/marginal/runtime/provenance.json index 141cb9f..1e042f2 100644 --- a/plugins/marginal/runtime/provenance.json +++ b/plugins/marginal/runtime/provenance.json @@ -1 +1 @@ -{"builder":"scripts/build_codex_plugin.py","python_requires":">=3.10","schema_version":1,"sha256":"57b0f86a9c49daff2c0213d5606d2c369c4e39580b52cd76cfd20fdb2de33167","source_hash":"fccff1138073eb96d48fbb60470bb7f4d6d27770da7a2f61bc490b83ee3a912a"} +{"builder":"scripts/build_codex_plugin.py","python_requires":">=3.10","schema_version":1,"sha256":"97a3774d21d4fa2e461cfabe0401ecfd474274fd9550534152cd6a44e5f3945a","source_hash":"e4f8f0486b6d71a20c3e53db928692918332c38789b0a88c4ff004e3f7ea90b5"} diff --git a/src/marginal/cli.py b/src/marginal/cli.py index 3802051..76a7ba7 100644 --- a/src/marginal/cli.py +++ b/src/marginal/cli.py @@ -154,7 +154,7 @@ def _build_parser() -> argparse.ArgumentParser: public_eval.add_argument("--seed", type=int, default=42) install_parser = subparsers.add_parser("install", help="install a native integration") - install_parser.add_argument("target", choices=["codex"]) + install_parser.add_argument("target", choices=["codex", "claude-code"]) install_parser.add_argument("--repository", default="SignalLayerLabs/Marginal") install_parser.add_argument("--ref", default="main") install_parser.add_argument("--data-dir", type=Path) @@ -162,7 +162,7 @@ def _build_parser() -> argparse.ArgumentParser: install_parser.add_argument("--json", action="store_true", dest="as_json") uninstall_parser = subparsers.add_parser("uninstall", help="remove a native integration") - uninstall_parser.add_argument("target", choices=["codex"]) + uninstall_parser.add_argument("target", choices=["codex", "claude-code"]) uninstall_parser.add_argument("--purge-data", action="store_true") uninstall_parser.add_argument("--yes", action="store_true") uninstall_parser.add_argument("--data-dir", type=Path) @@ -201,6 +201,23 @@ def main(argv: Sequence[str] | None = None) -> int: args = parser.parse_args(argv) if args.command == "install": + if args.target == "claude-code": + from .integrations.claude_code.installer import ( + MARKETPLACE_SOURCE, + ) + from .integrations.claude_code.installer import ( + install as install_claude_code, + ) + + claude_result = install_claude_code( + marketplace_source=args.repository or MARKETPLACE_SOURCE + ) + if args.as_json: + print(json.dumps(claude_result.to_dict(), sort_keys=True)) + else: + print(claude_result.message or claude_result.error_code or claude_result.selector) + return 0 if claude_result.installed else 1 + from .integrations.codex.installer import install result = install( @@ -217,6 +234,16 @@ def main(argv: Sequence[str] | None = None) -> int: return 0 if result.installed else 1 if args.command == "uninstall": + if args.target == "claude-code": + from .integrations.claude_code.installer import uninstall as uninstall_claude_code + + claude_result = uninstall_claude_code() + if args.as_json: + print(json.dumps(claude_result.to_dict(), sort_keys=True)) + else: + print(claude_result.message or claude_result.error_code or claude_result.selector) + return 0 if not claude_result.installed else 1 + from .integrations.codex.commands import default_data_dir, purge_data from .integrations.codex.installer import uninstall diff --git a/src/marginal/integrations/claude_code/__init__.py b/src/marginal/integrations/claude_code/__init__.py new file mode 100644 index 0000000..3e6de9c --- /dev/null +++ b/src/marginal/integrations/claude_code/__init__.py @@ -0,0 +1,60 @@ +"""Claude Code integration. + +Capability label: **Observe**. The adapter records normalized tool-call evidence +and repeated-work recommendations in a local Decision Ledger. It never blocks a +tool call and never changes what Claude Code does next. + +Claude Code separates ``PostToolUse`` from ``PostToolUseFailure``, so success and +failure are engine-declared facts here rather than inferences, and ``duration_ms`` +gives measured latency. Per-tool token usage is not exposed by the hook surface and +is reported as unavailable rather than as zero. +""" + +from .events import ( + PostToolUseEvent, + PostToolUseFailureEvent, + PreToolUseEvent, + SessionEvent, + parse_hook_event, +) +from .installer import ( + PLUGIN_SELECTOR, + ClaudeCodeDoctorReport, + ClaudeCodeInstallation, + inspect_claude_code, + install, + uninstall, +) +from .normalization import ENGINE, session_boundary, tool_call_end, tool_call_start +from .service import ( + HookResult, + hook_main, + observe_outcome, + run_hook, + start_session_service, + stop_session_service, +) + +__all__ = [ + "ENGINE", + "PLUGIN_SELECTOR", + "ClaudeCodeDoctorReport", + "ClaudeCodeInstallation", + "HookResult", + "PostToolUseEvent", + "PostToolUseFailureEvent", + "PreToolUseEvent", + "SessionEvent", + "hook_main", + "inspect_claude_code", + "install", + "observe_outcome", + "parse_hook_event", + "run_hook", + "session_boundary", + "start_session_service", + "stop_session_service", + "tool_call_end", + "tool_call_start", + "uninstall", +] diff --git a/src/marginal/integrations/claude_code/decisions.py b/src/marginal/integrations/claude_code/decisions.py new file mode 100644 index 0000000..48e78e7 --- /dev/null +++ b/src/marginal/integrations/claude_code/decisions.py @@ -0,0 +1,55 @@ +"""Claude Code hook output shapes. + +Shadow Mode emits nothing. A hook that printed an advisory would change what the +model sees, which is exactly what Shadow Mode promises not to do, so the only +Shadow Mode output is an empty one and the recommendation goes to the ledger. + +The blocking builders below encode the documented deny contract so a later Earned +Enforcement gate has a validated transport to use. Nothing in this integration +calls them yet: the adapter declares no blocking capability, and the core refuses +to run a non-blocking adapter in an enforcing mode. +""" + +from __future__ import annotations + +from typing import Any + + +def _reason_with_code(reason: str, reason_code: str) -> str: + if not isinstance(reason, str) or not reason.strip(): + raise ValueError("reason must be a non-empty string") + if not isinstance(reason_code, str) or not reason_code.strip(): + raise ValueError("reason_code must be a non-empty string") + return f"{reason.strip()} [{reason_code.strip()}]" + + +def build_pre_tool_use_output( + *, allowed: bool, reason: str, reason_code: str +) -> dict[str, Any] | None: + """Build the documented Claude Code ``PreToolUse`` denial shape.""" + + if allowed: + return None + return { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": _reason_with_code(reason, reason_code), + } + } + + +def build_post_tool_use_output( + *, blocked: bool, reason: str, reason_code: str +) -> dict[str, Any] | None: + """Build the documented Claude Code ``PostToolUse`` blocking shape.""" + + if not blocked: + return None + return { + "hookSpecificOutput": { + "hookEventName": "PostToolUse", + "decision": "block", + "reason": _reason_with_code(reason, reason_code), + } + } diff --git a/src/marginal/integrations/claude_code/events.py b/src/marginal/integrations/claude_code/events.py new file mode 100644 index 0000000..cf4ffe9 --- /dev/null +++ b/src/marginal/integrations/claude_code/events.py @@ -0,0 +1,177 @@ +"""Strict value objects for the supported Claude Code hook lifecycle. + +Field names follow the payloads Claude Code actually sends, verified against +Claude Code 2.1.233. Only the hooks MARGINAL has an explicit contract for are +parsed; anything else raises rather than being guessed at. + +``transcript_path`` and ``prompt`` are deliberately never retained. The transcript +is outside the Decision Ledger privacy boundary and prompt material is not needed +to detect repeated work. +""" + +from __future__ import annotations + +import math +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any + +SESSION_EVENTS = frozenset({"SessionStart", "SessionEnd"}) +TOOL_EVENTS = frozenset({"PreToolUse", "PostToolUse", "PostToolUseFailure"}) +SUPPORTED_EVENTS = SESSION_EVENTS | TOOL_EVENTS + + +@dataclass(frozen=True, slots=True) +class SessionEvent: + session_id: str + cwd: str + hook_event_name: str + source: str = "" + reason: str = "" + + +@dataclass(frozen=True, slots=True) +class PreToolUseEvent: + session_id: str + cwd: str + hook_event_name: str + tool_name: str + tool_use_id: str + tool_input: Mapping[str, Any] + prompt_id: str = "" + permission_mode: str = "" + + +@dataclass(frozen=True, slots=True) +class PostToolUseEvent: + """A tool call Claude Code documents as having completed successfully.""" + + session_id: str + cwd: str + hook_event_name: str + tool_name: str + tool_use_id: str + tool_input: Mapping[str, Any] + tool_response: Any + prompt_id: str = "" + permission_mode: str = "" + duration_ms: float | None = None + + +@dataclass(frozen=True, slots=True) +class PostToolUseFailureEvent: + """A tool call Claude Code documents as having failed. + + ``error`` is engine-supplied free text. It is used only to derive an evidence + digest and is never written to the ledger. + """ + + session_id: str + cwd: str + hook_event_name: str + tool_name: str + tool_use_id: str + tool_input: Mapping[str, Any] + error: str + prompt_id: str = "" + permission_mode: str = "" + duration_ms: float | None = None + is_interrupt: bool = False + + +ClaudeCodeHookEvent = SessionEvent | PreToolUseEvent | PostToolUseEvent | PostToolUseFailureEvent + + +def _required_text(payload: Mapping[str, Any], name: str) -> str: + value = payload.get(name) + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{name} must be a non-empty string") + return value + + +def _optional_text(payload: Mapping[str, Any], name: str) -> str: + value = payload.get(name) + if value is None: + return "" + if not isinstance(value, str): + raise ValueError(f"{name} must be a string or null") + return value + + +def _optional_duration(payload: Mapping[str, Any]) -> float | None: + value = payload.get("duration_ms") + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError("duration_ms must be a number or null") + duration = float(value) + if not math.isfinite(duration) or duration < 0.0: + raise ValueError("duration_ms must be finite and non-negative") + return duration + + +def _common(payload: Mapping[str, Any]) -> dict[str, Any]: + return { + "session_id": _required_text(payload, "session_id"), + "cwd": _required_text(payload, "cwd"), + "hook_event_name": _required_text(payload, "hook_event_name"), + } + + +def _tool_fields(payload: Mapping[str, Any]) -> dict[str, Any]: + tool_input = payload.get("tool_input") + if not isinstance(tool_input, Mapping): + raise ValueError("tool_input must be a mapping") + return { + "tool_name": _required_text(payload, "tool_name"), + "tool_use_id": _required_text(payload, "tool_use_id"), + "tool_input": dict(tool_input), + "prompt_id": _optional_text(payload, "prompt_id"), + "permission_mode": _optional_text(payload, "permission_mode"), + } + + +def _tool_response(payload: Mapping[str, Any]) -> Any: + """Accept either documented result key without inventing a value.""" + + for key in ("tool_response", "tool_output"): + if key in payload: + return payload[key] + raise ValueError("tool_response is required") + + +def parse_hook_event(payload: Mapping[str, Any]) -> ClaudeCodeHookEvent: + """Parse only hook events for which MARGINAL has an explicit contract.""" + + if not isinstance(payload, Mapping): + raise TypeError("Claude Code hook payload must be a mapping") + name = _required_text(payload, "hook_event_name") + if name not in SUPPORTED_EVENTS: + raise ValueError(f"unsupported Claude Code hook event: {name}") + common = _common(payload) + if name in SESSION_EVENTS: + return SessionEvent( + **common, + source=_optional_text(payload, "source"), + reason=_optional_text(payload, "reason"), + ) + tool_fields = _tool_fields(payload) + if name == "PreToolUse": + return PreToolUseEvent(**common, **tool_fields) + if name == "PostToolUse": + return PostToolUseEvent( + **common, + **tool_fields, + tool_response=_tool_response(payload), + duration_ms=_optional_duration(payload), + ) + is_interrupt = payload.get("is_interrupt", False) + if not isinstance(is_interrupt, bool): + raise ValueError("is_interrupt must be a boolean") + return PostToolUseFailureEvent( + **common, + **tool_fields, + error=_optional_text(payload, "error"), + duration_ms=_optional_duration(payload), + is_interrupt=is_interrupt, + ) diff --git a/src/marginal/integrations/claude_code/installer.py b/src/marginal/integrations/claude_code/installer.py new file mode 100644 index 0000000..4256dbb --- /dev/null +++ b/src/marginal/integrations/claude_code/installer.py @@ -0,0 +1,210 @@ +"""Read-only Claude Code discovery and reversible native plugin operations. + +Installation is delegated to Claude Code's own plugin commands so that removal is +always possible with the same tool the user already trusts. Nothing here reads +credentials, transcripts, or project configuration. +""" + +from __future__ import annotations + +import os +import re +import shutil +import subprocess +from dataclasses import dataclass +from pathlib import Path +from typing import Protocol + +MARKETPLACE_NAME = "marginal" +MARKETPLACE_SOURCE = "SignalLayerLabs/Marginal" +PLUGIN_NAME = "marginal-claude-code" +PLUGIN_SELECTOR = f"{PLUGIN_NAME}@{MARKETPLACE_NAME}" +_VERSION_PATTERN = re.compile(r"(\d+\.\d+\.\d+)") +_MINIMUM_VERSION = (2, 1, 0) + + +@dataclass(frozen=True, slots=True) +class CommandResult: + returncode: int + stdout: str + stderr: str + + +class CommandRunner(Protocol): + def run(self, args: list[str]) -> CommandResult: ... + + +class SubprocessRunner: + """Run Claude Code without a shell or inherited secret variables.""" + + def __init__(self, *, timeout_seconds: float = 30.0, max_output_bytes: int = 1_048_576): + self.timeout_seconds = timeout_seconds + self.max_output_bytes = max_output_bytes + + def run(self, args: list[str]) -> CommandResult: + environment = { + name: value + for name in ("PATH", "HOME", "CLAUDE_CONFIG_DIR", "LANG", "LC_ALL", "SYSTEMROOT") + if (value := os.environ.get(name)) is not None + } + try: + completed = subprocess.run( + args, + check=False, + capture_output=True, + text=False, + timeout=self.timeout_seconds, + env=environment, + stdin=subprocess.DEVNULL, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + return CommandResult(127, "", type(exc).__name__) + stdout = completed.stdout[: self.max_output_bytes].decode("utf-8", errors="replace") + stderr = completed.stderr[: self.max_output_bytes].decode("utf-8", errors="replace") + return CommandResult(completed.returncode, stdout, stderr) + + +@dataclass(frozen=True, slots=True) +class ClaudeCodeDoctorReport: + """What can be established about a Claude Code installation without secrets.""" + + available: bool + version: str + plugins_supported: bool + capability_level: str + blocking_reasons: tuple[str, ...] + + def to_dict(self) -> dict[str, object]: + return { + "available": self.available, + "version": self.version, + "plugins_supported": self.plugins_supported, + "capability_level": self.capability_level, + "capability_label": "Observe", + "blocking_reasons": list(self.blocking_reasons), + } + + +@dataclass(frozen=True, slots=True) +class ClaudeCodeInstallation: + installed: bool + changed: bool + selector: str = PLUGIN_SELECTOR + error_code: str = "" + message: str = "" + + def to_dict(self) -> dict[str, object]: + return { + "installed": self.installed, + "changed": self.changed, + "selector": self.selector, + "error_code": self.error_code, + "message": self.message, + } + + +def _executable() -> str | None: + return shutil.which("claude") + + +def _parse_version(output: str) -> str: + match = _VERSION_PATTERN.search(output) + return match.group(1) if match else "" + + +def _version_tuple(version: str) -> tuple[int, ...]: + try: + return tuple(int(part) for part in version.split(".")) + except ValueError: + return () + + +def inspect_claude_code(*, runner: CommandRunner | None = None) -> ClaudeCodeDoctorReport: + """Report installation facts. Absence is a normal answer, not an error.""" + + executable = _executable() + if executable is None: + return ClaudeCodeDoctorReport( + available=False, + version="", + plugins_supported=False, + capability_level="none", + blocking_reasons=("CLAUDE_CODE_NOT_FOUND",), + ) + selected = runner or SubprocessRunner() + version_result = selected.run([executable, "--version"]) + version = _parse_version(version_result.stdout or version_result.stderr) + reasons: list[str] = [] + if version_result.returncode != 0: + reasons.append("VERSION_UNAVAILABLE") + parsed = _version_tuple(version) + plugins_supported = bool(parsed) and parsed >= _MINIMUM_VERSION + if not plugins_supported: + reasons.append("PLUGIN_SUPPORT_UNVERIFIED") + return ClaudeCodeDoctorReport( + available=True, + version=version, + plugins_supported=plugins_supported, + capability_level="observe" if plugins_supported else "none", + blocking_reasons=tuple(reasons), + ) + + +def install( + *, + runner: CommandRunner | None = None, + marketplace_source: str | Path = MARKETPLACE_SOURCE, +) -> ClaudeCodeInstallation: + """Add the marketplace and install the plugin through Claude Code itself.""" + + executable = _executable() + if executable is None: + return ClaudeCodeInstallation( + installed=False, + changed=False, + error_code="CLAUDE_CODE_NOT_FOUND", + message="claude was not found on PATH", + ) + selected = runner or SubprocessRunner() + marketplace = selected.run( + [executable, "plugin", "marketplace", "add", str(marketplace_source)] + ) + if marketplace.returncode != 0 and "already" not in marketplace.stderr.casefold(): + return ClaudeCodeInstallation( + installed=False, + changed=False, + error_code="MARKETPLACE_FAILED", + message=marketplace.stderr.strip(), + ) + installation = selected.run([executable, "plugin", "install", PLUGIN_SELECTOR]) + if installation.returncode != 0: + return ClaudeCodeInstallation( + installed=False, + changed=False, + error_code="INSTALL_FAILED", + message=installation.stderr.strip(), + ) + return ClaudeCodeInstallation(installed=True, changed=True) + + +def uninstall(*, runner: CommandRunner | None = None) -> ClaudeCodeInstallation: + """Remove the plugin. A plugin that is already absent is a success.""" + + executable = _executable() + if executable is None: + return ClaudeCodeInstallation( + installed=False, + changed=False, + error_code="CLAUDE_CODE_NOT_FOUND", + message="claude was not found on PATH", + ) + selected = runner or SubprocessRunner() + removal = selected.run([executable, "plugin", "uninstall", PLUGIN_SELECTOR]) + if removal.returncode != 0 and "not installed" not in removal.stderr.casefold(): + return ClaudeCodeInstallation( + installed=True, + changed=False, + error_code="UNINSTALL_FAILED", + message=removal.stderr.strip(), + ) + return ClaudeCodeInstallation(installed=False, changed=removal.returncode == 0) diff --git a/src/marginal/integrations/claude_code/normalization.py b/src/marginal/integrations/claude_code/normalization.py new file mode 100644 index 0000000..9f407b8 --- /dev/null +++ b/src/marginal/integrations/claude_code/normalization.py @@ -0,0 +1,70 @@ +"""Map Claude Code hook events onto the engine-neutral hook contract. + +Claude Code separates success from failure at the event level: ``PostToolUse`` +documents a tool call that completed successfully and ``PostToolUseFailure`` +documents one that failed. That makes the outcome an engine-declared fact rather +than something inferred from response text. + +An interrupted call is neither. The tool never reached its own conclusion, so it +is reported as ``UNKNOWN`` instead of being charged as a tool failure. +""" + +from __future__ import annotations + +from marginal.controls import ActionOutcomeStatus + +from ..hookkit.events import SessionBoundary, ToolCallEnd, ToolCallStart +from .events import ( + PostToolUseEvent, + PostToolUseFailureEvent, + PreToolUseEvent, + SessionEvent, +) + +ENGINE = "claude-code" + + +def session_boundary(event: SessionEvent) -> SessionBoundary: + if not isinstance(event, SessionEvent): + raise TypeError("event must be a SessionEvent") + return SessionBoundary( + engine=ENGINE, + session_id=event.session_id, + workspace=event.cwd, + kind="start" if event.hook_event_name == "SessionStart" else "end", + source=event.source, + reason=event.reason, + ) + + +def tool_call_start(event: PreToolUseEvent) -> ToolCallStart: + if not isinstance(event, PreToolUseEvent): + raise TypeError("event must be a PreToolUseEvent") + return ToolCallStart( + session_id=event.session_id, + call_id=event.tool_use_id, + tool_name=event.tool_name, + tool_input=event.tool_input, + turn_id=event.prompt_id, + ) + + +def tool_call_end(event: PostToolUseEvent | PostToolUseFailureEvent) -> ToolCallEnd: + if isinstance(event, PostToolUseEvent): + outcome = ActionOutcomeStatus.SUCCESS + evidence: object = event.tool_response + elif isinstance(event, PostToolUseFailureEvent): + outcome = ActionOutcomeStatus.UNKNOWN if event.is_interrupt else ActionOutcomeStatus.FAILURE + evidence = {"error": event.error, "is_interrupt": event.is_interrupt} + else: + raise TypeError("event must be a PostToolUse or PostToolUseFailure event") + return ToolCallEnd( + session_id=event.session_id, + call_id=event.tool_use_id, + tool_name=event.tool_name, + outcome=outcome, + tool_input=event.tool_input, + evidence=evidence, + duration_ms=event.duration_ms, + turn_id=event.prompt_id, + ) diff --git a/src/marginal/integrations/claude_code/service.py b/src/marginal/integrations/claude_code/service.py new file mode 100644 index 0000000..704e3aa --- /dev/null +++ b/src/marginal/integrations/claude_code/service.py @@ -0,0 +1,413 @@ +"""Per-session Claude Code governance service and fail-open hook entry point. + +Claude Code runs every hook in its own short-lived process, so session state cannot +live in the hook. One authenticated loopback service per session owns the runtime, +exactly as the Codex integration does, which keeps authorization strictly before +execution and settlement strictly after it. + +Every failure path fails open. A hook that cannot reach its service, cannot parse +its payload, or raises for any reason exits 0 with no output, and Claude Code +proceeds as if MARGINAL were not installed. +""" + +from __future__ import annotations + +import json +import os +import secrets +import subprocess +import sys +import threading +import time +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from marginal.controls import ActionOutcomeStatus +from marginal.privacy import PrivacyProfile + +from ..hookkit.bootstrap import ObserveSession, build_observe_session +from ..hookkit.session import HookSessionRuntime +from ..transport import ( + ConnectionInfo, + SessionHandler, + SessionServer, + connection_filename, + request_session, +) +from .events import ( + ClaudeCodeHookEvent, + PostToolUseEvent, + PostToolUseFailureEvent, + PreToolUseEvent, + SessionEvent, + parse_hook_event, +) +from .normalization import ENGINE, tool_call_end, tool_call_start + +DATA_ROOT_VARIABLES = ("CLAUDE_PLUGIN_DATA", "MARGINAL_CLAUDE_CODE_DATA", "PLUGIN_DATA") +_READY_TIMEOUT_SECONDS = 5.0 +_SERVERS: dict[tuple[Path, str], tuple[SessionServer, HookSessionRuntime]] = {} + + +@dataclass(frozen=True, slots=True) +class HookResult: + """What one hook invocation reports back to Claude Code.""" + + exit_code: int + output: dict[str, Any] | None = None + warning_code: str = "" + + +def data_root_from_environment() -> str: + """Return the first configured plugin data directory, or an empty string.""" + + for name in DATA_ROOT_VARIABLES: + value = os.environ.get(name) + if value: + return value + return "" + + +def _connection_path(data_root: Path, session_id: str) -> Path: + return data_root / "sessions" / connection_filename(session_id) + + +def _bootstrap_path(data_root: Path, session_id: str) -> Path: + bootstrap_root = data_root / "bootstrap" + bootstrap_root.mkdir(parents=True, exist_ok=True, mode=0o700) + if os.name == "posix": + bootstrap_root.chmod(0o700) + from ..hookkit.bootstrap import session_hash + + return bootstrap_root / f"{session_hash(session_id)}-{secrets.token_hex(8)}.json" + + +def _bootstrap_event_payload(event: SessionEvent) -> dict[str, Any]: + """Keep the service bootstrap free of transcript and unrelated hook fields.""" + + return { + "session_id": event.session_id, + "cwd": event.cwd, + "hook_event_name": event.hook_event_name, + "source": event.source, + } + + +def _handler( + session: HookSessionRuntime, + *, + observe: ObserveSession, + shutdown_event: threading.Event | None = None, +) -> SessionHandler: + def handle(operation: str, payload: Mapping[str, Any]) -> Mapping[str, Any] | None: + if operation == "status": + return {**session.summary(), "workspace_hash": observe.workspace_hash} + if operation == "close": + session.close() + observe.record( + { + "event": "hook_session_end", + **{f"summary_{name}": value for name, value in session.summary().items()}, + } + ) + if shutdown_event is not None: + threading.Timer(0.05, shutdown_event.set).start() + return session.summary() + + event = parse_hook_event(payload) + if operation == "pre" and isinstance(event, PreToolUseEvent): + started = time.perf_counter_ns() + decision = session.tool_call_start(tool_call_start(event)) + latency_ms = (time.perf_counter_ns() - started) / 1_000_000 + signal = session.last_no_progress_signal + observe.record( + { + "event": "hook_decision", + **(session.last_action_evidence or {}), + "reason_code": decision.reason_code, + "recommended": decision.recommended, + "recommended_stop": bool(signal and signal.should_recommend_stop), + "no_progress_reason_code": signal.reason_code if signal else "", + "governance_latency_ms": latency_ms, + "enforced": False, + } + ) + # Shadow Mode never changes what Claude Code does next. + return None + if operation == "post" and isinstance(event, (PostToolUseEvent, PostToolUseFailureEvent)): + evidence = session.action_evidence(event.tool_use_id) or {} + outcome = session.tool_call_end(tool_call_end(event)) + observe.record( + { + "event": "hook_outcome", + **evidence, + "outcome": outcome.value, + "outcome_source": event.hook_event_name, + "duration_ms": event.duration_ms, + } + ) + return None + raise ValueError(f"unsupported operation for event: {operation}") + + return handle + + +def _spawn_session_service(event: SessionEvent, *, data_root: Path) -> ConnectionInfo: + existing_path = _connection_path(data_root, event.session_id) + if existing_path.exists(): + try: + existing = ConnectionInfo.from_file(existing_path) + if request_session(existing, operation="status", payload={}).get("ok") is True: + return existing + except (OSError, ValueError, KeyError): + pass + existing_path.unlink(missing_ok=True) + + bootstrap = _bootstrap_path(data_root, event.session_id) + payload = { + "event": _bootstrap_event_payload(event), + "data_root": str(data_root), + "token": secrets.token_hex(32), + } + descriptor = os.open(bootstrap, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + try: + os.write( + descriptor, + json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8"), + ) + os.fsync(descriptor) + finally: + os.close(descriptor) + + executable = Path(sys.argv[0]).resolve() if sys.argv and sys.argv[0] else None + if executable is not None and executable.suffix == ".pyz": + command = [sys.executable, str(executable), "--serve", str(bootstrap)] + else: + command = [ + sys.executable, + "-m", + "marginal.integrations.claude_code.service", + "--serve", + str(bootstrap), + ] + environment = { + name: value + for name in ( + "PATH", + "LANG", + "LC_ALL", + "SYSTEMROOT", + "PYTHONPATH", + "MARGINAL_PRIVACY_PROFILE", + *DATA_ROOT_VARIABLES, + ) + if (value := os.environ.get(name)) is not None + } + subprocess.Popen( + command, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + env=environment, + close_fds=True, + start_new_session=True, + ) + deadline = time.monotonic() + _READY_TIMEOUT_SECONDS + while time.monotonic() < deadline: + if existing_path.exists(): + try: + connection = ConnectionInfo.from_file(existing_path) + if request_session(connection, operation="status", payload={}).get("ok") is True: + return connection + except (OSError, ValueError, KeyError): + pass + time.sleep(0.05) + bootstrap.unlink(missing_ok=True) + raise RuntimeError("Claude Code session service did not become ready") + + +def selected_privacy_profile() -> PrivacyProfile: + """Return the configured ledger privacy profile, defaulting to local-only. + + The ledger lives in the plugin's own data directory on the user's machine, so + ``LOCAL_FULL`` is the default. Set ``MARGINAL_PRIVACY_PROFILE`` to + ``safe_telemetry`` or ``aggregate_export`` when the ledger may cross a trust + boundary. An unrecognized value falls back to the local-only default rather + than failing a hook. + """ + + configured = os.environ.get("MARGINAL_PRIVACY_PROFILE", "") + if not configured: + return PrivacyProfile.LOCAL_FULL + try: + return PrivacyProfile(configured.strip().casefold()) + except ValueError: + return PrivacyProfile.LOCAL_FULL + + +def _privacy_key_path(data_root: Path, profile: PrivacyProfile) -> Path | None: + if profile is not PrivacyProfile.SAFE_TELEMETRY: + return None + return data_root / "privacy" / "pseudonym.key" + + +def _build_session( + event: SessionEvent, *, data_root: Path +) -> tuple[ObserveSession, HookSessionRuntime]: + profile = selected_privacy_profile() + observe = build_observe_session( + engine=ENGINE, + session_id=event.session_id, + workspace=event.cwd, + data_root=data_root, + privacy_profile=profile, + privacy_key_path=_privacy_key_path(data_root, profile), + ) + observe.record( + { + "event": "hook_session_start", + "source": event.source, + "capability_level": observe.runtime.capabilities.level, + "privacy_profile_selected": selected_privacy_profile().value, + } + ) + return observe, HookSessionRuntime(observe.runtime, workspace=event.cwd) + + +def _serve_bootstrap(path: Path) -> int: + try: + payload = json.loads(path.read_text(encoding="utf-8")) + finally: + path.unlink(missing_ok=True) + event = parse_hook_event(payload["event"]) + if not isinstance(event, SessionEvent) or event.hook_event_name != "SessionStart": + return 2 + data_root = Path(payload["data_root"]).resolve() + observe, session = _build_session(event, data_root=data_root) + shutdown_event = threading.Event() + server = SessionServer( + data_root=data_root, + session_id=event.session_id, + token=str(payload["token"]), + handler=_handler(session, observe=observe, shutdown_event=shutdown_event), + ) + server.start() + shutdown_event.wait() + server.stop() + return 0 + + +def start_session_service(event: SessionEvent, *, data_root: str | Path) -> ConnectionInfo: + """Start an in-process session service. Used by tests and embedding callers.""" + + if event.hook_event_name != "SessionStart": + raise ValueError("start_session_service requires SessionStart") + root = Path(data_root).resolve() + key = (root, event.session_id) + active = _SERVERS.get(key) + if active is not None: + if request_session(active[0].connection, operation="status", payload={}).get("ok") is True: + return active[0].connection + _SERVERS.pop(key, None) + observe, session = _build_session(event, data_root=root) + server = SessionServer( + data_root=root, + session_id=event.session_id, + token=secrets.token_hex(32), + handler=_handler(session, observe=observe), + ) + connection = server.start() + _SERVERS[key] = (server, session) + return connection + + +def stop_session_service(session_id: str, *, data_root: str | Path) -> None: + """Close one session, whether it runs in this process or in a spawned service.""" + + root = Path(data_root).resolve() + key = (root, session_id) + active = _SERVERS.pop(key, None) + if active is not None: + server, session = active + try: + request_session(server.connection, operation="close", payload={}) + finally: + session.close() + server.stop() + return + connection_path = _connection_path(root, session_id) + if not connection_path.exists(): + return + try: + connection = ConnectionInfo.from_file(connection_path) + except (OSError, ValueError, KeyError): + connection_path.unlink(missing_ok=True) + return + request_session(connection, operation="close", payload={}) + + +def run_hook(payload: dict[str, Any], *, data_root: str | Path) -> HookResult: + """Execute one hook. Every integration fault fails open.""" + + root = Path(data_root).resolve() + event: ClaudeCodeHookEvent | None = None + try: + event = parse_hook_event(payload) + if isinstance(event, SessionEvent): + if event.hook_event_name == "SessionStart": + _spawn_session_service(event, data_root=root) + else: + stop_session_service(event.session_id, data_root=root) + return HookResult(exit_code=0) + + connection_path = _connection_path(root, event.session_id) + if not connection_path.exists(): + return HookResult(exit_code=0, warning_code="SERVICE_UNAVAILABLE") + connection = ConnectionInfo.from_file(connection_path) + operation = "pre" if isinstance(event, PreToolUseEvent) else "post" + response = request_session(connection, operation=operation, payload=payload) + if response.get("ok") is not True: + return HookResult( + exit_code=0, + warning_code=str(response.get("error_code", "SERVICE_ERROR")), + ) + result = response.get("result") + return HookResult(exit_code=0, output=result if isinstance(result, dict) else None) + except Exception: + return HookResult(exit_code=0, warning_code="INTEGRATION_ERROR") + + +def observe_outcome(payload: dict[str, Any]) -> ActionOutcomeStatus: + """Classify one completion payload without touching session state.""" + + event = parse_hook_event(payload) + if not isinstance(event, (PostToolUseEvent, PostToolUseFailureEvent)): + raise ValueError("payload must be a PostToolUse or PostToolUseFailure event") + return tool_call_end(event).outcome + + +def hook_main(argv: list[str] | None = None) -> int: + """Entry point used by the plugin hook shim and by ``python -m``.""" + + selected = list(sys.argv[1:] if argv is None else argv) + if len(selected) == 2 and selected[0] == "--serve": + return _serve_bootstrap(Path(selected[1]).resolve()) + data_root = data_root_from_environment() + if not data_root: + return 0 + try: + payload = json.load(sys.stdin) + except (json.JSONDecodeError, UnicodeDecodeError, OSError): + return 0 + if not isinstance(payload, dict): + return 0 + result = run_hook(payload, data_root=data_root) + if result.output is not None: + print(json.dumps(result.output, sort_keys=True, separators=(",", ":"))) + return result.exit_code + + +if __name__ == "__main__": + raise SystemExit(hook_main()) diff --git a/src/marginal/integrations/codex/transport.py b/src/marginal/integrations/codex/transport.py index e03d6ec..7fe04de 100644 --- a/src/marginal/integrations/codex/transport.py +++ b/src/marginal/integrations/codex/transport.py @@ -1,221 +1,26 @@ -"""Authenticated, bounded loopback transport for one Codex session.""" +"""Compatibility re-export of the engine-neutral session transport. -from __future__ import annotations - -import hashlib -import hmac -import json -import os -import socket -import socketserver -import threading -from collections.abc import Callable, Mapping -from dataclasses import asdict, dataclass -from pathlib import Path -from typing import Any - -MAX_MESSAGE_BYTES = 256 * 1024 - - -def connection_filename(session_id: str) -> str: - """Return a stable receipt name without exposing the raw Codex session identity.""" - - if not isinstance(session_id, str) or not session_id: - raise ValueError("session_id must be a non-empty string") - digest = hashlib.sha256(session_id.encode("utf-8")).hexdigest() - return f"{digest}.json" - - -@dataclass(frozen=True, slots=True) -class ConnectionInfo: - session_id: str - host: str - port: int - token: str - pid: int - connection_file: Path - - def to_dict(self) -> dict[str, Any]: - payload = asdict(self) - payload["connection_file"] = str(self.connection_file) - return payload - - @classmethod - def from_file(cls, path: str | Path) -> ConnectionInfo: - source = Path(path).resolve() - payload = json.loads(source.read_text(encoding="utf-8")) - return cls( - session_id=str(payload["session_id"]), - host=str(payload["host"]), - port=int(payload["port"]), - token=str(payload["token"]), - pid=int(payload["pid"]), - connection_file=source, - ) - - -SessionHandler = Callable[[str, Mapping[str, Any]], Mapping[str, Any] | None] - - -def _response_bytes(payload: Mapping[str, Any]) -> bytes: - return ( - json.dumps( - dict(payload), - sort_keys=True, - separators=(",", ":"), - ensure_ascii=False, - allow_nan=False, - ).encode("utf-8") - + b"\n" - ) +The transport carries no Codex-specific behavior, so it now lives in +``marginal.integrations.transport`` where every hook-based adapter can share it. +This module remains a stable import path for existing callers. +""" +from __future__ import annotations -class _BoundedRequestHandler(socketserver.StreamRequestHandler): - def handle(self) -> None: - owner: _LoopbackServer = self.server # type: ignore[assignment] - self.connection.settimeout(5.0) - raw = self.rfile.readline(MAX_MESSAGE_BYTES + 2) - if len(raw) > MAX_MESSAGE_BYTES + 1: - self.wfile.write(_response_bytes(_error("MESSAGE_TOO_LARGE"))) - return - try: - request = json.loads(raw) - except (json.JSONDecodeError, UnicodeDecodeError): - self.wfile.write(_response_bytes(_error("INVALID_MESSAGE"))) - return - if not isinstance(request, dict): - self.wfile.write(_response_bytes(_error("INVALID_MESSAGE"))) - return - supplied_token = request.get("token") - if not isinstance(supplied_token, str) or not hmac.compare_digest( - supplied_token, owner.token - ): - self.wfile.write(_response_bytes(_error("AUTH_FAILED"))) - return - operation = request.get("operation") - payload = request.get("payload") - if not isinstance(operation, str) or not isinstance(payload, dict): - self.wfile.write(_response_bytes(_error("INVALID_MESSAGE"))) - return - try: - result = owner.callback(operation, payload) - response: Mapping[str, Any] = {"ok": True, "result": result} - except Exception: - response = _error("SERVICE_ERROR") - self.wfile.write(_response_bytes(response)) - - -class _LoopbackServer(socketserver.ThreadingTCPServer): - allow_reuse_address = False - daemon_threads = True - - def __init__(self, token: str, callback: SessionHandler) -> None: - self.token = token - self.callback = callback - super().__init__(("127.0.0.1", 0), _BoundedRequestHandler) - - -def _error(code: str) -> dict[str, Any]: - return {"ok": False, "error_code": code} - - -class SessionServer: - """Own one authenticated server and its user-private connection receipt.""" - - def __init__( - self, - *, - data_root: str | Path, - session_id: str, - token: str, - handler: SessionHandler, - ) -> None: - if len(token.encode("utf-8")) < 16: - raise ValueError("session token must contain at least 128 bits") - self.data_root = Path(data_root).resolve() - self.data_root.mkdir(parents=True, exist_ok=True, mode=0o700) - self.sessions_root = self.data_root / "sessions" - self.sessions_root.mkdir(parents=True, exist_ok=True, mode=0o700) - if os.name == "posix": - self.data_root.chmod(0o700) - self.sessions_root.chmod(0o700) - self._server = _LoopbackServer(token, handler) - self._thread: threading.Thread | None = None - connection_path = self.sessions_root / connection_filename(session_id) - self.connection = ConnectionInfo( - session_id=session_id, - host="127.0.0.1", - port=int(self._server.server_address[1]), - token=token, - pid=os.getpid(), - connection_file=connection_path, - ) - - def start(self) -> ConnectionInfo: - if self._thread is not None: - return self.connection - descriptor = os.open( - self.connection.connection_file, - os.O_CREAT | os.O_TRUNC | os.O_WRONLY, - 0o600, - ) - try: - os.write(descriptor, _response_bytes(self.connection.to_dict())) - os.fsync(descriptor) - finally: - os.close(descriptor) - if os.name == "posix": - self.connection.connection_file.chmod(0o600) - self._thread = threading.Thread( - target=self._server.serve_forever, - name=f"marginal-{self.connection.session_id}", - daemon=True, - ) - self._thread.start() - return self.connection - - def stop(self) -> None: - if self._thread is not None: - self._server.shutdown() - self._thread.join(timeout=5) - self._thread = None - self._server.server_close() - self.connection.connection_file.unlink(missing_ok=True) - - -def request_session( - connection: ConnectionInfo, - *, - operation: str, - payload: Mapping[str, Any], - token: str | None = None, - timeout: float = 5.0, -) -> dict[str, Any]: - """Send one bounded request; transport errors are returned as stable error codes.""" - - request = _response_bytes( - { - "token": connection.token if token is None else token, - "operation": operation, - "payload": dict(payload), - } - ) - if len(request) > MAX_MESSAGE_BYTES + 1: - return _error("MESSAGE_TOO_LARGE") - try: - with socket.create_connection((connection.host, connection.port), timeout=timeout) as sock: - sock.settimeout(timeout) - sock.sendall(request) - reader = sock.makefile("rb") - raw = reader.readline(MAX_MESSAGE_BYTES + 2) - except (OSError, TimeoutError): - return _error("SERVICE_UNAVAILABLE") - if len(raw) > MAX_MESSAGE_BYTES + 1: - return _error("MESSAGE_TOO_LARGE") - try: - response = json.loads(raw) - except (json.JSONDecodeError, UnicodeDecodeError): - return _error("INVALID_RESPONSE") - if not isinstance(response, dict): - return _error("INVALID_RESPONSE") - return response +from ..transport import ( + MAX_MESSAGE_BYTES, + ConnectionInfo, + SessionHandler, + SessionServer, + connection_filename, + request_session, +) + +__all__ = [ + "MAX_MESSAGE_BYTES", + "ConnectionInfo", + "SessionHandler", + "SessionServer", + "connection_filename", + "request_session", +] diff --git a/src/marginal/integrations/hookkit/__init__.py b/src/marginal/integrations/hookkit/__init__.py new file mode 100644 index 0000000..ad2d8a9 --- /dev/null +++ b/src/marginal/integrations/hookkit/__init__.py @@ -0,0 +1,44 @@ +"""Shared building blocks for hook-based engine adapters. + +``hookkit`` holds the parts of a hook integration that are genuinely engine +independent: normalized events, privacy-safe action normalization, conservative +outcome classification, workspace state evidence, and session correlation. + +Each engine adapter keeps its own parser for its own native payload and declares +only what that engine documents. The Codex integration predates this module and +still carries its own copies; migrating it is deliberately separate work so that +the validated Codex path is not disturbed by a new adapter. +""" + +from .bootstrap import ( + OBSERVE_CAPABILITIES, + ObserveSession, + build_observe_session, + session_hash, + workspace_hash, +) +from .events import HookEvent, SessionBoundary, ToolCallEnd, ToolCallStart +from .normalization import classify_action, normalize_tool_call, semantic_key +from .outcomes import classify_structured_result, completion_evidence_hash +from .session import HookIntegrationError, HookSessionRuntime +from .state import workspace_state_hash + +__all__ = [ + "OBSERVE_CAPABILITIES", + "HookEvent", + "HookIntegrationError", + "HookSessionRuntime", + "ObserveSession", + "SessionBoundary", + "ToolCallEnd", + "ToolCallStart", + "build_observe_session", + "classify_action", + "classify_structured_result", + "completion_evidence_hash", + "normalize_tool_call", + "semantic_key", + "session_hash", + "workspace_hash", + "workspace_state_hash", +] diff --git a/src/marginal/integrations/hookkit/bootstrap.py b/src/marginal/integrations/hookkit/bootstrap.py new file mode 100644 index 0000000..a0fbc8f --- /dev/null +++ b/src/marginal/integrations/hookkit/bootstrap.py @@ -0,0 +1,109 @@ +"""Construct the Shadow Mode runtime one hook session governs itself with. + +Adapters built on ``hookkit`` observe. They declare no control capability, so the +core refuses to run them in a blocking mode, and the Decision Ledger is the only +place their evidence lands. +""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass +from pathlib import Path + +from marginal import BudgetLimits, Treasury +from marginal.ledger import DecisionLedgerContext, JsonlDecisionLedger +from marginal.privacy import PrivacyProfile +from marginal.protocol import AgentCapabilities +from marginal.runtime import UniversalRuntime + +OBSERVE_CAPABILITIES = AgentCapabilities() +"""Honest capability declaration for a hook adapter that only observes. + +These hook surfaces expose no per-tool token usage, no model-turn control, and no +verifier outcomes, so every capability flag stays false and +``AgentCapabilities.level`` reports ``observe``. +""" + + +def session_hash(session_id: str) -> str: + """Return a stable pseudonym for one engine session identifier.""" + + if not isinstance(session_id, str) or not session_id: + raise ValueError("session_id must be a non-empty string") + return hashlib.sha256(session_id.encode("utf-8")).hexdigest() + + +def workspace_hash(workspace: str | Path) -> str: + """Return a stable pseudonym for one workspace path.""" + + resolved = str(Path(workspace)) + return hashlib.sha256(resolved.encode("utf-8")).hexdigest() + + +@dataclass(frozen=True, slots=True) +class ObserveSession: + """A Shadow Mode runtime plus the ledger its evidence is written to.""" + + runtime: UniversalRuntime + ledger: JsonlDecisionLedger + ledger_path: Path + session_hash: str + workspace_hash: str + + def record(self, event: dict[str, object]) -> None: + """Append one governance evidence record, failing open on ledger errors.""" + + try: + self.ledger.emit(event) + except Exception: # pragma: no cover - evidence must never break a hook + return + + +def build_observe_session( + *, + engine: str, + session_id: str, + workspace: str | Path, + data_root: str | Path, + privacy_profile: PrivacyProfile | str = PrivacyProfile.LOCAL_FULL, + privacy_key_path: str | Path | None = None, +) -> ObserveSession: + """Build a Shadow Mode ``UniversalRuntime`` writing to a local Decision Ledger. + + The default profile is ``LOCAL_FULL`` because the ledger stays inside the + engine's own plugin data directory on the user's machine. Choose + ``SAFE_TELEMETRY`` when the ledger may cross a trust boundary. + """ + + if not isinstance(engine, str) or not engine.strip(): + raise ValueError("engine must be a non-empty string") + pseudonym = session_hash(session_id) + task_id = workspace_hash(workspace) + root = Path(data_root).resolve() + ledger_path = root / "ledger" / task_id / f"{pseudonym}.jsonl" + ledger = JsonlDecisionLedger( + ledger_path, + context=DecisionLedgerContext( + run_id=pseudonym, + task_id=task_id, + engine=engine, + ), + privacy_profile=privacy_profile, + privacy_key_path=privacy_key_path, + ) + treasury = Treasury(BudgetLimits(), trace_sink=ledger, mode="shadow") + runtime = UniversalRuntime( + treasury, + engine=engine, + session_id=session_id, + task_id=task_id, + capabilities=OBSERVE_CAPABILITIES, + ) + return ObserveSession( + runtime=runtime, + ledger=ledger, + ledger_path=ledger_path, + session_hash=pseudonym, + workspace_hash=task_id, + ) diff --git a/src/marginal/integrations/hookkit/events.py b/src/marginal/integrations/hookkit/events.py new file mode 100644 index 0000000..f911ab1 --- /dev/null +++ b/src/marginal/integrations/hookkit/events.py @@ -0,0 +1,114 @@ +"""Engine-neutral value objects for hook-based agent integrations. + +Every supported engine reports a slightly different hook payload. An adapter is +responsible for parsing its own native payload and for declaring what that engine +can actually prove; ``hookkit`` never guesses. These objects carry only the +minimum an adapter must supply for lifecycle correlation and repetition control. +""" + +from __future__ import annotations + +import math +from collections.abc import Mapping +from dataclasses import dataclass, field +from typing import Any + +from marginal.controls import ActionOutcomeStatus + + +def _required_text(value: Any, name: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{name} must be a non-empty string") + return value + + +def _optional_text(value: Any, name: str) -> str: + if value is None: + return "" + if not isinstance(value, str): + raise TypeError(f"{name} must be a string or None") + return value + + +@dataclass(frozen=True, slots=True) +class SessionBoundary: + """Start or end of one engine session.""" + + engine: str + session_id: str + workspace: str + kind: str + source: str = "" + reason: str = "" + + def __post_init__(self) -> None: + for name in ("engine", "session_id", "workspace", "kind"): + _required_text(getattr(self, name), name) + if self.kind not in {"start", "end"}: + raise ValueError("kind must be 'start' or 'end'") + object.__setattr__(self, "source", _optional_text(self.source, "source")) + object.__setattr__(self, "reason", _optional_text(self.reason, "reason")) + + +@dataclass(frozen=True, slots=True) +class ToolCallStart: + """A tool call an engine proposes to execute. + + ``turn_id`` is optional because engines disagree about turn identity. It is + recorded when available and never invented. + """ + + session_id: str + call_id: str + tool_name: str + tool_input: Mapping[str, Any] = field(default_factory=dict) + turn_id: str = "" + + def __post_init__(self) -> None: + for name in ("session_id", "call_id", "tool_name"): + _required_text(getattr(self, name), name) + if not isinstance(self.tool_input, Mapping): + raise TypeError("tool_input must be a mapping") + object.__setattr__(self, "tool_input", dict(self.tool_input)) + object.__setattr__(self, "turn_id", _optional_text(self.turn_id, "turn_id")) + + +@dataclass(frozen=True, slots=True) +class ToolCallEnd: + """A completed tool call with the outcome its engine can actually prove. + + ``outcome`` must be ``UNKNOWN`` unless the engine documents the signal the + adapter derived it from. ``evidence`` is any JSON-compatible completion + evidence; only its hash is retained. ``duration_ms`` is engine-measured + latency when the engine reports it. + """ + + session_id: str + call_id: str + tool_name: str + outcome: ActionOutcomeStatus + tool_input: Mapping[str, Any] = field(default_factory=dict) + evidence: Any = None + duration_ms: float | None = None + turn_id: str = "" + + def __post_init__(self) -> None: + for name in ("session_id", "call_id", "tool_name"): + _required_text(getattr(self, name), name) + object.__setattr__(self, "outcome", ActionOutcomeStatus.parse(self.outcome)) + if not isinstance(self.tool_input, Mapping): + raise TypeError("tool_input must be a mapping") + object.__setattr__(self, "tool_input", dict(self.tool_input)) + object.__setattr__(self, "turn_id", _optional_text(self.turn_id, "turn_id")) + duration = self.duration_ms + if duration is None: + return + if isinstance(duration, bool) or not isinstance(duration, (int, float)): + raise TypeError("duration_ms must be a number or None") + value = float(duration) + if not math.isfinite(value) or value < 0.0: + raise ValueError("duration_ms must be finite and non-negative") + object.__setattr__(self, "duration_ms", value) + + +HookEvent = SessionBoundary | ToolCallStart | ToolCallEnd diff --git a/src/marginal/integrations/hookkit/normalization.py b/src/marginal/integrations/hookkit/normalization.py new file mode 100644 index 0000000..b1db96e --- /dev/null +++ b/src/marginal/integrations/hookkit/normalization.py @@ -0,0 +1,128 @@ +"""Convert engine tool calls into privacy-safe protocol actions. + +Normalization keeps raw arguments out of the action. Only a semantic hash of the +tool identity and its arguments is retained, so repetition can be detected without +persisting file paths, commands, or prompt material. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from typing import Any + +from marginal.models import Cost +from marginal.protocol import AgentAction, DeduplicationScope + +from .events import ToolCallStart + +_VERIFICATION_PATTERN = re.compile( + r"(?:^|[\s/])(?:pytest|tox|nox|unittest|jest|vitest|mocha|rspec|" + r"go\s+test|cargo\s+test|npm\s+(?:run\s+)?test|pnpm\s+(?:run\s+)?test|" + r"yarn\s+test|ruff|mypy|pyright|eslint|tsc|git\s+diff\s+--check)(?:\s|$)", + re.IGNORECASE, +) + +_SHELL_TOOLS = frozenset({"bash", "shell", "exec", "exec_command", "terminal", "run"}) +_EDIT_TOOLS = frozenset( + {"apply_patch", "edit", "multiedit", "patch", "write", "write_file", "notebookedit"} +) +_READ_TOOLS = frozenset({"read", "read_file", "glob", "grep", "search", "list", "ls"}) +_FETCH_TOOLS = frozenset({"webfetch", "websearch", "fetch", "webread"}) + +_COMMAND_KEYS = ("command", "cmd", "script") + + +def _canonical_json(value: Any) -> str: + try: + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + allow_nan=False, + ) + except (TypeError, ValueError) as exc: + raise ValueError("tool_input must have a canonical JSON representation") from exc + + +def semantic_key(start: ToolCallStart) -> str: + """Hash the tool identity and arguments that define one semantic action.""" + + canonical = _canonical_json( + {"tool_name": start.tool_name.casefold(), "tool_input": dict(start.tool_input)} + ) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +def _command(start: ToolCallStart) -> str: + for key in _COMMAND_KEYS: + value = start.tool_input.get(key) + if isinstance(value, str): + return value + return "" + + +def classify_action(start: ToolCallStart) -> tuple[str, bool]: + """Return a generic action kind and whether the call reads as verification.""" + + tool = start.tool_name.casefold() + command = _command(start) + if command and _VERIFICATION_PATTERN.search(command): + return "verification", True + if tool in _SHELL_TOOLS: + return "shell", False + if tool in _EDIT_TOOLS: + return "edit", False + if tool in _READ_TOOLS: + return "read", False + if tool in _FETCH_TOOLS: + return "fetch", False + if tool.startswith("mcp"): + return "mcp", False + return "tool", False + + +def normalize_tool_call( + start: ToolCallStart, + *, + engine: str, + state_hash: str, + previous_evidence_hash: str = "", +) -> AgentAction: + """Return a normalized action that retains no raw arguments. + + ``state_hash`` may be empty when the workspace is not observable. Empty state + makes every repetition control fail open rather than invent certainty. + """ + + if not isinstance(start, ToolCallStart): + raise TypeError("start must be a ToolCallStart") + if not isinstance(engine, str) or not engine.strip(): + raise ValueError("engine must be a non-empty string") + if not isinstance(state_hash, str): + raise TypeError("state_hash must be a string") + if not isinstance(previous_evidence_hash, str): + raise TypeError("previous_evidence_hash must be a string") + + kind, is_verification = classify_action(start) + metadata = { + "session_id": start.session_id, + "turn_id": start.turn_id, + "tool_name": start.tool_name, + "state_hash": state_hash, + "evidence_hash": previous_evidence_hash, + "semantic_key": semantic_key(start), + } + return AgentAction( + action_id=start.call_id, + name=f"{engine} {start.tool_name} action", + kind=kind, + estimated_cost=Cost(), + is_verification=is_verification, + state_hash=state_hash, + phase=f"{engine}-tool-use", + deduplication_scope=DeduplicationScope.ONCE_PER_STATE, + metadata=metadata, + ) diff --git a/src/marginal/integrations/hookkit/outcomes.py b/src/marginal/integrations/hookkit/outcomes.py new file mode 100644 index 0000000..4f43e7a --- /dev/null +++ b/src/marginal/integrations/hookkit/outcomes.py @@ -0,0 +1,100 @@ +"""Conservative outcome classification for structured tool results. + +An engine that reports a structured result can prove success or failure. Anything +else is ``UNKNOWN``: human-readable output text is evidence of completion, never +evidence of success. +""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Mapping +from typing import Any + +from marginal.controls import ActionOutcomeStatus + +_SUCCESS_TEXT = frozenset({"success", "succeeded", "passed", "completed"}) +_FAILURE_TEXT = frozenset({"failure", "failed", "error", "errored"}) +_EXIT_KEYS = ("exit_code", "exitCode", "exit", "returncode", "status_code") +_NESTED_KEYS = ("metadata", "result", "state") + + +def _exit_signal(response: Mapping[str, Any]) -> set[ActionOutcomeStatus]: + signals: set[ActionOutcomeStatus] = set() + for key in _EXIT_KEYS: + value = response.get(key) + if isinstance(value, bool) or not isinstance(value, int): + continue + signals.add(ActionOutcomeStatus.SUCCESS if value == 0 else ActionOutcomeStatus.FAILURE) + return signals + + +def _boolean_signal(response: Mapping[str, Any]) -> set[ActionOutcomeStatus]: + signals: set[ActionOutcomeStatus] = set() + success = response.get("success") + if isinstance(success, bool): + signals.add(ActionOutcomeStatus.SUCCESS if success else ActionOutcomeStatus.FAILURE) + for key in ("is_error", "isError", "error"): + value = response.get(key) + if isinstance(value, bool): + signals.add(ActionOutcomeStatus.FAILURE if value else ActionOutcomeStatus.SUCCESS) + return signals + + +def _text_signal(response: Mapping[str, Any]) -> set[ActionOutcomeStatus]: + signals: set[ActionOutcomeStatus] = set() + for key in ("status", "outcome"): + value = response.get(key) + if not isinstance(value, str): + continue + normalized = value.strip().casefold() + if normalized in _SUCCESS_TEXT: + signals.add(ActionOutcomeStatus.SUCCESS) + elif normalized in _FAILURE_TEXT: + signals.add(ActionOutcomeStatus.FAILURE) + return signals + + +def _signals(response: Mapping[str, Any], *, depth: int) -> set[ActionOutcomeStatus]: + signals = _exit_signal(response) | _boolean_signal(response) | _text_signal(response) + if depth <= 0: + return signals + for key in _NESTED_KEYS: + nested = response.get(key) + if isinstance(nested, Mapping): + signals |= _signals(nested, depth=depth - 1) + return signals + + +def classify_structured_result(result: Any) -> ActionOutcomeStatus: + """Classify only explicit, mutually consistent structured outcome signals. + + Conflicting signals resolve to ``UNKNOWN`` because a governor must not pick a + winner between two claims the engine itself did not reconcile. + """ + + if not isinstance(result, Mapping): + return ActionOutcomeStatus.UNKNOWN + signals = _signals(result, depth=2) + if len(signals) != 1: + return ActionOutcomeStatus.UNKNOWN + return next(iter(signals)) + + +def completion_evidence_hash(evidence: Any) -> str: + """Hash JSON-compatible completion evidence without retaining its contents.""" + + if evidence is None: + return "" + try: + canonical = json.dumps( + evidence, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + allow_nan=False, + ) + except (TypeError, ValueError): + return "" + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() diff --git a/src/marginal/integrations/hookkit/session.py b/src/marginal/integrations/hookkit/session.py new file mode 100644 index 0000000..fa10189 --- /dev/null +++ b/src/marginal/integrations/hookkit/session.py @@ -0,0 +1,204 @@ +"""Engine-neutral session lifecycle for hook-based adapters. + +The session runtime correlates a proposed tool call with its completion, settles +each accepted action exactly once, and records repetition evidence. It contains no +economic policy: every decision comes from the shared core through +``UniversalRuntime``. + +This runtime observes. It never converts a stop recommendation into a denial. An +engine that can block still needs an evidence gate before enforcement, and no such +gate has been earned for the adapters built on this module. +""" + +from __future__ import annotations + +import hashlib +from pathlib import Path + +from marginal.controls import ActionOutcomeStatus, NoProgressDetector, NoProgressSignal +from marginal.models import Cost +from marginal.protocol import AgentAction, AgentDecision +from marginal.runtime import UniversalRuntime + +from .events import ToolCallEnd, ToolCallStart +from .normalization import normalize_tool_call +from .outcomes import completion_evidence_hash +from .state import workspace_state_hash + + +class HookIntegrationError(RuntimeError): + """Raised when hook lifecycle identity is missing or inconsistent.""" + + +class HookSessionRuntime: + """Correlate one engine session's tool calls over the provider-neutral runtime.""" + + def __init__( + self, + runtime: UniversalRuntime, + *, + workspace: str | Path, + detector: NoProgressDetector | None = None, + ) -> None: + if not isinstance(runtime, UniversalRuntime): + raise TypeError("runtime must be a UniversalRuntime") + self.runtime = runtime + self.workspace = Path(workspace) + self.detector = detector or NoProgressDetector() + self._pending: dict[str, AgentAction] = {} + self._evidence_by_semantic_key: dict[str, str] = {} + self._last_signal: NoProgressSignal | None = None + self._last_action_evidence: dict[str, str] | None = None + self._successful = 0 + self._failed = 0 + self._unknown = 0 + self._unmatched = 0 + self._recommended_stops = 0 + self._closed = False + + @property + def engine(self) -> str: + return self.runtime.engine + + @property + def last_no_progress_signal(self) -> NoProgressSignal | None: + return self._last_signal + + @property + def last_action_evidence(self) -> dict[str, str] | None: + return dict(self._last_action_evidence) if self._last_action_evidence else None + + def tool_call_start(self, start: ToolCallStart) -> AgentDecision: + """Authorize one proposed tool call and record its repetition evidence.""" + + self._ensure_open() + self._validate_session(start.session_id) + if start.call_id in self._pending: + raise HookIntegrationError(f"tool identity is already pending: {start.call_id}") + + state_hash = workspace_state_hash(self.workspace) + action = normalize_tool_call(start, engine=self.engine, state_hash=state_hash) + key = str(action.metadata["semantic_key"]) + evidence_hash = self._evidence_by_semantic_key.get(key, "") + if evidence_hash: + action = normalize_tool_call( + start, + engine=self.engine, + state_hash=state_hash, + previous_evidence_hash=evidence_hash, + ) + self._last_action_evidence = self._safe_action_evidence(action) + self._last_signal = self.detector.evaluate(key, state_hash, evidence_hash) + if self._last_signal.should_recommend_stop: + self._recommended_stops += 1 + + decision = self.runtime.before_action(action) + if decision.allowed: + self._pending[start.call_id] = action + return decision + + def tool_call_end(self, end: ToolCallEnd) -> ActionOutcomeStatus: + """Settle one completed tool call using only the outcome its engine proved.""" + + self._ensure_open() + self._validate_session(end.session_id) + action = self._pending.pop(end.call_id, None) + if action is None: + # A completion without a recorded proposal means hook coverage was + # incomplete for this call. Report it instead of settling an action + # the ledger never authorized. + self._unmatched += 1 + return ActionOutcomeStatus.UNKNOWN + if str(action.metadata.get("tool_name", "")) != end.tool_name: + raise HookIntegrationError( + f"completion tool identity does not match the proposal: {end.call_id}" + ) + + actual_cost = self._actual_cost(end) + if end.outcome is ActionOutcomeStatus.SUCCESS: + self.runtime.after_action(end.call_id, actual_cost=actual_cost) + self._successful += 1 + elif end.outcome is ActionOutcomeStatus.FAILURE: + self.runtime.fail_action( + end.call_id, + reason=f"{self.engine} reported an explicit tool failure", + actual_cost=actual_cost, + ) + self._failed += 1 + else: + self.runtime.fail_action( + end.call_id, + reason=f"{self.engine} completion outcome was not observable", + actual_cost=actual_cost, + ) + self._unknown += 1 + + key = str(action.metadata["semantic_key"]) + evidence_hash = completion_evidence_hash(end.evidence) + self.detector.observe( + key, + workspace_state_hash(self.workspace), + evidence_hash, + end.outcome, + ) + if evidence_hash: + self._evidence_by_semantic_key[key] = evidence_hash + return end.outcome + + def pending_action_ids(self) -> tuple[str, ...]: + return tuple(sorted(self._pending)) + + def action_evidence(self, action_id: str) -> dict[str, str] | None: + action = self._pending.get(action_id) + return self._safe_action_evidence(action) if action is not None else None + + def summary(self) -> dict[str, int]: + return { + "successful_observations": self._successful, + "failed_observations": self._failed, + "unknown_observations": self._unknown, + "completed_observations": self._successful + self._failed + self._unknown, + "unmatched_completions": self._unmatched, + "recommended_stops": self._recommended_stops, + "pending_actions": len(self._pending), + "enforced_denials": 0, + } + + def close(self) -> None: + """Settle every proposal the engine never reported a completion for.""" + + if self._closed: + return + for action_id in tuple(self._pending): + self.runtime.fail_action( + action_id, + reason=f"{self.engine} session ended before the outcome was observable", + ) + self._unknown += 1 + self._pending.pop(action_id) + self._closed = True + + @staticmethod + def _actual_cost(end: ToolCallEnd) -> Cost | None: + """Return measured latency only. These hook surfaces expose no token usage.""" + + if end.duration_ms is None: + return None + return Cost(latency_ms=int(end.duration_ms)) + + @staticmethod + def _safe_action_evidence(action: AgentAction) -> dict[str, str]: + return { + "action_hash": hashlib.sha256(action.action_id.encode("utf-8")).hexdigest(), + "semantic_key": str(action.metadata.get("semantic_key", "")), + "state_hash": action.state_hash, + "evidence_hash": str(action.metadata.get("evidence_hash", "")), + } + + def _ensure_open(self) -> None: + if self._closed: + raise HookIntegrationError(f"{self.engine} session runtime is closed") + + def _validate_session(self, session_id: str) -> None: + if session_id != self.runtime.session_id: + raise HookIntegrationError("hook session identity does not match runtime identity") diff --git a/src/marginal/integrations/hookkit/state.py b/src/marginal/integrations/hookkit/state.py new file mode 100644 index 0000000..5acb3a7 --- /dev/null +++ b/src/marginal/integrations/hookkit/state.py @@ -0,0 +1,140 @@ +"""Privacy-safe workspace evidence for hook-based integrations. + +The hash covers tracked changes against ``HEAD`` plus safe untracked content, so a +repeated action can be recognized as acting on unchanged state. Only the digest is +retained; file names and contents are never persisted. + +Unlike the Codex integration, which requires the session directory to be a +repository root, this helper accepts any directory inside a Git work tree and +returns an empty string when state is not observable. An empty state hash makes +every repetition control fail open. +""" + +from __future__ import annotations + +import hashlib +import os +import subprocess +from pathlib import Path +from typing import Protocol + +_IGNORED_PARTS = frozenset( + { + ".claude", + ".codex", + ".git", + ".marginal", + ".mypy_cache", + ".opencode", + ".pytest_cache", + ".ruff_cache", + ".tox", + ".venv", + "__pycache__", + "node_modules", + } +) + +EMPTY_TREE_OBJECT = "4b825dc642cb6eb9a060e54bf8d69288fbee4904" +"""Git's constant empty tree object, used when a repository has no commits yet.""" + +_EXCLUSIONS = ( + ":(exclude).claude/**", + ":(exclude).codex/**", + ":(exclude).marginal/**", + ":(exclude).opencode/**", + ":(exclude).venv/**", + ":(exclude)**/__pycache__/**", +) + + +class _Hash(Protocol): + def update(self, data: bytes, /) -> object: ... + + def hexdigest(self) -> str: ... + + +class WorkspaceNotObservable(RuntimeError): + """Raised internally when Git cannot describe the workspace.""" + + +def _git(repo: Path, *args: str) -> bytes: + environment = { + "PATH": os.environ.get("PATH", ""), + "LANG": "C", + "LC_ALL": "C", + "GIT_CONFIG_NOSYSTEM": "1", + } + try: + completed = subprocess.run( + ["git", *args], + cwd=repo, + env=environment, + check=False, + capture_output=True, + ) + except OSError as exc: + raise WorkspaceNotObservable("git is not available") from exc + if completed.returncode != 0: + raise WorkspaceNotObservable("git could not describe the workspace") + return completed.stdout + + +def _ignored(path: str) -> bool: + return any(part in _IGNORED_PARTS for part in Path(path).parts) + + +def _update_untracked(digest: _Hash, repo: Path) -> None: + output = _git(repo, "ls-files", "--others", "--exclude-standard", "-z") + for raw_path in sorted(filter(None, output.split(b"\0"))): + path = raw_path.decode("utf-8", errors="surrogateescape") + if _ignored(path): + continue + target = repo / path + if not target.is_file(): + continue + try: + content = target.read_bytes() + except OSError: + continue + digest.update(b"untracked\0") + digest.update(raw_path) + digest.update(b"\0") + digest.update(content) + digest.update(b"\0") + + +def _base_revision(repo: Path) -> str: + """Return the revision to diff against, tolerating a repository with no commits. + + A freshly initialized repository has no ``HEAD``. Diffing against the empty tree + keeps such a workspace observable instead of discarding all state evidence for + the whole session. + """ + + try: + return _git(repo, "rev-parse", "HEAD").decode("utf-8").strip() + except WorkspaceNotObservable: + return EMPTY_TREE_OBJECT + + +def workspace_state_hash(workspace: str | Path) -> str: + """Return a workspace state digest, or an empty string when unobservable.""" + + try: + directory = Path(workspace).resolve() + except OSError: + return "" + if not directory.is_dir(): + return "" + try: + root = _git(directory, "rev-parse", "--show-toplevel").decode("utf-8").strip() + repo = Path(root).resolve() + digest = hashlib.sha256() + base = _base_revision(repo) + digest.update(base.encode("utf-8")) + digest.update(_git(repo, "diff", "--binary", base, "--", ".", *_EXCLUSIONS)) + _update_untracked(digest, repo) + except (WorkspaceNotObservable, OSError, UnicodeDecodeError): + return "" + return digest.hexdigest() diff --git a/src/marginal/integrations/transport.py b/src/marginal/integrations/transport.py new file mode 100644 index 0000000..e03d6ec --- /dev/null +++ b/src/marginal/integrations/transport.py @@ -0,0 +1,221 @@ +"""Authenticated, bounded loopback transport for one Codex session.""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import os +import socket +import socketserver +import threading +from collections.abc import Callable, Mapping +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + +MAX_MESSAGE_BYTES = 256 * 1024 + + +def connection_filename(session_id: str) -> str: + """Return a stable receipt name without exposing the raw Codex session identity.""" + + if not isinstance(session_id, str) or not session_id: + raise ValueError("session_id must be a non-empty string") + digest = hashlib.sha256(session_id.encode("utf-8")).hexdigest() + return f"{digest}.json" + + +@dataclass(frozen=True, slots=True) +class ConnectionInfo: + session_id: str + host: str + port: int + token: str + pid: int + connection_file: Path + + def to_dict(self) -> dict[str, Any]: + payload = asdict(self) + payload["connection_file"] = str(self.connection_file) + return payload + + @classmethod + def from_file(cls, path: str | Path) -> ConnectionInfo: + source = Path(path).resolve() + payload = json.loads(source.read_text(encoding="utf-8")) + return cls( + session_id=str(payload["session_id"]), + host=str(payload["host"]), + port=int(payload["port"]), + token=str(payload["token"]), + pid=int(payload["pid"]), + connection_file=source, + ) + + +SessionHandler = Callable[[str, Mapping[str, Any]], Mapping[str, Any] | None] + + +def _response_bytes(payload: Mapping[str, Any]) -> bytes: + return ( + json.dumps( + dict(payload), + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + allow_nan=False, + ).encode("utf-8") + + b"\n" + ) + + +class _BoundedRequestHandler(socketserver.StreamRequestHandler): + def handle(self) -> None: + owner: _LoopbackServer = self.server # type: ignore[assignment] + self.connection.settimeout(5.0) + raw = self.rfile.readline(MAX_MESSAGE_BYTES + 2) + if len(raw) > MAX_MESSAGE_BYTES + 1: + self.wfile.write(_response_bytes(_error("MESSAGE_TOO_LARGE"))) + return + try: + request = json.loads(raw) + except (json.JSONDecodeError, UnicodeDecodeError): + self.wfile.write(_response_bytes(_error("INVALID_MESSAGE"))) + return + if not isinstance(request, dict): + self.wfile.write(_response_bytes(_error("INVALID_MESSAGE"))) + return + supplied_token = request.get("token") + if not isinstance(supplied_token, str) or not hmac.compare_digest( + supplied_token, owner.token + ): + self.wfile.write(_response_bytes(_error("AUTH_FAILED"))) + return + operation = request.get("operation") + payload = request.get("payload") + if not isinstance(operation, str) or not isinstance(payload, dict): + self.wfile.write(_response_bytes(_error("INVALID_MESSAGE"))) + return + try: + result = owner.callback(operation, payload) + response: Mapping[str, Any] = {"ok": True, "result": result} + except Exception: + response = _error("SERVICE_ERROR") + self.wfile.write(_response_bytes(response)) + + +class _LoopbackServer(socketserver.ThreadingTCPServer): + allow_reuse_address = False + daemon_threads = True + + def __init__(self, token: str, callback: SessionHandler) -> None: + self.token = token + self.callback = callback + super().__init__(("127.0.0.1", 0), _BoundedRequestHandler) + + +def _error(code: str) -> dict[str, Any]: + return {"ok": False, "error_code": code} + + +class SessionServer: + """Own one authenticated server and its user-private connection receipt.""" + + def __init__( + self, + *, + data_root: str | Path, + session_id: str, + token: str, + handler: SessionHandler, + ) -> None: + if len(token.encode("utf-8")) < 16: + raise ValueError("session token must contain at least 128 bits") + self.data_root = Path(data_root).resolve() + self.data_root.mkdir(parents=True, exist_ok=True, mode=0o700) + self.sessions_root = self.data_root / "sessions" + self.sessions_root.mkdir(parents=True, exist_ok=True, mode=0o700) + if os.name == "posix": + self.data_root.chmod(0o700) + self.sessions_root.chmod(0o700) + self._server = _LoopbackServer(token, handler) + self._thread: threading.Thread | None = None + connection_path = self.sessions_root / connection_filename(session_id) + self.connection = ConnectionInfo( + session_id=session_id, + host="127.0.0.1", + port=int(self._server.server_address[1]), + token=token, + pid=os.getpid(), + connection_file=connection_path, + ) + + def start(self) -> ConnectionInfo: + if self._thread is not None: + return self.connection + descriptor = os.open( + self.connection.connection_file, + os.O_CREAT | os.O_TRUNC | os.O_WRONLY, + 0o600, + ) + try: + os.write(descriptor, _response_bytes(self.connection.to_dict())) + os.fsync(descriptor) + finally: + os.close(descriptor) + if os.name == "posix": + self.connection.connection_file.chmod(0o600) + self._thread = threading.Thread( + target=self._server.serve_forever, + name=f"marginal-{self.connection.session_id}", + daemon=True, + ) + self._thread.start() + return self.connection + + def stop(self) -> None: + if self._thread is not None: + self._server.shutdown() + self._thread.join(timeout=5) + self._thread = None + self._server.server_close() + self.connection.connection_file.unlink(missing_ok=True) + + +def request_session( + connection: ConnectionInfo, + *, + operation: str, + payload: Mapping[str, Any], + token: str | None = None, + timeout: float = 5.0, +) -> dict[str, Any]: + """Send one bounded request; transport errors are returned as stable error codes.""" + + request = _response_bytes( + { + "token": connection.token if token is None else token, + "operation": operation, + "payload": dict(payload), + } + ) + if len(request) > MAX_MESSAGE_BYTES + 1: + return _error("MESSAGE_TOO_LARGE") + try: + with socket.create_connection((connection.host, connection.port), timeout=timeout) as sock: + sock.settimeout(timeout) + sock.sendall(request) + reader = sock.makefile("rb") + raw = reader.readline(MAX_MESSAGE_BYTES + 2) + except (OSError, TimeoutError): + return _error("SERVICE_UNAVAILABLE") + if len(raw) > MAX_MESSAGE_BYTES + 1: + return _error("MESSAGE_TOO_LARGE") + try: + response = json.loads(raw) + except (json.JSONDecodeError, UnicodeDecodeError): + return _error("INVALID_RESPONSE") + if not isinstance(response, dict): + return _error("INVALID_RESPONSE") + return response diff --git a/tests/integrations/claude_code/__init__.py b/tests/integrations/claude_code/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/integrations/claude_code/conftest.py b/tests/integrations/claude_code/conftest.py new file mode 100644 index 0000000..69c7e2a --- /dev/null +++ b/tests/integrations/claude_code/conftest.py @@ -0,0 +1,148 @@ +"""Shared fixtures for the Claude Code integration tests. + +The payload shapes mirror Claude Code 2.1.233. Every identifier, path, and value is +synthetic. +""" + +import subprocess +from pathlib import Path +from typing import Any + +import pytest + +SESSION_ID = "synthetic-session-0001" +PROMPT_ID = "synthetic-prompt-0001" + + +def _git(repo: Path, *args: str) -> None: + subprocess.run( + ["git", *args], + cwd=repo, + check=True, + capture_output=True, + env={ + "PATH": "/usr/bin:/bin:/usr/local/bin", + "HOME": str(repo), + "GIT_AUTHOR_NAME": "Test", + "GIT_AUTHOR_EMAIL": "test@example.invalid", + "GIT_COMMITTER_NAME": "Test", + "GIT_COMMITTER_EMAIL": "test@example.invalid", + }, + ) + + +@pytest.fixture +def workspace(tmp_path: Path) -> Path: + """A committed Git work tree standing in for a Claude Code session directory.""" + + directory = tmp_path / "workspace" + directory.mkdir() + _git(directory, "init", "-q") + (directory / "example.txt").write_text("hello\n", encoding="utf-8") + _git(directory, "add", "example.txt") + _git(directory, "commit", "-qm", "initial") + return directory + + +@pytest.fixture +def payloads(workspace: Path) -> "PayloadFactory": + return PayloadFactory(workspace) + + +class PayloadFactory: + """Build hook payloads in the shape Claude Code sends them.""" + + def __init__(self, workspace: Path) -> None: + self.workspace = workspace + + def _common(self) -> dict[str, Any]: + return { + "session_id": SESSION_ID, + "cwd": str(self.workspace), + "transcript_path": str(self.workspace / "transcript.jsonl"), + } + + def session_start(self) -> dict[str, Any]: + return {**self._common(), "hook_event_name": "SessionStart", "source": "startup"} + + def session_end(self) -> dict[str, Any]: + return { + **self._common(), + "hook_event_name": "SessionEnd", + "prompt_id": PROMPT_ID, + "reason": "clear", + } + + def pre_tool_use( + self, + call_id: str = "toolu_synthetic0001", + *, + tool_name: str = "Read", + tool_input: dict[str, Any] | None = None, + ) -> dict[str, Any]: + return { + **self._common(), + "hook_event_name": "PreToolUse", + "permission_mode": "default", + "prompt_id": PROMPT_ID, + "tool_name": tool_name, + "tool_use_id": call_id, + "tool_input": tool_input or {"file_path": str(self.workspace / "example.txt")}, + } + + def post_tool_use( + self, + call_id: str = "toolu_synthetic0001", + *, + tool_name: str = "Read", + tool_input: dict[str, Any] | None = None, + tool_response: Any = None, + duration_ms: float | None = 11.0, + ) -> dict[str, Any]: + if tool_response is None: + tool_response = { + "type": "text", + "file": { + "filePath": str(self.workspace / "example.txt"), + "content": "hello\n", + "numLines": 1, + "startLine": 1, + "totalLines": 1, + }, + } + payload = { + **self._common(), + "hook_event_name": "PostToolUse", + "permission_mode": "default", + "prompt_id": PROMPT_ID, + "tool_name": tool_name, + "tool_use_id": call_id, + "tool_input": tool_input or {"file_path": str(self.workspace / "example.txt")}, + "tool_response": tool_response, + } + if duration_ms is not None: + payload["duration_ms"] = duration_ms + return payload + + def post_tool_use_failure( + self, + call_id: str = "toolu_synthetic0002", + *, + tool_name: str = "Bash", + tool_input: dict[str, Any] | None = None, + error: str = "Exit code 3", + is_interrupt: bool = False, + duration_ms: float | None = 4.0, + ) -> dict[str, Any]: + return { + **self._common(), + "hook_event_name": "PostToolUseFailure", + "permission_mode": "default", + "prompt_id": PROMPT_ID, + "tool_name": tool_name, + "tool_use_id": call_id, + "tool_input": tool_input or {"command": "exit 3", "description": "Run exit 3"}, + "error": error, + "is_interrupt": is_interrupt, + "duration_ms": duration_ms, + } diff --git a/tests/integrations/claude_code/test_events.py b/tests/integrations/claude_code/test_events.py new file mode 100644 index 0000000..d3b2a50 --- /dev/null +++ b/tests/integrations/claude_code/test_events.py @@ -0,0 +1,137 @@ +import pytest + +from marginal.integrations.claude_code.events import ( + PostToolUseEvent, + PostToolUseFailureEvent, + PreToolUseEvent, + SessionEvent, + parse_hook_event, +) + +from .conftest import PayloadFactory + + +def test_session_start_is_parsed_without_transcript_material( + payloads: PayloadFactory, +) -> None: + event = parse_hook_event(payloads.session_start()) + assert isinstance(event, SessionEvent) + assert event.hook_event_name == "SessionStart" + assert event.source == "startup" + assert not hasattr(event, "transcript_path") + + +def test_session_end_carries_its_reason(payloads: PayloadFactory) -> None: + event = parse_hook_event(payloads.session_end()) + assert isinstance(event, SessionEvent) + assert event.reason == "clear" + + +def test_pre_tool_use_is_parsed(payloads: PayloadFactory) -> None: + event = parse_hook_event(payloads.pre_tool_use()) + assert isinstance(event, PreToolUseEvent) + assert event.tool_name == "Read" + assert event.tool_use_id == "toolu_synthetic0001" + assert event.prompt_id == "synthetic-prompt-0001" + assert event.permission_mode == "default" + + +def test_pre_tool_use_tolerates_a_missing_permission_mode( + payloads: PayloadFactory, +) -> None: + payload = payloads.pre_tool_use() + del payload["permission_mode"] + event = parse_hook_event(payload) + assert isinstance(event, PreToolUseEvent) + assert event.permission_mode == "" + + +def test_post_tool_use_is_parsed_with_measured_latency(payloads: PayloadFactory) -> None: + event = parse_hook_event(payloads.post_tool_use()) + assert isinstance(event, PostToolUseEvent) + assert event.duration_ms == 11.0 + assert event.tool_response["type"] == "text" + + +def test_post_tool_use_accepts_the_documented_output_key(payloads: PayloadFactory) -> None: + payload = payloads.post_tool_use() + payload["tool_output"] = payload.pop("tool_response") + event = parse_hook_event(payload) + assert isinstance(event, PostToolUseEvent) + assert event.tool_response["type"] == "text" + + +def test_post_tool_use_requires_a_result(payloads: PayloadFactory) -> None: + payload = payloads.post_tool_use() + del payload["tool_response"] + with pytest.raises(ValueError): + parse_hook_event(payload) + + +def test_post_tool_use_without_latency_reports_none(payloads: PayloadFactory) -> None: + event = parse_hook_event(payloads.post_tool_use(duration_ms=None)) + assert isinstance(event, PostToolUseEvent) + assert event.duration_ms is None + + +def test_post_tool_use_failure_is_parsed(payloads: PayloadFactory) -> None: + event = parse_hook_event(payloads.post_tool_use_failure()) + assert isinstance(event, PostToolUseFailureEvent) + assert event.error == "Exit code 3" + assert event.is_interrupt is False + assert event.duration_ms == 4.0 + + +def test_an_interrupt_is_recorded_as_such(payloads: PayloadFactory) -> None: + event = parse_hook_event(payloads.post_tool_use_failure(is_interrupt=True)) + assert isinstance(event, PostToolUseFailureEvent) + assert event.is_interrupt is True + + +@pytest.mark.parametrize("field", ["session_id", "cwd", "hook_event_name"]) +def test_required_identity_is_enforced(payloads: PayloadFactory, field: str) -> None: + payload = payloads.pre_tool_use() + payload[field] = "" + with pytest.raises(ValueError): + parse_hook_event(payload) + + +@pytest.mark.parametrize("field", ["tool_name", "tool_use_id"]) +def test_required_tool_identity_is_enforced(payloads: PayloadFactory, field: str) -> None: + payload = payloads.pre_tool_use() + del payload[field] + with pytest.raises(ValueError): + parse_hook_event(payload) + + +def test_tool_input_must_be_a_mapping(payloads: PayloadFactory) -> None: + payload = payloads.pre_tool_use() + payload["tool_input"] = ["not", "a", "mapping"] + with pytest.raises(ValueError): + parse_hook_event(payload) + + +def test_an_unsupported_event_is_rejected(payloads: PayloadFactory) -> None: + payload = payloads.pre_tool_use() + payload["hook_event_name"] = "UserPromptSubmit" + with pytest.raises(ValueError): + parse_hook_event(payload) + + +def test_a_non_mapping_payload_is_rejected() -> None: + with pytest.raises(TypeError): + parse_hook_event(["not", "a", "mapping"]) # type: ignore[arg-type] + + +def test_a_malformed_duration_is_rejected(payloads: PayloadFactory) -> None: + payload = payloads.post_tool_use() + payload["duration_ms"] = "fast" + with pytest.raises(ValueError): + parse_hook_event(payload) + + +def test_a_malformed_interrupt_flag_is_rejected(payloads: PayloadFactory) -> None: + payload = payloads.post_tool_use_failure() + payload["is_interrupt"] = "yes" + with pytest.raises(ValueError): + parse_hook_event(payload) diff --git a/tests/integrations/claude_code/test_installer.py b/tests/integrations/claude_code/test_installer.py new file mode 100644 index 0000000..baa2a94 --- /dev/null +++ b/tests/integrations/claude_code/test_installer.py @@ -0,0 +1,121 @@ +import pytest + +from marginal.integrations.claude_code import installer + + +class FakeRunner: + def __init__(self, results: dict[str, installer.CommandResult]) -> None: + self.results = results + self.calls: list[list[str]] = [] + + def run(self, args: list[str]) -> installer.CommandResult: + self.calls.append(args) + for key, result in self.results.items(): + if key in args: + return result + return installer.CommandResult(0, "", "") + + +def _ok(stdout: str = "") -> installer.CommandResult: + return installer.CommandResult(0, stdout, "") + + +def _fail(stderr: str) -> installer.CommandResult: + return installer.CommandResult(1, "", stderr) + + +@pytest.fixture +def on_path(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(installer.shutil, "which", lambda name: f"/usr/bin/{name}") + + +def test_a_missing_cli_is_a_normal_answer(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(installer.shutil, "which", lambda name: None) + report = installer.inspect_claude_code() + assert report.available is False + assert report.capability_level == "none" + assert report.blocking_reasons == ("CLAUDE_CODE_NOT_FOUND",) + assert installer.install().error_code == "CLAUDE_CODE_NOT_FOUND" + assert installer.uninstall().error_code == "CLAUDE_CODE_NOT_FOUND" + + +def test_a_supported_version_reports_observe(on_path: None) -> None: + runner = FakeRunner({"--version": _ok("2.1.233 (Claude Code)")}) + report = installer.inspect_claude_code(runner=runner) + assert report.available is True + assert report.version == "2.1.233" + assert report.plugins_supported is True + assert report.capability_level == "observe" + assert report.blocking_reasons == () + assert report.to_dict()["capability_label"] == "Observe" + + +def test_an_old_version_blocks_installation(on_path: None) -> None: + runner = FakeRunner({"--version": _ok("1.0.9 (Claude Code)")}) + report = installer.inspect_claude_code(runner=runner) + assert report.plugins_supported is False + assert "PLUGIN_SUPPORT_UNVERIFIED" in report.blocking_reasons + + +def test_an_unreadable_version_is_reported(on_path: None) -> None: + runner = FakeRunner({"--version": _fail("boom")}) + report = installer.inspect_claude_code(runner=runner) + assert "VERSION_UNAVAILABLE" in report.blocking_reasons + + +def test_install_adds_the_marketplace_then_the_plugin(on_path: None) -> None: + runner = FakeRunner({}) + result = installer.install(runner=runner) + assert result.installed is True + assert result.changed is True + assert result.selector == "marginal-claude-code@marginal" + assert runner.calls[0][1:4] == ["plugin", "marketplace", "add"] + assert runner.calls[1][1:3] == ["plugin", "install"] + + +def test_install_tolerates_an_existing_marketplace(on_path: None) -> None: + runner = FakeRunner({"marketplace": _fail("marketplace already configured")}) + assert installer.install(runner=runner).installed is True + + +def test_install_reports_a_marketplace_failure(on_path: None) -> None: + runner = FakeRunner({"marketplace": _fail("network unreachable")}) + result = installer.install(runner=runner) + assert result.installed is False + assert result.error_code == "MARKETPLACE_FAILED" + assert result.message == "network unreachable" + + +def test_install_reports_a_plugin_failure(on_path: None) -> None: + runner = FakeRunner({"install": _fail("unknown plugin")}) + result = installer.install(runner=runner) + assert result.installed is False + assert result.error_code == "INSTALL_FAILED" + + +def test_uninstall_is_reversible_and_idempotent(on_path: None) -> None: + assert installer.uninstall(runner=FakeRunner({})).installed is False + absent = installer.uninstall(runner=FakeRunner({"uninstall": _fail("plugin not installed")})) + assert absent.installed is False + assert absent.error_code == "" + + +def test_uninstall_reports_a_real_failure(on_path: None) -> None: + result = installer.uninstall(runner=FakeRunner({"uninstall": _fail("permission denied")})) + assert result.installed is True + assert result.error_code == "UNINSTALL_FAILED" + + +def test_the_subprocess_runner_never_inherits_secrets( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("ANTHROPIC_API_KEY", "must-not-be-inherited") + runner = installer.SubprocessRunner(timeout_seconds=10.0) + result = runner.run(["/usr/bin/env"]) + assert "must-not-be-inherited" not in result.stdout + + +def test_the_subprocess_runner_reports_a_missing_binary() -> None: + runner = installer.SubprocessRunner() + result = runner.run(["/nonexistent/marginal-test-binary"]) + assert result.returncode == 127 diff --git a/tests/integrations/claude_code/test_normalization.py b/tests/integrations/claude_code/test_normalization.py new file mode 100644 index 0000000..055f329 --- /dev/null +++ b/tests/integrations/claude_code/test_normalization.py @@ -0,0 +1,61 @@ +import pytest + +from marginal.controls import ActionOutcomeStatus +from marginal.integrations.claude_code.events import parse_hook_event +from marginal.integrations.claude_code.normalization import ( + ENGINE, + session_boundary, + tool_call_end, + tool_call_start, +) + +from .conftest import PayloadFactory + + +def test_session_boundaries_map_to_start_and_end(payloads: PayloadFactory) -> None: + start = session_boundary(parse_hook_event(payloads.session_start())) # type: ignore[arg-type] + end = session_boundary(parse_hook_event(payloads.session_end())) # type: ignore[arg-type] + assert (start.kind, end.kind) == ("start", "end") + assert start.engine == ENGINE == "claude-code" + + +def test_a_proposal_keeps_its_call_and_turn_identity(payloads: PayloadFactory) -> None: + start = tool_call_start(parse_hook_event(payloads.pre_tool_use())) # type: ignore[arg-type] + assert start.call_id == "toolu_synthetic0001" + assert start.turn_id == "synthetic-prompt-0001" + assert start.tool_name == "Read" + + +def test_post_tool_use_is_a_documented_success(payloads: PayloadFactory) -> None: + end = tool_call_end(parse_hook_event(payloads.post_tool_use())) # type: ignore[arg-type] + assert end.outcome is ActionOutcomeStatus.SUCCESS + assert end.duration_ms == 11.0 + assert end.evidence is not None + + +def test_post_tool_use_failure_is_a_documented_failure(payloads: PayloadFactory) -> None: + end = tool_call_end(parse_hook_event(payloads.post_tool_use_failure())) # type: ignore[arg-type] + assert end.outcome is ActionOutcomeStatus.FAILURE + + +def test_an_interrupted_call_is_not_charged_as_a_tool_failure( + payloads: PayloadFactory, +) -> None: + payload = payloads.post_tool_use_failure(is_interrupt=True) + end = tool_call_end(parse_hook_event(payload)) # type: ignore[arg-type] + assert end.outcome is ActionOutcomeStatus.UNKNOWN + + +def test_different_failures_produce_different_evidence(payloads: PayloadFactory) -> None: + first = tool_call_end(parse_hook_event(payloads.post_tool_use_failure(error="Exit code 3"))) # type: ignore[arg-type] + second = tool_call_end(parse_hook_event(payloads.post_tool_use_failure(error="Exit code 4"))) # type: ignore[arg-type] + assert first.evidence != second.evidence + + +def test_normalization_rejects_foreign_event_types(payloads: PayloadFactory) -> None: + with pytest.raises(TypeError): + tool_call_start(parse_hook_event(payloads.post_tool_use())) # type: ignore[arg-type] + with pytest.raises(TypeError): + tool_call_end(parse_hook_event(payloads.pre_tool_use())) # type: ignore[arg-type] + with pytest.raises(TypeError): + session_boundary(parse_hook_event(payloads.pre_tool_use())) # type: ignore[arg-type] diff --git a/tests/integrations/claude_code/test_plugin.py b/tests/integrations/claude_code/test_plugin.py new file mode 100644 index 0000000..37019f9 --- /dev/null +++ b/tests/integrations/claude_code/test_plugin.py @@ -0,0 +1,103 @@ +"""The shipped plugin package must match the documented Claude Code contracts.""" + +import json +from pathlib import Path + +import pytest + +from marginal.integrations.claude_code.decisions import ( + build_post_tool_use_output, + build_pre_tool_use_output, +) +from marginal.integrations.claude_code.events import SUPPORTED_EVENTS +from marginal.integrations.claude_code.installer import MARKETPLACE_NAME, PLUGIN_NAME + +REPOSITORY_ROOT = Path(__file__).resolve().parents[3] +PLUGIN_ROOT = REPOSITORY_ROOT / "plugins" / PLUGIN_NAME + + +def _load(path: Path) -> dict[str, object]: + return json.loads(path.read_text(encoding="utf-8")) + + +def test_the_marketplace_lists_the_plugin() -> None: + marketplace = _load(REPOSITORY_ROOT / ".claude-plugin" / "marketplace.json") + assert marketplace["name"] == MARKETPLACE_NAME + plugins = marketplace["plugins"] + assert isinstance(plugins, list) + entry = next(plugin for plugin in plugins if plugin["name"] == PLUGIN_NAME) + assert (REPOSITORY_ROOT / str(entry["source"])).is_dir() + + +def test_the_plugin_manifest_matches_the_installer_selector() -> None: + manifest = _load(PLUGIN_ROOT / ".claude-plugin" / "plugin.json") + assert manifest["name"] == PLUGIN_NAME + assert manifest["license"] == "Apache-2.0" + description = manifest["description"] + assert isinstance(description, str) + assert "Shadow Mode" in description + + +def test_the_plugin_declares_only_supported_hooks() -> None: + hooks = _load(PLUGIN_ROOT / "hooks" / "hooks.json")["hooks"] + assert isinstance(hooks, dict) + assert set(hooks) == set(SUPPORTED_EVENTS) + + +def test_every_hook_runs_the_bundled_launcher() -> None: + hooks = _load(PLUGIN_ROOT / "hooks" / "hooks.json")["hooks"] + assert isinstance(hooks, dict) + for matchers in hooks.values(): + for matcher in matchers: + for hook in matcher["hooks"]: + assert hook["type"] == "command" + assert "${CLAUDE_PLUGIN_ROOT}/scripts/marginal_hook.py" in hook["command"] + assert hook["timeout"] <= 10 + + +def test_the_launcher_exists_and_never_raises_on_import() -> None: + launcher = PLUGIN_ROOT / "scripts" / "marginal_hook.py" + source = launcher.read_text(encoding="utf-8") + assert "CLAUDE_PLUGIN_ROOT" in source + assert "except Exception" in source + + +def test_shadow_mode_emits_no_output() -> None: + assert build_pre_tool_use_output(allowed=True, reason="allowed", reason_code="APPROVED") is None + assert build_post_tool_use_output(blocked=False, reason="ok", reason_code="APPROVED") is None + + +def test_the_deny_shape_follows_the_documented_contract() -> None: + output = build_pre_tool_use_output( + allowed=False, + reason="Repeated proven-success action produced no new evidence", + reason_code="NO_PROGRESS_ENFORCED", + ) + assert output is not None + specific = output["hookSpecificOutput"] + assert specific["hookEventName"] == "PreToolUse" + assert specific["permissionDecision"] == "deny" + assert specific["permissionDecisionReason"].endswith("[NO_PROGRESS_ENFORCED]") + + +def test_the_block_shape_follows_the_documented_contract() -> None: + output = build_post_tool_use_output( + blocked=True, + reason="Completion evidence did not change", + reason_code="NO_PROGRESS_ENFORCED", + ) + assert output is not None + specific = output["hookSpecificOutput"] + assert specific["hookEventName"] == "PostToolUse" + assert specific["decision"] == "block" + + +@pytest.mark.parametrize("reason_code", ["", " "]) +def test_a_blank_reason_code_is_rejected(reason_code: str) -> None: + with pytest.raises(ValueError): + build_pre_tool_use_output(allowed=False, reason="denied", reason_code=reason_code) + + +def test_a_blank_reason_is_rejected() -> None: + with pytest.raises(ValueError): + build_pre_tool_use_output(allowed=False, reason=" ", reason_code="NO_PROGRESS_ENFORCED") diff --git a/tests/integrations/claude_code/test_service.py b/tests/integrations/claude_code/test_service.py new file mode 100644 index 0000000..c7fb31e --- /dev/null +++ b/tests/integrations/claude_code/test_service.py @@ -0,0 +1,258 @@ +import json +import subprocess +import sys +import time +from pathlib import Path + +import pytest + +from marginal.controls import ActionOutcomeStatus +from marginal.integrations.claude_code import service +from marginal.integrations.claude_code.events import parse_hook_event +from marginal.ledger import read_decision_ledger + +from .conftest import SESSION_ID, PayloadFactory + + +@pytest.fixture +def data_root(tmp_path: Path) -> Path: + return tmp_path / "plugin-data" + + +@pytest.fixture +def session(data_root: Path, payloads: PayloadFactory): # type: ignore[no-untyped-def] + start = parse_hook_event(payloads.session_start()) + service.start_session_service(start, data_root=data_root) # type: ignore[arg-type] + yield + service.stop_session_service(SESSION_ID, data_root=data_root) + + +def _ledger_records(data_root: Path) -> list[dict[str, object]]: + files = sorted(data_root.rglob("*.jsonl")) + assert files, "the session must write exactly one decision ledger" + return read_decision_ledger(files[0]) + + +def test_a_full_tool_lifecycle_is_recorded( + session: None, data_root: Path, payloads: PayloadFactory +) -> None: + assert service.run_hook(payloads.pre_tool_use(), data_root=data_root).output is None + assert service.run_hook(payloads.post_tool_use(), data_root=data_root).output is None + events = [record["event"] for record in _ledger_records(data_root)] + assert events[0] == "hook_session_start" + assert "hook_decision" in events + assert "authorization" in events + assert "hook_outcome" in events + assert "commit" in events + + +def test_shadow_mode_never_returns_output( + session: None, data_root: Path, payloads: PayloadFactory +) -> None: + for index in range(4): + call_id = f"toolu_synthetic000{index}" + assert service.run_hook(payloads.pre_tool_use(call_id), data_root=data_root).output is None + assert service.run_hook(payloads.post_tool_use(call_id), data_root=data_root).output is None + + +def test_repeated_identical_work_is_recommended_against( + session: None, data_root: Path, payloads: PayloadFactory +) -> None: + for index in range(4): + call_id = f"toolu_synthetic000{index}" + service.run_hook(payloads.pre_tool_use(call_id), data_root=data_root) + service.run_hook(payloads.post_tool_use(call_id), data_root=data_root) + decisions = [ + record for record in _ledger_records(data_root) if record["event"] == "hook_decision" + ] + assert len(decisions) == 4 + assert decisions[0]["recommended"] is True + assert decisions[-1]["recommended"] is False + assert decisions[-1]["recommended_stop"] is True + assert decisions[-1]["enforced"] is False + assert decisions[-1]["no_progress_reason_code"] == "NO_PROGRESS_ENFORCEMENT_ELIGIBLE" + + +def test_measured_latency_and_outcome_source_are_recorded( + session: None, data_root: Path, payloads: PayloadFactory +) -> None: + service.run_hook(payloads.pre_tool_use(), data_root=data_root) + service.run_hook(payloads.post_tool_use(duration_ms=17.0), data_root=data_root) + failure_id = "toolu_synthetic0002" + service.run_hook( + payloads.pre_tool_use(failure_id, tool_name="Bash", tool_input={"command": "exit 3"}), + data_root=data_root, + ) + service.run_hook(payloads.post_tool_use_failure(failure_id), data_root=data_root) + outcomes = [ + record for record in _ledger_records(data_root) if record["event"] == "hook_outcome" + ] + assert outcomes[0]["outcome"] == "success" + assert outcomes[0]["outcome_source"] == "PostToolUse" + assert outcomes[0]["duration_ms"] == 17.0 + assert outcomes[1]["outcome"] == "failure" + assert outcomes[1]["outcome_source"] == "PostToolUseFailure" + + +def test_the_ledger_holds_no_paths_or_commands( + session: None, data_root: Path, payloads: PayloadFactory, workspace: Path +) -> None: + service.run_hook( + payloads.pre_tool_use(tool_name="Bash", tool_input={"command": "cat /etc/passwd"}), + data_root=data_root, + ) + service.run_hook( + payloads.post_tool_use( + tool_name="Bash", + tool_input={"command": "cat /etc/passwd"}, + tool_response={"stdout": "root:x:0:0", "stderr": ""}, + ), + data_root=data_root, + ) + files = sorted(data_root.rglob("*.jsonl")) + contents = files[0].read_text(encoding="utf-8") + assert "/etc/passwd" not in contents + assert "root:x:0:0" not in contents + assert str(workspace) not in contents + + +def test_a_strict_profile_pseudonymizes_the_session_identity( + monkeypatch: pytest.MonkeyPatch, data_root: Path, payloads: PayloadFactory +) -> None: + monkeypatch.setenv("MARGINAL_PRIVACY_PROFILE", "safe_telemetry") + start = parse_hook_event(payloads.session_start()) + service.start_session_service(start, data_root=data_root) # type: ignore[arg-type] + try: + service.run_hook(payloads.pre_tool_use(), data_root=data_root) + service.run_hook(payloads.post_tool_use(), data_root=data_root) + finally: + service.stop_session_service(SESSION_ID, data_root=data_root) + contents = sorted(data_root.rglob("*.jsonl"))[0].read_text(encoding="utf-8") + assert SESSION_ID not in contents + assert "safe_telemetry" in contents + + +def test_an_unknown_privacy_profile_falls_back_to_local( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("MARGINAL_PRIVACY_PROFILE", "not-a-profile") + assert service.selected_privacy_profile().value == "local_full" + monkeypatch.setenv("MARGINAL_PRIVACY_PROFILE", "safe_telemetry") + assert service.selected_privacy_profile().value == "safe_telemetry" + + +def test_the_session_end_summary_is_recorded(data_root: Path, payloads: PayloadFactory) -> None: + start = parse_hook_event(payloads.session_start()) + service.start_session_service(start, data_root=data_root) # type: ignore[arg-type] + service.run_hook(payloads.pre_tool_use(), data_root=data_root) + service.run_hook(payloads.session_end(), data_root=data_root) + summaries = [ + record for record in _ledger_records(data_root) if record["event"] == "hook_session_end" + ] + assert len(summaries) == 1 + assert summaries[0]["summary_pending_actions"] == 0 + assert summaries[0]["summary_unknown_observations"] == 1 + + +def test_a_hook_without_a_service_fails_open(data_root: Path, payloads: PayloadFactory) -> None: + result = service.run_hook(payloads.pre_tool_use(), data_root=data_root) + assert result.exit_code == 0 + assert result.output is None + assert result.warning_code == "SERVICE_UNAVAILABLE" + + +def test_an_unparsable_payload_fails_open(data_root: Path) -> None: + result = service.run_hook({"hook_event_name": "Nonsense"}, data_root=data_root) + assert result.exit_code == 0 + assert result.warning_code == "INTEGRATION_ERROR" + + +def test_a_completion_without_a_proposal_fails_open( + session: None, data_root: Path, payloads: PayloadFactory +) -> None: + result = service.run_hook(payloads.post_tool_use("toolu_synthetic9999"), data_root=data_root) + assert result.exit_code == 0 + assert result.output is None + outcomes = [ + record for record in _ledger_records(data_root) if record["event"] == "hook_outcome" + ] + assert outcomes[-1]["outcome"] == "unknown" + + +def test_stopping_an_unknown_session_is_harmless(data_root: Path) -> None: + service.stop_session_service("synthetic-session-9999", data_root=data_root) + + +def test_starting_requires_session_start(payloads: PayloadFactory, data_root: Path) -> None: + end = parse_hook_event(payloads.session_end()) + with pytest.raises(ValueError): + service.start_session_service(end, data_root=data_root) # type: ignore[arg-type] + + +def test_outcome_classification_is_available_without_a_session( + payloads: PayloadFactory, +) -> None: + assert service.observe_outcome(payloads.post_tool_use()) is ActionOutcomeStatus.SUCCESS + assert service.observe_outcome(payloads.post_tool_use_failure()) is ActionOutcomeStatus.FAILURE + with pytest.raises(ValueError): + service.observe_outcome(payloads.pre_tool_use()) + + +def test_the_hook_entry_point_is_silent_without_a_data_root( + monkeypatch: pytest.MonkeyPatch, +) -> None: + for name in service.DATA_ROOT_VARIABLES: + monkeypatch.delenv(name, raising=False) + assert service.hook_main([]) == 0 + + +def test_the_hook_entry_point_reads_stdin_and_exits_zero( + session: None, data_root: Path, payloads: PayloadFactory +) -> None: + completed = subprocess.run( + [sys.executable, "-m", "marginal.integrations.claude_code.service"], + input=json.dumps(payloads.pre_tool_use()), + capture_output=True, + text=True, + env={ + "PATH": "/usr/bin:/bin:/usr/local/bin", + "CLAUDE_PLUGIN_DATA": str(data_root), + "PYTHONPATH": str(Path(__file__).resolve().parents[3] / "src"), + }, + check=False, + ) + assert completed.returncode == 0 + assert completed.stdout.strip() == "" + + +def test_a_spawned_service_governs_a_full_lifecycle( + data_root: Path, payloads: PayloadFactory +) -> None: + environment = { + "PATH": "/usr/bin:/bin:/usr/local/bin", + "CLAUDE_PLUGIN_DATA": str(data_root), + "PYTHONPATH": str(Path(__file__).resolve().parents[3] / "src"), + } + + def hook(payload: dict[str, object]) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [sys.executable, "-m", "marginal.integrations.claude_code.service"], + input=json.dumps(payload), + capture_output=True, + text=True, + env=environment, + check=False, + ) + + assert hook(payloads.session_start()).returncode == 0 + assert hook(payloads.pre_tool_use()).returncode == 0 + assert hook(payloads.post_tool_use()).returncode == 0 + assert hook(payloads.session_end()).returncode == 0 + events = [record["event"] for record in _ledger_records(data_root)] + assert "hook_decision" in events + assert "hook_outcome" in events + assert "hook_session_end" in events + deadline = time.monotonic() + 5.0 + while time.monotonic() < deadline and list((data_root / "sessions").glob("*.json")): + time.sleep(0.05) + assert not list((data_root / "sessions").glob("*.json")) diff --git a/tests/integrations/hookkit/__init__.py b/tests/integrations/hookkit/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/integrations/hookkit/conftest.py b/tests/integrations/hookkit/conftest.py new file mode 100644 index 0000000..4228e5a --- /dev/null +++ b/tests/integrations/hookkit/conftest.py @@ -0,0 +1,34 @@ +import subprocess +from pathlib import Path + +import pytest + + +def _git(repo: Path, *args: str) -> None: + subprocess.run( + ["git", *args], + cwd=repo, + check=True, + capture_output=True, + env={ + "PATH": "/usr/bin:/bin:/usr/local/bin", + "HOME": str(repo), + "GIT_AUTHOR_NAME": "Test", + "GIT_AUTHOR_EMAIL": "test@example.invalid", + "GIT_COMMITTER_NAME": "Test", + "GIT_COMMITTER_EMAIL": "test@example.invalid", + }, + ) + + +@pytest.fixture +def git_workspace(tmp_path: Path) -> Path: + """A committed Git work tree, so workspace state is observable.""" + + workspace = tmp_path / "workspace" + workspace.mkdir() + _git(workspace, "init", "-q") + (workspace / "tracked.txt").write_text("first\n", encoding="utf-8") + _git(workspace, "add", "tracked.txt") + _git(workspace, "commit", "-qm", "initial") + return workspace diff --git a/tests/integrations/hookkit/test_bootstrap.py b/tests/integrations/hookkit/test_bootstrap.py new file mode 100644 index 0000000..255064f --- /dev/null +++ b/tests/integrations/hookkit/test_bootstrap.py @@ -0,0 +1,93 @@ +from pathlib import Path + +import pytest + +from marginal import BudgetLimits, Treasury +from marginal.integrations.hookkit.bootstrap import ( + OBSERVE_CAPABILITIES, + build_observe_session, + session_hash, + workspace_hash, +) +from marginal.ledger import read_decision_ledger +from marginal.privacy import PrivacyProfile +from marginal.runtime import UniversalRuntime + + +def test_observe_capabilities_declare_no_control() -> None: + assert OBSERVE_CAPABILITIES.level == "observe" + assert OBSERVE_CAPABILITIES.block_actions is False + assert OBSERVE_CAPABILITIES.observe_model_usage is False + assert OBSERVE_CAPABILITIES.record_outcomes is False + + +def test_an_observe_adapter_cannot_be_run_in_an_enforcing_mode() -> None: + with pytest.raises(ValueError): + UniversalRuntime( + Treasury(BudgetLimits(), mode="enforce"), + engine="test-engine", + session_id="session-1", + task_id="task-1", + capabilities=OBSERVE_CAPABILITIES, + ) + + +def test_identifiers_are_pseudonymized(tmp_path: Path) -> None: + assert len(session_hash("session-1")) == 64 + assert session_hash("session-1") != "session-1" + assert workspace_hash(tmp_path) == workspace_hash(tmp_path) + with pytest.raises(ValueError): + session_hash("") + + +def test_the_session_writes_a_pseudonymous_local_ledger(tmp_path: Path) -> None: + session = build_observe_session( + engine="test-engine", + session_id="session-1", + workspace=tmp_path / "workspace", + data_root=tmp_path / "data", + ) + assert session.runtime.treasury.mode.value == "shadow" + assert session.ledger_path.name == f"{session_hash('session-1')}.jsonl" + assert "session-1" not in str(session.ledger_path) + session.record({"event": "hook_session_start"}) + records = read_decision_ledger(session.ledger_path) + assert [record["event"] for record in records] == ["hook_session_start"] + assert records[0]["engine"] == "test-engine" + assert records[0]["run_id"] == session.session_hash + + +def test_recording_fails_open_on_an_invalid_event(tmp_path: Path) -> None: + session = build_observe_session( + engine="test-engine", + session_id="session-1", + workspace=tmp_path / "workspace", + data_root=tmp_path / "data", + ) + session.record({"event": ""}) + session.record({"event": "hook_decision", "engine": "spoofed"}) + assert not session.ledger_path.exists() or read_decision_ledger(session.ledger_path) == [] + + +def test_a_strict_privacy_profile_is_selectable(tmp_path: Path) -> None: + session = build_observe_session( + engine="test-engine", + session_id="session-1", + workspace=tmp_path / "workspace", + data_root=tmp_path / "data", + privacy_profile=PrivacyProfile.SAFE_TELEMETRY, + privacy_key_path=tmp_path / "key", + ) + session.record({"event": "hook_session_start"}) + records = read_decision_ledger(session.ledger_path) + assert records[0]["privacy_profile"] == PrivacyProfile.SAFE_TELEMETRY.value + + +def test_an_empty_engine_is_rejected(tmp_path: Path) -> None: + with pytest.raises(ValueError): + build_observe_session( + engine=" ", + session_id="session-1", + workspace=tmp_path, + data_root=tmp_path / "data", + ) diff --git a/tests/integrations/hookkit/test_events.py b/tests/integrations/hookkit/test_events.py new file mode 100644 index 0000000..f071d8f --- /dev/null +++ b/tests/integrations/hookkit/test_events.py @@ -0,0 +1,75 @@ +import pytest + +from marginal.controls import ActionOutcomeStatus +from marginal.integrations.hookkit.events import SessionBoundary, ToolCallEnd, ToolCallStart + + +def test_session_boundary_requires_a_known_kind() -> None: + with pytest.raises(ValueError): + SessionBoundary( + engine="test-engine", + session_id="session-1", + workspace="/workspace", + kind="restart", + ) + + +def test_session_boundary_rejects_empty_identity() -> None: + with pytest.raises(ValueError): + SessionBoundary(engine="", session_id="session-1", workspace="/workspace", kind="start") + + +def test_tool_call_start_copies_and_defaults_turn_identity() -> None: + arguments = {"file_path": "/workspace/example.txt"} + start = ToolCallStart( + session_id="session-1", + call_id="call-1", + tool_name="Read", + tool_input=arguments, + ) + arguments["file_path"] = "/workspace/mutated.txt" + assert start.tool_input == {"file_path": "/workspace/example.txt"} + assert start.turn_id == "" + + +def test_tool_call_start_rejects_non_mapping_arguments() -> None: + with pytest.raises(TypeError): + ToolCallStart( + session_id="session-1", + call_id="call-1", + tool_name="Read", + tool_input=["not", "a", "mapping"], # type: ignore[arg-type] + ) + + +def test_tool_call_end_parses_outcome_and_duration() -> None: + end = ToolCallEnd( + session_id="session-1", + call_id="call-1", + tool_name="Read", + outcome="success", # type: ignore[arg-type] + duration_ms=12, + ) + assert end.outcome is ActionOutcomeStatus.SUCCESS + assert end.duration_ms == 12.0 + + +def test_tool_call_end_rejects_negative_duration() -> None: + with pytest.raises(ValueError): + ToolCallEnd( + session_id="session-1", + call_id="call-1", + tool_name="Read", + outcome=ActionOutcomeStatus.SUCCESS, + duration_ms=-1.0, + ) + + +def test_tool_call_end_rejects_unknown_outcome_text() -> None: + with pytest.raises(ValueError): + ToolCallEnd( + session_id="session-1", + call_id="call-1", + tool_name="Read", + outcome="probably-fine", # type: ignore[arg-type] + ) diff --git a/tests/integrations/hookkit/test_normalization.py b/tests/integrations/hookkit/test_normalization.py new file mode 100644 index 0000000..8e5efb7 --- /dev/null +++ b/tests/integrations/hookkit/test_normalization.py @@ -0,0 +1,108 @@ +import json + +import pytest + +from marginal.integrations.hookkit.events import ToolCallStart +from marginal.integrations.hookkit.normalization import ( + classify_action, + normalize_tool_call, + semantic_key, +) +from marginal.protocol import DeduplicationScope + + +def _start(tool: str = "Read", **arguments: object) -> ToolCallStart: + return ToolCallStart( + session_id="session-1", + call_id="call-1", + tool_name=tool, + tool_input=arguments or {"file_path": "/workspace/example.txt"}, + turn_id="turn-1", + ) + + +def test_semantic_key_is_stable_and_argument_sensitive() -> None: + first = semantic_key(_start()) + assert first == semantic_key(_start()) + assert first != semantic_key(_start(file_path="/workspace/other.txt")) + assert first != semantic_key(_start(tool="Write")) + + +def test_semantic_key_ignores_tool_name_case_only() -> None: + assert semantic_key(_start(tool="read")) == semantic_key(_start(tool="READ")) + + +@pytest.mark.parametrize( + ("tool", "arguments", "expected_kind", "expected_verification"), + [ + ("Bash", {"command": "pytest -q"}, "verification", True), + ("bash", {"command": "echo hi"}, "shell", False), + ("Edit", {"file_path": "/workspace/a.py"}, "edit", False), + ("Read", {"file_path": "/workspace/a.py"}, "read", False), + ("WebFetch", {"url": "https://example.invalid"}, "fetch", False), + ("mcp__server__tool", {"argument": 1}, "mcp", False), + ("Task", {"prompt_kind": "subagent"}, "tool", False), + ], +) +def test_classify_action_labels_generic_kinds( + tool: str, + arguments: dict[str, object], + expected_kind: str, + expected_verification: bool, +) -> None: + kind, is_verification = classify_action(_start(tool=tool, **arguments)) + assert kind == expected_kind + assert is_verification is expected_verification + + +def test_normalized_action_retains_no_raw_arguments() -> None: + action = normalize_tool_call( + _start(file_path="/workspace/secret-plan.txt"), + engine="test-engine", + state_hash="state-hash", + ) + serialized = json.dumps( + { + "name": action.name, + "kind": action.kind, + "phase": action.phase, + "state_hash": action.state_hash, + "metadata": dict(action.metadata), + } + ) + assert "secret-plan" not in serialized + assert action.deduplication_scope is DeduplicationScope.ONCE_PER_STATE + assert action.phase == "test-engine-tool-use" + assert action.metadata["turn_id"] == "turn-1" + assert action.metadata["evidence_hash"] == "" + + +def test_normalization_accepts_unobservable_state() -> None: + action = normalize_tool_call(_start(), engine="test-engine", state_hash="") + assert action.state_hash == "" + + +def test_normalization_records_previous_evidence() -> None: + action = normalize_tool_call( + _start(), + engine="test-engine", + state_hash="state-hash", + previous_evidence_hash="evidence-hash", + ) + assert action.metadata["evidence_hash"] == "evidence-hash" + + +def test_normalization_rejects_an_empty_engine() -> None: + with pytest.raises(ValueError): + normalize_tool_call(_start(), engine=" ", state_hash="state-hash") + + +def test_normalization_rejects_unserializable_arguments() -> None: + start = ToolCallStart( + session_id="session-1", + call_id="call-1", + tool_name="Read", + tool_input={"handle": object()}, + ) + with pytest.raises(ValueError): + normalize_tool_call(start, engine="test-engine", state_hash="state-hash") diff --git a/tests/integrations/hookkit/test_outcomes.py b/tests/integrations/hookkit/test_outcomes.py new file mode 100644 index 0000000..7f61b6b --- /dev/null +++ b/tests/integrations/hookkit/test_outcomes.py @@ -0,0 +1,54 @@ +import pytest + +from marginal.controls import ActionOutcomeStatus +from marginal.integrations.hookkit.outcomes import ( + classify_structured_result, + completion_evidence_hash, +) + + +@pytest.mark.parametrize( + ("result", "expected"), + [ + ({"exit_code": 0}, ActionOutcomeStatus.SUCCESS), + ({"exit_code": 3}, ActionOutcomeStatus.FAILURE), + ({"metadata": {"exit": 0}}, ActionOutcomeStatus.SUCCESS), + ({"metadata": {"exit": 3}}, ActionOutcomeStatus.FAILURE), + ({"success": True}, ActionOutcomeStatus.SUCCESS), + ({"is_error": True}, ActionOutcomeStatus.FAILURE), + ({"status": "passed"}, ActionOutcomeStatus.SUCCESS), + ({"outcome": "failed"}, ActionOutcomeStatus.FAILURE), + ], +) +def test_structured_signals_are_classified( + result: dict[str, object], expected: ActionOutcomeStatus +) -> None: + assert classify_structured_result(result) is expected + + +@pytest.mark.parametrize( + "result", + [ + "completed successfully", + {"output": "tests passed"}, + {"exit_code": 0, "success": False}, + {"metadata": {"exit": 0}, "status": "failed"}, + {"exit_code": True}, + None, + [], + ], +) +def test_unprovable_results_stay_unknown(result: object) -> None: + assert classify_structured_result(result) is ActionOutcomeStatus.UNKNOWN + + +def test_evidence_hash_is_stable_and_content_sensitive() -> None: + first = completion_evidence_hash({"type": "text", "content": "hello"}) + assert first == completion_evidence_hash({"content": "hello", "type": "text"}) + assert first != completion_evidence_hash({"type": "text", "content": "hello world"}) + assert len(first) == 64 + + +def test_evidence_hash_is_empty_when_unavailable() -> None: + assert completion_evidence_hash(None) == "" + assert completion_evidence_hash({"handle": object()}) == "" diff --git a/tests/integrations/hookkit/test_session.py b/tests/integrations/hookkit/test_session.py new file mode 100644 index 0000000..a1c3a24 --- /dev/null +++ b/tests/integrations/hookkit/test_session.py @@ -0,0 +1,186 @@ +from pathlib import Path + +import pytest + +from marginal import BudgetLimits, Treasury +from marginal.controls import ActionOutcomeStatus +from marginal.integrations.hookkit.bootstrap import OBSERVE_CAPABILITIES +from marginal.integrations.hookkit.events import ToolCallEnd, ToolCallStart +from marginal.integrations.hookkit.session import HookIntegrationError, HookSessionRuntime +from marginal.runtime import UniversalRuntime + +SESSION = "session-1" + + +def _runtime() -> UniversalRuntime: + return UniversalRuntime( + Treasury(BudgetLimits(), mode="shadow"), + engine="test-engine", + session_id=SESSION, + task_id="task-1", + capabilities=OBSERVE_CAPABILITIES, + ) + + +def _session(workspace: Path) -> HookSessionRuntime: + return HookSessionRuntime(_runtime(), workspace=workspace) + + +def _start(call_id: str, tool: str = "Read", **arguments: object) -> ToolCallStart: + return ToolCallStart( + session_id=SESSION, + call_id=call_id, + tool_name=tool, + tool_input=arguments or {"file_path": "/workspace/example.txt"}, + ) + + +def _end( + call_id: str, + *, + tool: str = "Read", + outcome: ActionOutcomeStatus = ActionOutcomeStatus.SUCCESS, + evidence: object = None, + duration_ms: float | None = None, +) -> ToolCallEnd: + return ToolCallEnd( + session_id=SESSION, + call_id=call_id, + tool_name=tool, + outcome=outcome, + evidence=evidence if evidence is not None else {"content": "hello"}, + duration_ms=duration_ms, + ) + + +def test_a_completed_call_is_settled_exactly_once(tmp_path: Path) -> None: + session = _session(tmp_path) + decision = session.tool_call_start(_start("call-1")) + assert decision.allowed is True + assert session.pending_action_ids() == ("call-1",) + assert session.tool_call_end(_end("call-1", duration_ms=7.0)) is ActionOutcomeStatus.SUCCESS + assert session.pending_action_ids() == () + assert session.summary()["successful_observations"] == 1 + assert session.summary()["enforced_denials"] == 0 + + +def test_an_observed_session_never_denies(tmp_path: Path) -> None: + session = _session(tmp_path) + for index in range(4): + decision = session.tool_call_start(_start(f"call-{index}")) + assert decision.allowed is True + session.tool_call_end(_end(f"call-{index}")) + assert session.summary()["enforced_denials"] == 0 + + +def test_repeated_identical_work_is_recommended_against(git_workspace: Path) -> None: + session = _session(git_workspace) + recommendations: list[bool] = [] + signals: list[str] = [] + for index in range(4): + decision = session.tool_call_start(_start(f"call-{index}")) + recommendations.append(decision.recommended) + signal = session.last_no_progress_signal + signals.append(signal.reason_code if signal else "") + session.tool_call_end(_end(f"call-{index}")) + assert recommendations[0] is True + assert recommendations[-1] is False + assert signals[-1] == "NO_PROGRESS_ENFORCEMENT_ELIGIBLE" + assert session.summary()["recommended_stops"] >= 1 + + +def test_new_evidence_clears_the_repetition_signal(git_workspace: Path) -> None: + session = _session(git_workspace) + session.tool_call_start(_start("call-1")) + session.tool_call_end(_end("call-1", evidence={"content": "first"})) + session.tool_call_start(_start("call-2")) + session.tool_call_end(_end("call-2", evidence={"content": "second"})) + session.tool_call_start(_start("call-3")) + signal = session.last_no_progress_signal + assert signal is not None + assert signal.should_recommend_stop is False + + +def test_a_duplicate_pending_identity_is_rejected(tmp_path: Path) -> None: + session = _session(tmp_path) + session.tool_call_start(_start("call-1")) + with pytest.raises(HookIntegrationError): + session.tool_call_start(_start("call-1")) + + +def test_a_completion_for_a_different_tool_is_rejected(tmp_path: Path) -> None: + session = _session(tmp_path) + session.tool_call_start(_start("call-1", tool="Read")) + with pytest.raises(HookIntegrationError): + session.tool_call_end(_end("call-1", tool="Bash")) + + +def test_a_completion_without_a_proposal_is_reported_not_settled(tmp_path: Path) -> None: + session = _session(tmp_path) + assert session.tool_call_end(_end("call-unknown")) is ActionOutcomeStatus.UNKNOWN + assert session.summary()["unmatched_completions"] == 1 + assert session.summary()["completed_observations"] == 0 + + +def test_a_foreign_session_identity_is_rejected(tmp_path: Path) -> None: + session = _session(tmp_path) + foreign = ToolCallStart( + session_id="other-session", + call_id="call-1", + tool_name="Read", + tool_input={"file_path": "/workspace/example.txt"}, + ) + with pytest.raises(HookIntegrationError): + session.tool_call_start(foreign) + + +def test_close_settles_unobserved_proposals(tmp_path: Path) -> None: + session = _session(tmp_path) + session.tool_call_start(_start("call-1")) + session.close() + assert session.pending_action_ids() == () + assert session.summary()["unknown_observations"] == 1 + with pytest.raises(HookIntegrationError): + session.tool_call_start(_start("call-2")) + + +def test_close_is_idempotent(tmp_path: Path) -> None: + session = _session(tmp_path) + session.close() + session.close() + assert session.summary()["pending_actions"] == 0 + + +def test_action_evidence_never_exposes_raw_identity(tmp_path: Path) -> None: + session = _session(tmp_path) + session.tool_call_start(_start("call-1", file_path="/workspace/secret.txt")) + evidence = session.action_evidence("call-1") + assert evidence is not None + assert "secret" not in str(evidence) + assert "call-1" not in str(evidence) + assert len(evidence["action_hash"]) == 64 + assert session.action_evidence("missing") is None + + +def test_an_explicit_failure_is_settled_as_a_failure(tmp_path: Path) -> None: + session = _session(tmp_path) + session.tool_call_start(_start("call-1")) + outcome = session.tool_call_end( + _end("call-1", outcome=ActionOutcomeStatus.FAILURE, duration_ms=3.0) + ) + assert outcome is ActionOutcomeStatus.FAILURE + assert session.summary()["failed_observations"] == 1 + + +def test_an_unknown_outcome_is_not_counted_as_success(tmp_path: Path) -> None: + session = _session(tmp_path) + session.tool_call_start(_start("call-1")) + outcome = session.tool_call_end(_end("call-1", outcome=ActionOutcomeStatus.UNKNOWN)) + assert outcome is ActionOutcomeStatus.UNKNOWN + assert session.summary()["unknown_observations"] == 1 + assert session.summary()["successful_observations"] == 0 + + +def test_the_runtime_must_be_a_universal_runtime(tmp_path: Path) -> None: + with pytest.raises(TypeError): + HookSessionRuntime(object(), workspace=tmp_path) # type: ignore[arg-type] diff --git a/tests/integrations/hookkit/test_state.py b/tests/integrations/hookkit/test_state.py new file mode 100644 index 0000000..1c07ad2 --- /dev/null +++ b/tests/integrations/hookkit/test_state.py @@ -0,0 +1,78 @@ +import subprocess +from pathlib import Path + +from marginal.integrations.hookkit.state import workspace_state_hash + + +def _git(repo: Path, *args: str) -> None: + subprocess.run( + ["git", *args], + cwd=repo, + check=True, + capture_output=True, + env={ + "PATH": "/usr/bin:/bin:/usr/local/bin", + "HOME": str(repo), + "GIT_AUTHOR_NAME": "Test", + "GIT_AUTHOR_EMAIL": "test@example.invalid", + "GIT_COMMITTER_NAME": "Test", + "GIT_COMMITTER_EMAIL": "test@example.invalid", + }, + ) + + +def _repository(root: Path, *, commit: bool = True) -> Path: + _git(root, "init", "-q") + (root / "tracked.txt").write_text("first\n", encoding="utf-8") + if commit: + _git(root, "add", "tracked.txt") + _git(root, "commit", "-qm", "initial") + return root + + +def test_state_hash_is_stable_for_an_unchanged_workspace(tmp_path: Path) -> None: + repo = _repository(tmp_path) + assert workspace_state_hash(repo) == workspace_state_hash(repo) + assert len(workspace_state_hash(repo)) == 64 + + +def test_state_hash_changes_when_tracked_content_changes(tmp_path: Path) -> None: + repo = _repository(tmp_path) + before = workspace_state_hash(repo) + (repo / "tracked.txt").write_text("second\n", encoding="utf-8") + assert workspace_state_hash(repo) != before + + +def test_state_hash_changes_when_untracked_content_changes(tmp_path: Path) -> None: + repo = _repository(tmp_path) + before = workspace_state_hash(repo) + (repo / "untracked.txt").write_text("new\n", encoding="utf-8") + assert workspace_state_hash(repo) != before + + +def test_a_repository_without_commits_is_still_observable(tmp_path: Path) -> None: + repo = _repository(tmp_path, commit=False) + first = workspace_state_hash(repo) + assert first != "" + (repo / "untracked.txt").write_text("new\n", encoding="utf-8") + assert workspace_state_hash(repo) != first + + +def test_a_subdirectory_resolves_to_its_work_tree(tmp_path: Path) -> None: + repo = _repository(tmp_path) + nested = repo / "nested" + nested.mkdir() + assert workspace_state_hash(nested) == workspace_state_hash(repo) + + +def test_state_is_unobservable_outside_a_repository(tmp_path: Path) -> None: + assert workspace_state_hash(tmp_path / "not-a-repository") == "" + + +def test_ignored_directories_do_not_change_state(tmp_path: Path) -> None: + repo = _repository(tmp_path) + before = workspace_state_hash(repo) + marginal_state = repo / ".marginal" + marginal_state.mkdir() + (marginal_state / "ledger.jsonl").write_text("{}\n", encoding="utf-8") + assert workspace_state_hash(repo) == before