Merge pull request #688 from AssetOverflow/codex/feat-claude-tool-authority-demo

feat(demo): add frontier-proposer tool authority demo

demos/claude_tool_authority/.gitignore

@@ -0,0 +1 @@
+out/

demos/claude_tool_authority/README.md

@@ -0,0 +1,86 @@
+# Frontier-Proposer-to-CORE Tool Authority Demo
+
+This demo proves one narrow boundary:
+
+```text
+A model-style proposer suggests a digital action.
+CORE alone decides authorized | ask | refused | invalid.
+The output is a deterministic authority artifact.
+Nothing executes.
+```
+
+## What this proves
+
+* A proposer can submit an MCP-shaped action request without gaining execution authority.
+* CORE alone derives the final authority status.
+* CORE ignores any proposer-supplied trace hash and regenerates its own deterministic trace hash.
+* Invalid payloads fail at the typed boundary before authority evaluation.
+* An `authorized` result emits only an inert `licensed_action` record, not an execution path.
+
+## What this does not prove
+
+* It is not a production MCP server.
+* It does not send email, run shell commands, call a network, or invoke a model API.
+* It does not prove runtime integration, serving integration, or embodied authority.
+* It does not claim broader safety than the local deterministic envelope encoded here.
+
+## Why no real side effects are executed
+
+The authority substrate never dispatches tools.  `authority.py` validates a closed
+payload, evaluates local policy, and returns JSON only.  Even when a proposal is
+authorized, the returned `licensed_action` is an inert description with
+`effect: "inert_license_only"`.  No file write, email send, shell execution,
+network access, subprocess launch, `eval`, or `exec` path exists in the demo.
+
+## Why this is MCP-shaped, not production MCP
+
+The payload is structured like a tool invocation so future proposer lanes can
+hand the same kind of typed request to CORE.  This remains a local demo
+contract: no server transport, session handling, production adapter, or real
+side-effecting tool substrate is present.
+
+## Relation to #687
+
+#687 proved the earlier reasoning boundary:
+
+```text
+System 1-style proposal
+-> CORE deterministic System 2 verification/refusal/ask/invalid
+-> audited envelope
+-> deterministic trace artifacts
+-> no proposer execution authority
+```
+
+This demo advances the same doctrine one layer outward:
+
+```text
+model-style proposer
+-> proposes digital actions
+-> CORE authorizes/refuses/asks/invalidates
+-> inert licensed action artifact only when authorized
+-> no proposer authority and no execution path
+```
+
+It therefore proves digital tool/action authority before any embodied-authority simulation.
+
+## The four scenarios
+
+* `authorized-low-risk-local-action`: low-risk local note request inside the envelope.
+* `ask-required-action`: external email draft request without explicit confirmation.
+* `refused-outside-envelope-action`: shell command proposal refused as an unauthorized tool.
+* `invalid-smuggling-attempt`: proposer tries to smuggle `licensed_action` and authorization state.
+
+## Honesty ledger
+
+* Real: closed schema validation, local authority evaluation, deterministic trace hashing, expected artifact pinning, double-run determinism.
+* Simulated: the proposer side is static fixture data standing in for a model-style proposer.
+* Not claimed: production MCP, runtime authority integration, external side effects, or any broader guarantee than this fail-closed local envelope.
+
+## Example commands
+
+```bash
+python demos/claude_tool_authority/run_demo.py
+python demos/claude_tool_authority/run_demo.py --json
+python demos/claude_tool_authority/run_demo.py --update-expected
+pytest -q tests/test_claude_tool_authority_demo.py
+```

demos/claude_tool_authority/__init__.py

@@ -0,0 +1,6 @@
+"""Model-style proposer to CORE tool-authority demo.
+
+This lane proves a narrow boundary: a frontier-proposer fixture may suggest a
+digital action, but only CORE may authorize, ask, refuse, or invalidate it.  The
+output is an inert authority artifact; no proposed action is executed.
+"""

demos/claude_tool_authority/authority.py

