feat(security): diff-scope assertion and credential withholding for delegates

src/crossagent/advisors.py

@@ -46,6 +46,13 @@ class Advisor:
     result_parser: str = "text"
     resume_command: tuple[str, ...] | None = None
     session_event_field: str | None = None
+    # Flag that requests a machine-checkable JSON output contract for the
+    # independent verification pass (slice S5). ``None`` means the advisor has no
+    # such contract, so a verifier built on it degrades to parsing a JSON verdict
+    # out of the answer text (D4 graceful degradation — never a hard failure).
+    # Claude exposes ``--json-schema`` (research finding [6]: the payload lands in
+    # ``structured_output``); no other built-in advisor has a verified equivalent.
+    json_schema_flag: str | None = None
     experimental: bool = False
     notes: str = ""
 
@@ -84,6 +91,7 @@ def supports_stream(self) -> bool:
         session_name_flag="--name",
         fork_flag="--fork-session",
         result_parser="claude-stream",
+        json_schema_flag="--json-schema",
     ),
     "codex": Advisor(
         name="codex",

src/crossagent/check.py

@@ -23,7 +23,7 @@
 from dataclasses import dataclass
 from typing import Optional
 
-from .jobs import CheckResultDict
+from .types import CheckResultDict
 
 # Keep only a bounded tail of the check's output — a full test run can emit
 # megabytes, and the whole Job record is loaded into memory (and the dashboard).

src/crossagent/cli.py

@@ -21,6 +21,7 @@
 from . import __version__
 from . import advisors as advisors_mod
 from . import check as check_mod
+from . import credentials as credentials_mod
 from . import jobs as jobs_mod
 from . import parsers as parsers_mod
 from . import registry as reg
@@ -109,10 +110,16 @@ def build_command(
 
 
 def _run_advisor(
-    cmd: list[str], cwd: str | None, parser_name: str
+    cmd: list[str],
+    cwd: str | None,
+    parser_name: str,
+    *,
+    env: dict[str, str] | None = None,
 ) -> tuple[int, parsers_mod.ParsedResult]:
     parser = parsers_mod.get_parser(parser_name)
-    outcome = runner_mod.run(cmd, cwd=cwd, consumer=parser, max_runtime_seconds=None)
+    outcome = runner_mod.run(
+        cmd, cwd=cwd, env=env, consumer=parser, max_runtime_seconds=None
+    )
     parsed = (
         outcome.result
         if isinstance(outcome.result, parsers_mod.ParsedResult)
@@ -199,6 +206,18 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser.add_argument(
         "--registry", default=str(reg.DEFAULT_REGISTRY), help="Session registry path."
     )
+    parser.add_argument(
+        "--pass-env",
+        action="append",
+        default=[],
+        dest="pass_env",
+        help=(
+            "Name of an environment variable to pass through to the advisor even "
+            "though it matches a credential pattern (e.g. the advisor's own API "
+            "key). Repeatable. By default all credential-bearing env vars are "
+            "withheld from the advisor, matching the durable-job path."
+        ),
+    )
     parser.add_argument(
         "--list-advisors", action="store_true", help="Print known advisors and exit."
     )
@@ -278,7 +297,14 @@ def _dispatch(
     registry: dict[str, Any],
     registry_path: Path,
 ) -> int:
-    code, parsed = _run_advisor(cmd, args.cwd, advisor.result_parser)
+    # A foreground advisor is a delegate too: withhold the caller's ambient
+    # credentials (S4 policy), sharing the exact scrub the durable-job path uses
+    # (worker.build_advisor_env delegates to the same helper) so the two dispatch
+    # modes can never drift. ``--pass-env NAME`` is the caller's opt-out.
+    advisor_env = credentials_mod.scrub_env(
+        os.environ, pass_through=getattr(args, "pass_env", [])
+    )
+    code, parsed = _run_advisor(cmd, args.cwd, advisor.result_parser, env=advisor_env)
 
     if parsed.failure:
         error = parsed.error or f"{advisor.name} exited with code {code}"
@@ -361,6 +387,63 @@ def _parse_job_args(subcommand: str, argv: list[str]) -> argparse.Namespace:
                 "failed (default 600)."
             ),
         )
+        parser.add_argument(
+            "--allow-path",
+            action="append",
+            default=None,
+            dest="allow_path",
+            help=(
+                "Declare a path (file or directory subtree) the delegate is "
+                "allowed to modify. Repeatable. When given, crossagent asserts "
+                "after the run that the delegate touched nothing outside the "
+                "declared set (compared by resolved real location, so symlink/.. "
+                "escapes cannot defeat it) and fails the delegation otherwise. "
+                "Omit to leave scope enforcement off."
+            ),
+        )
+        parser.add_argument(
+            "--pass-env",
+            action="append",
+            default=[],
+            dest="pass_env",
+            help=(
+                "Name of an environment variable to pass through to the delegate "
+                "even though it matches a credential pattern (e.g. the advisor's "
+                "own API key). Repeatable. By default all credential-bearing env "
+                "vars are withheld from the delegate."
+            ),
+        )
+        parser.add_argument(
+            "--verify-with",
+            dest="verify_with",
+            help=(
+                "Advisor that independently verifies the delegate's work in a "
+                "FRESH peer session, with the diff/answer supplied as user-turn "
+                "input (removes the implicit-authorship channel that weakens "
+                "self-grading). A failing verdict blocks the green path; a "
+                "prose-only or errored verifier degrades to unverified, never a "
+                "pass. Omit to leave verification off."
+            ),
+        )
+        parser.add_argument(
+            "--verify-model",
+            dest="verify_model",
+            help="Model/alias for the --verify-with advisor (advisor default if omitted).",
+        )
+        parser.add_argument(
+            "--escalate-to",
+            action="append",
+            default=None,
+            dest="escalate_to",
+            metavar="ADVISOR[:MODEL]",
+            help=(
+                "Re-dispatch a FAILED delegation (failing check, scope violation, "
+                "or failing verification) to this larger peer as a same-trace "
+                "child job. Repeatable to form an escalation ladder; each rung is "
+                "tried in turn as the previous fails. Bounded by the nesting-depth "
+                "cap. Omit to leave escalation off."
+            ),
+        )
         parser.add_argument("--json", action="store_true")
         parser.set_defaults(stream=True)
     elif subcommand == "wait":
@@ -631,23 +714,38 @@ def _cmd_result(args: argparse.Namespace) -> int:
 
 def _print_verdict(job: jobs_mod.Job, verdict: str, *, file: Any = sys.stderr) -> None:
     """Print a one-line delegation verdict to *file* (stderr by default)."""
-    check = job.check_result
     if verdict == "verified":
-        print("[crossagent] delegation verified — check passed", file=file)
+        print("[crossagent] delegation verified — all declared gates passed", file=file)
     elif verdict == "unverified":
         print(
-            "[crossagent] delegation UNVERIFIED — no --check ran; the delegate "
-            "finished but its work was not checked",
+            "[crossagent] delegation UNVERIFIED — the delegate finished but no "
+            "gate confirmed its work (no --check/--verify-with, or an "
+            "inconclusive verifier)",
             file=file,
         )
     elif verdict == "failed":
-        exit_code = check.get("exit_code") if check else None
         print(
-            f"[crossagent] delegation FAILED verification — check exited {exit_code}",
+            f"[crossagent] delegation FAILED verification — {_failed_reason(job)}",
             file=file,
         )
 
 
+def _failed_reason(job: jobs_mod.Job) -> str:
+    """Describe why a delegation failed, naming the actual failing gate."""
+    if job.status != jobs_mod.JobState.SUCCEEDED:
+        return f"delegate did not finish cleanly (status {job.status.value})"
+    scope = job.scope_result
+    if scope is not None and scope.get("status") != "ok":
+        return f"scope {scope.get('status')} ({len(scope.get('violating_paths') or [])} path(s))"
+    check = job.check_result
+    if check is not None and check.get("exit_code") != 0:
+        return f"check exited {check.get('exit_code')}"
+    verify = job.verify_result
+    if verify is not None and verify.get("verdict") == "fail":
+        return f"independent verification by {verify.get('advisor')} returned fail"
+    return "a declared gate did not pass"
+
+
 def _metrics_summary(job: jobs_mod.Job) -> str:
     """Return a one-line advisor-metrics summary, or ``""`` when nothing was
     measured. Cost/token/duration go to stderr so piped stdout stays the result.
@@ -869,6 +967,15 @@ def _write_command_info(
         "check_timeout": getattr(
             args, "check_timeout", check_mod.CHECK_DEFAULT_TIMEOUT_SECONDS
         ),
+        # Delegation security posture (S4). ``scope_paths`` is ``None`` when no
+        # --allow-path was given (enforcement off), distinct from an empty list.
+        "scope_paths": getattr(args, "allow_path", None),
+        "pass_env": getattr(args, "pass_env", []),
+        # Independent verification + escalation ladder (S5). ``verify_with`` is
+        # ``None`` when off; ``escalate_to`` is the (possibly empty) rung list.
+        "verify_with": getattr(args, "verify_with", None),
+        "verify_model": getattr(args, "verify_model", None),
+        "escalate_to": getattr(args, "escalate_to", None) or [],
     }
     jobs_mod.atomic_json_write(info, job_dir / "command.json")
 

src/crossagent/credentials.py

@@ -0,0 +1,76 @@
+"""Credential scrubbing for the advisor child environment (slice S4).
+
+A delegate is a partially-untrusted actor (research finding [10]: two-tier
+delegation has a documented path from untrusted repo text to a committed
+backdoor). crossagent must therefore not hand the delegate the caller's ambient
+secrets — a compromised or prompt-injected delegate with a cloud key in its
+environment is a direct exfiltration path.
+
+By default every environment variable whose NAME matches a credential pattern is
+withheld from the child. A caller that genuinely needs one passed through (for
+example the advisor's own API key) opts in explicitly with ``--pass-env NAME``.
+
+Only variable NAMES are ever recorded (in the audit log or the job record). A
+name such as ``AWS_SECRET_ACCESS_KEY`` is not itself a secret, but its VALUE is
+and must never appear in ``events.jsonl``, the redacted command, or any job
+record field.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+
+# Substrings (matched case-insensitively against the variable NAME) that mark a
+# variable as credential-bearing. Curated to catch the common secret shapes
+# without over-matching benign names: ``ACCESS_KEY`` matches
+# ``AWS_ACCESS_KEY_ID`` but a bare ``KEY`` (which would also hit
+# ``KEYBOARD_LAYOUT``) is deliberately not listed.
+_CREDENTIAL_SUBSTRINGS: tuple[str, ...] = (
+    "SECRET",
+    "TOKEN",
+    "PASSWORD",
+    "PASSWD",
+    "CREDENTIAL",
+    "API_KEY",
+    "APIKEY",
+    "ACCESS_KEY",
+    "PRIVATE_KEY",
+)
+
+
+def is_credential_name(name: str) -> bool:
+    """Return True when *name* looks like a credential-bearing env var."""
+    upper = name.upper()
+    return any(marker in upper for marker in _CREDENTIAL_SUBSTRINGS)
+
+
+def withheld_names(
+    env: Mapping[str, str], pass_through: list[str] | None = None
+) -> list[str]:
+    """Return the sorted NAMES of credential vars that would be withheld.
+
+    A name listed in *pass_through* is never withheld. Values are never
+    returned — only names, which are safe to record.
+    """
+    allow = set(pass_through or [])
+    return sorted(
+        name for name in env if name not in allow and is_credential_name(name)
+    )
+
+
+def scrub_env(
+    env: Mapping[str, str], *, pass_through: list[str] | None = None
+) -> dict[str, str]:
+    """Return a copy of *env* with credential-bearing variables removed.
+
+    A variable is kept only when its name is explicitly allow-listed in
+    *pass_through* or does not match any credential pattern. Uses the same
+    :func:`is_credential_name` predicate as :func:`withheld_names`, so the two
+    can never disagree about what was scrubbed.
+    """
+    allow = set(pass_through or [])
+    return {
+        name: value
+        for name, value in env.items()
+        if name in allow or not is_credential_name(name)
+    }