@@ -0,0 +1,414 @@
+"""Local deterministic authority substrate for the tool-authority demo.
+
+The proposer contributes a typed request only.  CORE owns validation, envelope
+evaluation, and trace generation.  Even an ``authorized`` decision is inert:
+the returned ``licensed_action`` is a data artifact, not an execution path.
+"""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+import re
+from functools import lru_cache
+from pathlib import Path, PurePosixPath
+from typing import Any, Final
+
+TOOL_NAME: Final[str] = "core.tool_authority.review"
+_HERE: Final[Path] = Path(__file__).resolve().parent
+SCHEMA_PATH: Final[Path] = _HERE / "schema.json"
+
+_SUPPORTED_SCHEMA_KEYS: Final[frozenset[str]] = frozenset(
+    {
+        "type",
+        "properties",
+        "required",
+        "additionalProperties",
+        "enum",
+        "pattern",
+        "minLength",
+        "maxLength",
+        "items",
+    }
+)
+_SCALAR_TYPES: Final[dict[str, type]] = {"string": str, "boolean": bool}
+_ROOT_AUTHORITY: Final[str] = "demos.claude_tool_authority.authority.validate_payload"
+_DECIDE_AUTHORITY: Final[str] = "demos.claude_tool_authority.authority.evaluate_authority"
+_ENVELOPE_AUTHORITY: Final[str] = "demo_tool_authority_envelope(local-v1)"
+_LICENSE_AUTHORITY: Final[str] = "demos.claude_tool_authority.authority.build_license"
+
+_ALLOWED_NOTE_PREFIX: Final[str] = "workspace/demo_notes/"
+_ALLOWED_NOTE_ROOT: Final[PurePosixPath] = PurePosixPath("workspace/demo_notes")
+_PROTECTED_PATH_PREFIXES: Final[tuple[str, ...]] = (
+    ".git/",
+    "chat/",
+    "core/",
+    "docs/",
+    "tests/",
+)
+
+
+def _canonical(payload: dict[str, Any]) -> str:
+    return json.dumps(payload, sort_keys=True, separators=(",", ":"))
+
+
+def _hash_text(text: str) -> str:
+    return hashlib.sha256(text.encode("utf-8")).hexdigest()
+
+
+@lru_cache(maxsize=1)
+def load_schema() -> dict[str, Any]:
+    return json.loads(SCHEMA_PATH.read_text(encoding="utf-8"))
+
+
+def _clip(value: object) -> str:
+    rendered = repr(value)
+    return rendered if len(rendered) <= 80 else rendered[:79] + "…"
+
+
+def _ensure_supported_schema(spec: dict[str, Any], *, path: str) -> None:
+    unsupported = set(spec) - _SUPPORTED_SCHEMA_KEYS
+    if unsupported:
+        raise ValueError(
+            f"{path} uses unsupported schema keywords {sorted(unsupported)}; "
+            "extend the validator before extending the schema"
+        )
+    schema_type = spec.get("type")
+    if isinstance(schema_type, list):
+        return
+    if schema_type == "object":
+        for name, child in spec.get("properties", {}).items():
+            _ensure_supported_schema(child, path=f"{path}.{name}")
+    elif schema_type == "array":
+        _ensure_supported_schema(spec["items"], path=f"{path}[]")
+    elif schema_type not in _SCALAR_TYPES:
+        raise ValueError(f"{path} has unsupported schema type {schema_type!r}")
+
+
+@lru_cache(maxsize=1)
+def _input_schema() -> dict[str, Any]:
+    schema = load_schema()["inputSchema"]
+    _ensure_supported_schema(schema, path="inputSchema")
+    if schema.get("type") != "object" or schema.get("additionalProperties") is not False:
+        raise ValueError("inputSchema must be a closed object")
+    return schema
+
+
+def _validate(spec: dict[str, Any], value: Any, *, path: str, errors: list[str]) -> None:
+    schema_type = spec["type"]
+    if isinstance(schema_type, list):
+        allowed = []
+        for entry in schema_type:
+            if entry == "null" and value is None:
+                return
+            allowed.append(entry)
+        if value is None:
+            errors.append(f"{path} must be one of {allowed}")
+            return
+        non_null = [entry for entry in schema_type if entry != "null"]
+        if len(non_null) != 1:
+            raise ValueError(f"{path} uses unsupported union type {schema_type!r}")
+        schema_type = non_null[0]
+
+    if schema_type == "object":
+        if not isinstance(value, dict):
+            errors.append(f"{path} must be object")
+            return
+        props = spec.get("properties", {})
+        if spec.get("additionalProperties") is False:
+            for unknown in sorted(set(value) - set(props)):
+                errors.append(
+                    f"{path} unexpected property {_clip(unknown)} "
+                    "(additionalProperties is false)"
+                )
+        for required in spec.get("required", []):
+            if required not in value:
+                errors.append(f"{path} missing required property {required!r}")
+        for name, child in props.items():
+            if name in value:
+                _validate(child, value[name], path=f"{path}.{name}", errors=errors)
+        return
+
+    if schema_type == "array":
+        if not isinstance(value, list):
+            errors.append(f"{path} must be array")
+            return
+        for index, item in enumerate(value):
+            _validate(spec["items"], item, path=f"{path}[{index}]", errors=errors)
+        return
+
+    expected = _SCALAR_TYPES[schema_type]
+    if type(value) is not expected:
+        errors.append(f"{path} must be {schema_type}")
+        return
+    if isinstance(value, str):
+        if "minLength" in spec and len(value) < spec["minLength"]:
+            errors.append(f"{path} shorter than {spec['minLength']}")
+        if "maxLength" in spec and len(value) > spec["maxLength"]:
+            errors.append(f"{path} longer than {spec['maxLength']}")
+        if "enum" in spec and value not in spec["enum"]:
+            errors.append(f"{path} must be one of {spec['enum']}")
+        if "pattern" in spec and re.fullmatch(spec["pattern"], value) is None:
+            errors.append(f"{path} does not match {spec['pattern']!r}")
+    elif "enum" in spec and value not in spec["enum"]:
+        errors.append(f"{path} must be one of {spec['enum']}")
+
+
+def validate_payload(payload: Any) -> tuple[str, ...]:
+    errors: list[str] = []
+    _validate(_input_schema(), payload, path="payload", errors=errors)
+    return tuple(errors)
+
+
+def _safe_request_id(payload: Any) -> str | None:
+    request_id = payload.get("request_id") if isinstance(payload, dict) else None
+    if not isinstance(request_id, str):
+        return None
+    pattern = _input_schema()["properties"]["request_id"]["pattern"]
+    return request_id if re.fullmatch(pattern, request_id) else None
+
+
+def _safe_scenario_id(payload: Any) -> str | None:
+    scenario_id = payload.get("scenario_id") if isinstance(payload, dict) else None
+    if not isinstance(scenario_id, str):
+        return None
+    pattern = _input_schema()["properties"]["scenario_id"]["pattern"]
+    return scenario_id if re.fullmatch(pattern, scenario_id) else None
+
+
+def _response_hash(response: dict[str, Any]) -> str:
+    body = dict(response)
+    body.pop("trace_hash", None)
+    return _hash_text(_canonical(body))
+
+
+def _finalize(response: dict[str, Any]) -> dict[str, Any]:
+    response["trace_hash"] = _response_hash(response)
+    return response
+
+
+def _invalid_response(payload: Any, errors: tuple[str, ...]) -> dict[str, Any]:
+    return _finalize(
+        {
+            "tool": TOOL_NAME,
+            "status": "invalid",
+            "request_id": _safe_request_id(payload),
+            "scenario_id": _safe_scenario_id(payload),
+            "authority_path": [_ROOT_AUTHORITY],
+            "decision_reason": "invalid_payload",
+            "trace_summary": {
+                "authority_evaluated": False,
+                "validation_errors": list(errors),
+                "proposer_trace_hash_ignored": bool(
+                    isinstance(payload, dict)
+                    and isinstance(payload.get("proposer"), dict)
+                    and "trace_hash" in payload["proposer"]
+                ),
+            },
+            "invalid_reason": "; ".join(errors),
+        }
+    )
+
+
+def _action_fingerprint(payload: dict[str, Any]) -> str:
+    action = {
+        "action_request": payload["action_request"],
+        "proposer": {
+            "lane": payload["proposer"]["lane"],
+            "model_family": payload["proposer"]["model_family"],
+            "proposal_id": payload["proposer"]["proposal_id"],
+        },
+        "request_id": payload["request_id"],
+        "scenario_id": payload["scenario_id"],
+    }
+    return _hash_text(_canonical(action))
+
+
+def _protected_target(path: str | None) -> bool:
+    if path is None:
+        return False
+    return any(path.startswith(prefix) for prefix in _PROTECTED_PATH_PREFIXES)
+
+
+def _normalize_note_target_path(raw_path: Any) -> tuple[str | None, str | None]:
+    """Canonicalize demo note paths under deterministic POSIX semantics.
+
+    The input is JSON/demo artifact data, not a host filesystem path.  Be
+    conservative: any absolute path, empty path, or ``..`` segment is refused.
+    """
+    if not isinstance(raw_path, str):
+        return None, "malformed_target_path"
+    if raw_path == "":
+        return None, "empty_target_path"
+
+    normalized = PurePosixPath(raw_path)
+    if normalized.is_absolute():
+        return None, "absolute_target_path"
+
+    parts = normalized.parts
+    if not parts:
+        return None, "malformed_target_path"
+    if any(part == ".." for part in parts):
+        return None, "path_traversal_detected"
+
+    cleaned_parts = tuple(part for part in parts if part not in ("", "."))
+    if not cleaned_parts:
+        return None, "malformed_target_path"
+
+    cleaned = PurePosixPath(*cleaned_parts)
+    cleaned_text = cleaned.as_posix()
+    if _protected_target(cleaned_text):
+        return None, "protected_path"
+    if not cleaned_text.startswith(_ALLOWED_NOTE_PREFIX):
+        return None, "outside_authority_envelope"
+    if cleaned == _ALLOWED_NOTE_ROOT:
+        return None, "malformed_target_path"
+    if not cleaned.is_relative_to(_ALLOWED_NOTE_ROOT):
+        return None, "outside_authority_envelope"
+    return cleaned_text, None
+
+
+def _normalized_payload(payload: dict[str, Any]) -> tuple[dict[str, Any], str | None]:
+    """Copy and canonicalize caller payload before authority evaluation."""
+    normalized = copy.deepcopy(payload)
+    action_request = normalized["action_request"]
+    if action_request["action_type"] != "write_local_note":
+        return normalized, None
+
+    target = action_request["target"]
+    normalized_path, refusal_reason = _normalize_note_target_path(target.get("path"))
+    if refusal_reason is not None:
+        return normalized, refusal_reason
+    assert normalized_path is not None
+    target["path"] = normalized_path
+    return normalized, None
+
+
+def build_license(payload: dict[str, Any]) -> dict[str, Any]:
+    target_path = payload["action_request"]["target"]["path"]
+    content = payload["action_request"]["payload"]["content"]
+    return {
+        "action_type": "write_local_note",
+        "target_path": target_path,
+        "content_sha256": _hash_text(content),
+        "authority_scope": "demo.local.low_risk_workspace_note",
+        "effect": "inert_license_only",
+    }
+
+
+def evaluate_authority(payload: dict[str, Any]) -> dict[str, Any]:
+    action_request = payload["action_request"]
+    action_type = action_request["action_type"]
+    target = action_request["target"]
+    payload_body = action_request["payload"]
+    authorities = [_ROOT_AUTHORITY, _DECIDE_AUTHORITY, _ENVELOPE_AUTHORITY]
+    trace_summary = {
+        "authority_evaluated": True,
+        "envelope_version": "local-v1",
+        "action_fingerprint": _action_fingerprint(payload),
+        "proposer_trace_hash_ignored": "trace_hash" in payload["proposer"],
+        "execution_performed": False,
+    }
+
+    if action_type == "write_local_note":
+        target_path = target.get("path")
+        if (
+            target.get("kind") == "workspace_path"
+            and isinstance(target_path, str)
+            and target_path.startswith(_ALLOWED_NOTE_PREFIX)
+            and not _protected_target(target_path)
+            and "title" in payload_body
+            and "content" in payload_body
+        ):
+            authorities.append(_LICENSE_AUTHORITY)
+            return _finalize(
+                {
+                    "tool": TOOL_NAME,
+                    "status": "authorized",
+                    "request_id": payload["request_id"],
+                    "scenario_id": payload["scenario_id"],
+                    "authority_path": authorities,
+                    "decision_reason": "inside_low_risk_envelope",
+                    "trace_summary": trace_summary,
+                    "licensed_action": build_license(payload),
+                }
+            )
+
+    if action_type == "prepare_email_draft":
+        confirmation = action_request.get("confirmation")
+        if not (isinstance(confirmation, dict) and confirmation.get("user_confirmed") is True):
+            return _finalize(
+                {
+                    "tool": TOOL_NAME,
+                    "status": "ask",
+                    "request_id": payload["request_id"],
+                    "scenario_id": payload["scenario_id"],
+                    "authority_path": authorities,
+                    "decision_reason": "missing_explicit_confirmation",
+                    "trace_summary": trace_summary,
+                    "question": (
+                        "CORE requires explicit user confirmation before licensing "
+                        "an external email draft. Confirm the recipient and intent."
+                    ),
+                }
+            )
+
+    refusal_reason = "outside_authority_envelope"
+    if action_type == "shell_command":
+        refusal_reason = "unauthorized_tool"
+    elif action_type == "write_local_note" and _protected_target(target.get("path")):
+        refusal_reason = "protected_path"
+    return _finalize(
+        {
+            "tool": TOOL_NAME,
+            "status": "refused",
+            "request_id": payload["request_id"],
+            "scenario_id": payload["scenario_id"],
+            "authority_path": authorities,
+            "decision_reason": refusal_reason,
+            "trace_summary": trace_summary,
+            "refusal_reason": refusal_reason,
+        }
+    )
+
+
+def run_authority(payload: Any) -> dict[str, Any]:
+    errors = validate_payload(payload)
+    if errors:
+        return _invalid_response(payload, errors)
+    assert isinstance(payload, dict)
+    normalized_payload, refusal_reason = _normalized_payload(payload)
+    if refusal_reason is not None:
+        return _finalize(
+            {
+                "tool": TOOL_NAME,
+                "status": "refused",
+                "request_id": normalized_payload["request_id"],
+                "scenario_id": normalized_payload["scenario_id"],
+                "authority_path": [_ROOT_AUTHORITY, _DECIDE_AUTHORITY, _ENVELOPE_AUTHORITY],
+                "decision_reason": refusal_reason,
+                "trace_summary": {
+                    "authority_evaluated": True,
+                    "envelope_version": "local-v1",
+                    "action_fingerprint": _action_fingerprint(normalized_payload),
+                    "proposer_trace_hash_ignored": "trace_hash"
+                    in normalized_payload["proposer"],
+                    "execution_performed": False,
+                    "path_normalized": False,
+                },
+                "refusal_reason": refusal_reason,
+            }
+        )
+    return evaluate_authority(normalized_payload)
+
+
+__all__ = [
+    "SCHEMA_PATH",
+    "TOOL_NAME",
+    "build_license",
+    "evaluate_authority",
+    "load_schema",
+    "run_authority",
+    "validate_payload",
+]

demos/claude_tool_authority/expected/ask-required-action.json

@@ -0,0 +1,21 @@
+{
+  "authority_path": [
+    "demos.claude_tool_authority.authority.validate_payload",
+    "demos.claude_tool_authority.authority.evaluate_authority",
+    "demo_tool_authority_envelope(local-v1)"
+  ],
+  "decision_reason": "missing_explicit_confirmation",
+  "question": "CORE requires explicit user confirmation before licensing an external email draft. Confirm the recipient and intent.",
+  "request_id": "demo-tool-s2",
+  "scenario_id": "ask-required-action",
+  "status": "ask",
+  "tool": "core.tool_authority.review",
+  "trace_hash": "eeb8ed87e83ed410b852c0e5340a203c0c0445a9001449896cb3026e5332f728",
+  "trace_summary": {
+    "action_fingerprint": "ab384a8903bb4ec28d9ecb410fbe086a9708f7124347eda655c0b9b68e124f7f",
+    "authority_evaluated": true,
+    "envelope_version": "local-v1",
+    "execution_performed": false,
+    "proposer_trace_hash_ignored": false
+  }
+}

demos/claude_tool_authority/expected/authorized-low-risk-local-action.json

@@ -0,0 +1,28 @@
+{
+  "authority_path": [
+    "demos.claude_tool_authority.authority.validate_payload",
+    "demos.claude_tool_authority.authority.evaluate_authority",
+    "demo_tool_authority_envelope(local-v1)",
+    "demos.claude_tool_authority.authority.build_license"
+  ],
+  "decision_reason": "inside_low_risk_envelope",
+  "licensed_action": {
+    "action_type": "write_local_note",
+    "authority_scope": "demo.local.low_risk_workspace_note",
+    "content_sha256": "385d436778e0dd08056a957b10a7acc088dac77ccc2b742f006bfff9d90833dc",
+    "effect": "inert_license_only",
+    "target_path": "workspace/demo_notes/meeting-summary.txt"
+  },
+  "request_id": "demo-tool-s1",
+  "scenario_id": "authorized-low-risk-local-action",
+  "status": "authorized",
+  "tool": "core.tool_authority.review",
+  "trace_hash": "9e797710ed34dfa547fb9a2ac18421054071a7cb000baf9667d3962937e13f16",
+  "trace_summary": {
+    "action_fingerprint": "22b26058c4f227dc17a91223dde60911ccf3e801558b6aee598ff4b1d4f47559",
+    "authority_evaluated": true,
+    "envelope_version": "local-v1",
+    "execution_performed": false,
+    "proposer_trace_hash_ignored": true
+  }
+}

demos/claude_tool_authority/expected/invalid-smuggling-attempt.json

@@ -0,0 +1,20 @@
+{
+  "authority_path": [
+    "demos.claude_tool_authority.authority.validate_payload"
+  ],
+  "decision_reason": "invalid_payload",
+  "invalid_reason": "payload unexpected property 'authorization_status' (additionalProperties is false); payload unexpected property 'licensed_action' (additionalProperties is false)",
+  "request_id": "demo-tool-s4",
+  "scenario_id": "invalid-smuggling-attempt",
+  "status": "invalid",
+  "tool": "core.tool_authority.review",
+  "trace_hash": "a336294778c1f4967d1f9f357fb33ef5f5ce75433a97adbd96a72ecb97ac04fd",
+  "trace_summary": {
+    "authority_evaluated": false,
+    "proposer_trace_hash_ignored": true,
+    "validation_errors": [
+      "payload unexpected property 'authorization_status' (additionalProperties is false)",
+      "payload unexpected property 'licensed_action' (additionalProperties is false)"
+    ]
+  }
+}

demos/claude_tool_authority/expected/refused-outside-envelope-action.json

@@ -0,0 +1,21 @@
+{
+  "authority_path": [
+    "demos.claude_tool_authority.authority.validate_payload",
+    "demos.claude_tool_authority.authority.evaluate_authority",
+    "demo_tool_authority_envelope(local-v1)"
+  ],
+  "decision_reason": "unauthorized_tool",
+  "refusal_reason": "unauthorized_tool",
+  "request_id": "demo-tool-s3",
+  "scenario_id": "refused-outside-envelope-action",
+  "status": "refused",
+  "tool": "core.tool_authority.review",
+  "trace_hash": "fa1d2511f953306f9b0e73dc623aa1a832f408b2a80feae48252b76df2adb90f",
+  "trace_summary": {
+    "action_fingerprint": "08b61128a081cd3f9f028c89f7f6fcb2a6579efddf407c6c10f588f73704b64e",
+    "authority_evaluated": true,
+    "envelope_version": "local-v1",
+    "execution_performed": false,
+    "proposer_trace_hash_ignored": false
+  }
+}

demos/claude_tool_authority/fixtures/ask-required-action.json

@@ -0,0 +1,24 @@
+{
+  "tool": "core.tool_authority.review",
+  "expected_status": "ask",
+  "arguments": {
+    "request_id": "demo-tool-s2",
+    "scenario_id": "ask-required-action",
+    "proposer": {
+      "lane": "frontier_fixture",
+      "model_family": "model_style_proposer",
+      "proposal_id": "proposal-s2"
+    },
+    "action_request": {
+      "action_type": "prepare_email_draft",
+      "target": {
+        "kind": "email_recipient",
+        "recipient": "finance@example.com"
+      },
+      "payload": {
+        "subject": "Budget update",
+        "body": "Draft a budget update email for review."
+      }
+    }
+  }
+}

demos/claude_tool_authority/fixtures/authorized-low-risk-local-action.json

@@ -0,0 +1,25 @@
+{
+  "tool": "core.tool_authority.review",
+  "expected_status": "authorized",
+  "arguments": {
+    "request_id": "demo-tool-s1",
+    "scenario_id": "authorized-low-risk-local-action",
+    "proposer": {
+      "lane": "frontier_fixture",
+      "model_family": "model_style_proposer",
+      "proposal_id": "proposal-s1",
+      "trace_hash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+    },
+    "action_request": {
+      "action_type": "write_local_note",
+      "target": {
+        "kind": "workspace_path",
+        "path": "workspace/demo_notes/meeting-summary.txt"
+      },
+      "payload": {
+        "title": "Meeting summary",
+        "content": "Summarize the reviewed action plan in a local note."
+      }
+    }
+  }
+}

demos/claude_tool_authority/fixtures/invalid-smuggling-attempt.json

@@ -0,0 +1,29 @@
+{
+  "tool": "core.tool_authority.review",
+  "expected_status": "invalid",
+  "arguments": {
+    "request_id": "demo-tool-s4",
+    "scenario_id": "invalid-smuggling-attempt",
+    "proposer": {
+      "lane": "frontier_fixture",
+      "model_family": "model_style_proposer",
+      "proposal_id": "proposal-s4",
+      "trace_hash": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+    },
+    "action_request": {
+      "action_type": "write_local_note",
+      "target": {
+        "kind": "workspace_path",
+        "path": "workspace/demo_notes/should-not-run.txt"
+      },
+      "payload": {
+        "title": "Smuggling attempt",
+        "content": "This should be rejected."
+      }
+    },
+    "licensed_action": {
+      "action_type": "shell_command"
+    },
+    "authorization_status": "authorized"
+  }
+}

demos/claude_tool_authority/fixtures/refused-outside-envelope-action.json

@@ -0,0 +1,21 @@
+{
+  "tool": "core.tool_authority.review",
+  "expected_status": "refused",
+  "arguments": {
+    "request_id": "demo-tool-s3",
+    "scenario_id": "refused-outside-envelope-action",
+    "proposer": {
+      "lane": "frontier_fixture",
+      "model_family": "model_style_proposer",
+      "proposal_id": "proposal-s3"
+    },
+    "action_request": {
+      "action_type": "shell_command",
+      "target": {
+        "kind": "shell",
+        "command": "rm -rf /tmp/demo"
+      },
+      "payload": {}
+    }
+  }
+}

demos/claude_tool_authority/run_demo.py

@@ -0,0 +1,177 @@
+"""Run the frontier-proposer-to-CORE tool-authority demo.
+
+Each fixture is evaluated twice.  The run fails if any scenario drifts from its
+committed expected artifact or if the two executions differ byte-for-byte.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import shutil
+import sys
+import tempfile
+from pathlib import Path
+from typing import Any
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(REPO_ROOT))
+
+from demos.claude_tool_authority.authority import TOOL_NAME, load_schema, run_authority  # noqa: E402
+
+_HERE = Path(__file__).resolve().parent
+FIXTURES_DIR = _HERE / "fixtures"
+EXPECTED_DIR = _HERE / "expected"
+DEFAULT_OUT_DIR = _HERE / "out"
+_TMP_ROOT = Path(tempfile.gettempdir()).resolve()
+
+
+def _render(payload: dict[str, Any]) -> str:
+    return json.dumps(payload, sort_keys=True, indent=2) + "\n"
+
+
+def fixture_paths() -> list[Path]:
+    return sorted(FIXTURES_DIR.glob("*.json"))
+
+
+def expected_path(scenario_id: str) -> Path:
+    return EXPECTED_DIR / f"{scenario_id}.json"
+
+
+def run_fixture(path: Path, out_dir: Path) -> dict[str, Any]:
+    fixture = json.loads(path.read_text(encoding="utf-8"))
+    if fixture["tool"] != TOOL_NAME:
+        raise ValueError(f"{path.name} names {fixture['tool']!r}; expected {TOOL_NAME!r}")
+    run_a = run_authority(fixture["arguments"])
+    run_b = run_authority(fixture["arguments"])
+    if _render(run_a) != _render(run_b):
+        raise AssertionError(f"{path.name} is not deterministic across double-run execution")
+    return run_a
+
+
+def _is_relative_to(path: Path, root: Path) -> bool:
+    try:
+        path.relative_to(root)
+        return True
+    except ValueError:
+        return False
+
+
+def _custom_out_allowed(out_dir: Path) -> bool:
+    resolved = out_dir.resolve()
+    if resolved in {
+        Path("/"),
+        REPO_ROOT.resolve(),
+        _HERE.resolve(),
+        _HERE.parent.resolve(),
+        Path.home().resolve(),
+        Path.cwd().resolve(),
+    }:
+        return False
+    return _is_relative_to(resolved, _HERE.resolve()) or _is_relative_to(
+        resolved, _TMP_ROOT
+    )
+
+
+def _clearable_out_dir(out_dir: Path) -> tuple[bool, str | None]:
+    resolved = out_dir.resolve()
+    default_resolved = DEFAULT_OUT_DIR.resolve()
+    if resolved == default_resolved:
+        return True, None
+    if not _custom_out_allowed(out_dir):
+        return False, (
+            f"refusing to clear {out_dir}: custom out dir must resolve under "
+            f"{_HERE} or {_TMP_ROOT}"
+        )
+    if not (out_dir / "summary.json").exists():
+        return False, (
+            f"refusing to clear {out_dir}: custom out dir requires demo summary.json marker"
+        )
+    return True, None
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--out", type=Path, default=DEFAULT_OUT_DIR)
+    parser.add_argument("--allow-custom-out", action="store_true")
+    parser.add_argument("--json", action="store_true")
+    parser.add_argument("--update-expected", action="store_true")
+    args = parser.parse_args()
+
+    out_dir = args.out
+    if out_dir.resolve() != DEFAULT_OUT_DIR.resolve() and args.allow_custom_out:
+        print(
+            "--allow-custom-out is deprecated; this demo now enforces the output-root policy directly",
+            file=sys.stderr,
+        )
+    if out_dir.exists():
+        allowed, reason = _clearable_out_dir(out_dir)
+        if not allowed:
+            assert reason is not None
+            print(reason, file=sys.stderr)
+            return 2
+        shutil.rmtree(out_dir)
+    out_dir.mkdir(parents=True)
+
+    schema_doc = load_schema()
+    results: list[dict[str, Any]] = []
+    all_passed = True
+
+    for fixture_path in fixture_paths():
+        fixture = json.loads(fixture_path.read_text(encoding="utf-8"))
+        scenario_id = fixture["arguments"]["scenario_id"]
+        scenario_dir = out_dir / scenario_id
+        scenario_dir.mkdir(parents=True, exist_ok=True)
+        response = run_fixture(fixture_path, scenario_dir)
+        artifact = _render(response)
+        (scenario_dir / "response.json").write_text(artifact, encoding="utf-8")
+
+        problems: list[str] = []
+        if response["status"] != fixture["expected_status"]:
+            problems.append(
+                f"status {response['status']!r} != expected {fixture['expected_status']!r}"
+            )
+        ref = expected_path(scenario_id)
+        if args.update_expected:
+            EXPECTED_DIR.mkdir(exist_ok=True)
+            ref.write_text(artifact, encoding="utf-8")
+        elif not ref.exists():
+            problems.append("missing committed expected artifact")
+        elif ref.read_text(encoding="utf-8") != artifact:
+            problems.append("response drifted from committed expected artifact")
+
+        all_passed &= not problems
+        results.append(
+            {
+                "scenario_id": scenario_id,
+                "status": response["status"],
+                "decision_reason": response["decision_reason"],
+                "trace_hash": response["trace_hash"],
+                "passed": not problems,
+                "problems": problems,
+            }
+        )
+
+    summary = {
+        "tool": schema_doc["name"],
+        "all_passed": all_passed,
+        "scenarios": results,
+        "updated_expected": bool(args.update_expected),
+    }
+    summary_text = _render(summary)
+    (out_dir / "summary.json").write_text(summary_text, encoding="utf-8")
+    if args.json:
+        print(summary_text, end="")
+    else:
+        for row in results:
+            mark = "PASS" if row["passed"] else "FAIL"
+            print(f"[{mark}] {row['scenario_id']}: {row['status']} ({row['decision_reason']})")
+            print(f"       trace_hash: {row['trace_hash'][:16]}…")
+            for problem in row["problems"]:
+                print(f"       x {problem}")
+        print(f"{sum(r['passed'] for r in results)}/{len(results)} scenarios passed")
+    return 0 if all_passed else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

demos/claude_tool_authority/schema.json

@@ -0,0 +1,168 @@
+{
+  "name": "core.tool_authority.review",
+  "title": "CORE tool authority review demo",
+  "description": "Submit one model-style proposed digital action to CORE's local deterministic authority substrate. CORE validates a closed MCP-shaped payload, ignores any proposer trace hash, evaluates the proposal against a local authority envelope, and returns authorized | ask | refused | invalid plus a deterministic trace artifact. The proposer cannot set authorization status, cannot provide a licensed action, and cannot execute anything through this contract.",
+  "inputSchema": {
+    "type": "object",
+    "properties": {
+      "request_id": {
+        "type": "string",
+        "pattern": "^[A-Za-z0-9._-]{1,64}$"
+      },
+      "scenario_id": {
+        "type": "string",
+        "pattern": "^[a-z0-9._-]{1,64}$"
+      },
+      "proposer": {
+        "type": "object",
+        "properties": {
+          "lane": {
+            "type": "string",
+            "enum": ["frontier_fixture", "local_fixture"]
+          },
+          "model_family": {
+            "type": "string",
+            "enum": ["model_style_proposer"]
+          },
+          "proposal_id": {
+            "type": "string",
+            "pattern": "^[A-Za-z0-9._-]{1,64}$"
+          },
+          "trace_hash": {
+            "type": "string",
+            "pattern": "^[a-f0-9]{64}$"
+          }
+        },
+        "required": ["lane", "model_family", "proposal_id"],
+        "additionalProperties": false
+      },
+      "action_request": {
+        "type": "object",
+        "properties": {
+          "action_type": {
+            "type": "string",
+            "enum": ["write_local_note", "prepare_email_draft", "shell_command"]
+          },
+          "target": {
+            "type": "object",
+            "properties": {
+              "kind": {
+                "type": "string",
+                "enum": ["workspace_path", "email_recipient", "shell"]
+              },
+              "path": {
+                "type": "string",
+                "minLength": 1,
+                "maxLength": 160
+              },
+              "recipient": {
+                "type": "string",
+                "minLength": 3,
+                "maxLength": 160
+              },
+              "command": {
+                "type": "string",
+                "minLength": 1,
+                "maxLength": 160
+              }
+            },
+            "required": ["kind"],
+            "additionalProperties": false
+          },
+          "payload": {
+            "type": "object",
+            "properties": {
+              "title": {
+                "type": "string",
+                "minLength": 1,
+                "maxLength": 120
+              },
+              "content": {
+                "type": "string",
+                "minLength": 1,
+                "maxLength": 500
+              },
+              "subject": {
+                "type": "string",
+                "minLength": 1,
+                "maxLength": 120
+              },
+              "body": {
+                "type": "string",
+                "minLength": 1,
+                "maxLength": 500
+              }
+            },
+            "required": [],
+            "additionalProperties": false
+          },
+          "confirmation": {
+            "type": "object",
+            "properties": {
+              "user_confirmed": {
+                "type": "boolean"
+              }
+            },
+            "required": ["user_confirmed"],
+            "additionalProperties": false
+          }
+        },
+        "required": ["action_type", "target", "payload"],
+        "additionalProperties": false
+      }
+    },
+    "required": ["request_id", "scenario_id", "proposer", "action_request"],
+    "additionalProperties": false
+  },
+  "outputSchema": {
+    "type": "object",
+    "properties": {
+      "status": {
+        "type": "string",
+        "enum": ["authorized", "ask", "refused", "invalid"]
+      },
+      "request_id": {
+        "type": ["string", "null"]
+      },
+      "scenario_id": {
+        "type": ["string", "null"]
+      },
+      "authority_path": {
+        "type": "array",
+        "items": {
+          "type": "string"
+        }
+      },
+      "decision_reason": {
+        "type": "string"
+      },
+      "trace_hash": {
+        "type": "string"
+      },
+      "trace_summary": {
+        "type": "object"
+      },
+      "licensed_action": {
+        "type": ["object", "null"]
+      },
+      "question": {
+        "type": ["string", "null"]
+      },
+      "refusal_reason": {
+        "type": ["string", "null"]
+      },
+      "invalid_reason": {
+        "type": ["string", "null"]
+      }
+    },
+    "required": [
+      "status",
+      "request_id",
+      "scenario_id",
+      "authority_path",
+      "decision_reason",
+      "trace_hash",
+      "trace_summary"
+    ]
+  }
+}

tests/test_claude_tool_authority_demo.py

@@ -0,0 +1,223 @@
+from __future__ import annotations
+
+import ast
+import copy
+import hashlib
+import json
+from pathlib import Path
+
+import pytest
+
+from demos.claude_tool_authority import authority
+from demos.claude_tool_authority import run_demo as demo_runner
+
+DEMO_DIR = Path(authority.__file__).resolve().parent
+FIXTURES_DIR = DEMO_DIR / "fixtures"
+EXPECTED_DIR = DEMO_DIR / "expected"
+PROTECTED_PATHS = [
+    Path("CLAIMS.md"),
+    Path("chat/runtime.py"),
+    Path("docs/runtime_contracts.md"),
+]
+
+
+def _fixture(name: str) -> dict[str, object]:
+    return json.loads((FIXTURES_DIR / name).read_text(encoding="utf-8"))
+
+
+def _run_arguments(name: str) -> dict[str, object]:
+    fixture = _fixture(name)
+    return authority.run_authority(fixture["arguments"])
+
+
+def _hash_path(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def test_schema_is_closed_recursively():
+    schema = authority.load_schema()["inputSchema"]
+    assert schema["additionalProperties"] is False
+    assert schema["properties"]["proposer"]["additionalProperties"] is False
+    assert schema["properties"]["action_request"]["additionalProperties"] is False
+    assert schema["properties"]["action_request"]["properties"]["target"]["additionalProperties"] is False
+    assert schema["properties"]["action_request"]["properties"]["payload"]["additionalProperties"] is False
+
+
+def test_all_four_scenarios_work():
+    assert _run_arguments("authorized-low-risk-local-action.json")["status"] == "authorized"
+    assert _run_arguments("ask-required-action.json")["status"] == "ask"
+    assert _run_arguments("refused-outside-envelope-action.json")["status"] == "refused"
+    assert _run_arguments("invalid-smuggling-attempt.json")["status"] == "invalid"
+
+
+def test_proposer_cannot_smuggle_licensed_action():
+    response = _run_arguments("invalid-smuggling-attempt.json")
+    assert response["status"] == "invalid"
+    assert "licensed_action" in response["invalid_reason"]
+
+
+def test_proposer_cannot_set_authorization_result():
+    payload = _fixture("authorized-low-risk-local-action.json")["arguments"]
+    payload["status"] = "authorized"
+    response = authority.run_authority(payload)
+    assert response["status"] == "invalid"
+    assert "payload unexpected property 'status'" in response["invalid_reason"]
+
+
+def test_fake_trace_hash_is_ignored_and_regenerated():
+    payload = _fixture("authorized-low-risk-local-action.json")["arguments"]
+    response = authority.run_authority(payload)
+    assert response["status"] == "authorized"
+    assert response["trace_summary"]["proposer_trace_hash_ignored"] is True
+    assert response["trace_hash"] != payload["proposer"]["trace_hash"]
+
+
+def test_unauthorized_tool_refuses():
+    response = _run_arguments("refused-outside-envelope-action.json")
+    assert response["status"] == "refused"
+    assert response["refusal_reason"] == "unauthorized_tool"
+
+
+def test_missing_confirmation_asks():
+    response = _run_arguments("ask-required-action.json")
+    assert response["status"] == "ask"
+    assert response["decision_reason"] == "missing_explicit_confirmation"
+    assert "explicit user confirmation" in response["question"]
+
+
+def test_authorized_only_inside_envelope():
+    response = _run_arguments("authorized-low-risk-local-action.json")
+    assert response["status"] == "authorized"
+    assert response["licensed_action"]["effect"] == "inert_license_only"
+
+    payload = _fixture("authorized-low-risk-local-action.json")["arguments"]
+    payload["action_request"]["target"]["path"] = "docs/runtime_contracts.md"
+    refused = authority.run_authority(payload)
+    assert refused["status"] == "refused"
+    assert refused["refusal_reason"] == "protected_path"
+
+
+@pytest.mark.parametrize(
+    ("target_path", "expected_reason"),
+    [
+        ("workspace/demo_notes/../../core/authority.py", "path_traversal_detected"),
+        ("workspace/demo_notes/../demo_notes/allowed.md", "path_traversal_detected"),
+        ("/absolute/path.md", "absolute_target_path"),
+        ("workspace/demo_notes/subdir/../../../README.md", "path_traversal_detected"),
+    ],
+)
+def test_traversal_and_absolute_paths_are_refused(target_path: str, expected_reason: str):
+    payload = _fixture("authorized-low-risk-local-action.json")["arguments"]
+    original = copy.deepcopy(payload)
+    payload["action_request"]["target"]["path"] = target_path
+    response = authority.run_authority(payload)
+    assert response["status"] == "refused"
+    assert response["refusal_reason"] == expected_reason
+    assert response["decision_reason"] == expected_reason
+    assert response["trace_summary"]["authority_evaluated"] is True
+    assert original["action_request"]["target"]["path"] == "workspace/demo_notes/meeting-summary.txt"
+    assert payload["action_request"]["target"]["path"] == target_path
+
+
+def test_invalid_payload_fails_before_authority_evaluation(monkeypatch):
+    def _explode(_payload: dict[str, object]) -> dict[str, object]:
+        raise AssertionError("authority evaluation ran for invalid payload")
+
+    monkeypatch.setattr(authority, "evaluate_authority", _explode)
+    response = _run_arguments("invalid-smuggling-attempt.json")
+    assert response["status"] == "invalid"
+    assert response["trace_summary"]["authority_evaluated"] is False
+
+
+def test_double_run_outputs_byte_identical(tmp_path):
+    payload = _fixture("authorized-low-risk-local-action.json")["arguments"]
+    run_a = authority.run_authority(payload)
+    run_b = authority.run_authority(payload)
+    assert json.dumps(run_a, sort_keys=True, indent=2) == json.dumps(run_b, sort_keys=True, indent=2)
+
+
+def test_expected_artifact_tampering_fails(tmp_path):
+    scenario_id = "authorized-low-risk-local-action"
+    original = (EXPECTED_DIR / f"{scenario_id}.json").read_text(encoding="utf-8")
+    try:
+        (EXPECTED_DIR / f"{scenario_id}.json").write_text(original.replace("authorized", "refused", 1), encoding="utf-8")
+        argv = ["run_demo.py", "--out", str(tmp_path / "out")]
+        with pytest.MonkeyPatch.context() as patch:
+            patch.setattr("sys.argv", argv)
+            assert demo_runner.main() == 1
+    finally:
+        (EXPECTED_DIR / f"{scenario_id}.json").write_text(original, encoding="utf-8")
+
+
+def test_run_demo_all_passes_against_expected(tmp_path, monkeypatch):
+    out_dir = tmp_path / "demo-out"
+    monkeypatch.setattr("sys.argv", ["run_demo.py", "--out", str(out_dir)])
+    assert demo_runner.main() == 0
+
+
+def test_run_demo_refuses_non_default_dir_without_marker(tmp_path, monkeypatch):
+    out_dir = tmp_path / "missing-marker"
+    out_dir.mkdir()
+    monkeypatch.setattr("sys.argv", ["run_demo.py", "--out", str(out_dir)])
+    assert demo_runner.main() == 2
+
+
+@pytest.mark.parametrize("target", [".", ".."])
+def test_run_demo_refuses_parent_like_paths(target: str, monkeypatch):
+    monkeypatch.setattr("sys.argv", ["run_demo.py", "--out", target])
+    assert demo_runner.main() == 2
+
+
+def test_run_demo_refuses_arbitrary_external_dir_even_with_marker(tmp_path, monkeypatch):
+    external_root = DEMO_DIR.parent / "external-run-demo-root"
+    external_root.mkdir(exist_ok=True)
+    out_dir = external_root / "out"
+    out_dir.mkdir(parents=True)
+    (out_dir / "summary.json").write_text("{}", encoding="utf-8")
+    with pytest.MonkeyPatch.context() as patch:
+        patch.chdir(external_root)
+        patch.setattr("sys.argv", ["run_demo.py", "--out", str(out_dir)])
+        try:
+            assert demo_runner.main() == 2
+        finally:
+            (out_dir / "summary.json").unlink(missing_ok=True)
+            out_dir.rmdir()
+            external_root.rmdir()
+
+
+def test_run_demo_default_out_still_works(monkeypatch):
+    monkeypatch.setattr("sys.argv", ["run_demo.py"])
+    assert demo_runner.main() == 0
+
+
+def test_no_network_subprocess_eval_or_exec_imports_or_calls():
+    forbidden_imports = {"subprocess", "socket", "requests", "httpx", "urllib", "urllib.request"}
+    forbidden_calls = {"eval", "exec", "compile", "open_connection", "create_connection", "Popen", "run"}
+    scanned = [
+        DEMO_DIR / "__init__.py",
+        DEMO_DIR / "authority.py",
+        DEMO_DIR / "run_demo.py",
+    ]
+    for path in scanned:
+        tree = ast.parse(path.read_text(encoding="utf-8"))
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Import):
+                for alias in node.names:
+                    assert alias.name not in forbidden_imports
+            elif isinstance(node, ast.ImportFrom):
+                module = node.module or ""
+                assert module not in forbidden_imports
+            elif isinstance(node, ast.Call):
+                if isinstance(node.func, ast.Name):
+                    assert node.func.id not in forbidden_calls
+                elif isinstance(node.func, ast.Attribute):
+                    assert node.func.attr not in forbidden_calls
+
+
+def test_protected_paths_unchanged(tmp_path, monkeypatch):
+    before = {path: _hash_path(path) for path in PROTECTED_PATHS}
+    out_dir = tmp_path / "demo-out"
+    monkeypatch.setattr("sys.argv", ["run_demo.py", "--out", str(out_dir)])
+    assert demo_runner.main() == 0
+    after = {path: _hash_path(path) for path in PROTECTED_PATHS}
+    assert before == after