Harness foundation: INTENT, ADR-001, architecture, prototype adoption

- INTENT.md: three-layer model (blueprint/instance/harness), single
  shared runtime, never-become boundaries
- ADR-001 (accepted): DEC-2026-002 resolution — one harness repo for
  all projects; instances are declarative state in consuming repos
- docs/architecture.md: components, contracts (manifest, tool
  profiles, completion events, credential lanes), deployment shape
- agent_harness/: executor-worker prototype adopted and renamed
  (6/6 tests green); HARNESS-WP-0001 initial workplan (7 tasks)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

agent_harness/__init__.py

@@ -0,0 +1,10 @@
+"""Thin executor worker (BINKY-WP-0004-T04, DEC-2026-002).
+
+activity-core schedules and emits tasks; this worker executes exactly one
+task per invocation: load persona orientation (kaizen-agentic schedule
+prepare), run a bounded agentic coding session via an llm-connect adapter,
+verify the session committed to the target repo, and report to the
+Custodian State Hub (progress event + optional task close).
+"""
+
+__version__ = "0.1.0"

agent_harness/adapter.py

@@ -0,0 +1,83 @@
+"""Agentic Claude Code adapter.
+
+llm-connect's ClaudeCodeAdapter is a text-generation adapter (`claude
+--print`, no working directory, no tool grants). An executor run needs an
+*agentic* session: file edits and git commits inside the target repo, under
+a hard tool allow-list. This adapter subclasses it, keeping the llm-connect
+LLMAdapter interface so a hosted adapter can be swapped in later, and adds:
+
+- cwd pinned to the target repo
+- --permission-mode acceptEdits
+- an allow-list identical in spirit to binky-control/scripts/rhythm-session.sh:
+  read/edit tools plus local git add/commit/status/log/diff — no push, no
+  network, no arbitrary shell.
+"""
+
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+
+from llm_connect.claude_code import ClaudeCodeAdapter
+from llm_connect.exceptions import LLMSubprocessError, LLMTimeoutError
+from llm_connect.models import LLMResponse, RunConfig
+
+ALLOWED_TOOLS = (
+    "Read,Write,Edit,Glob,Grep,"
+    "Bash(git add:*),Bash(git commit:*),Bash(git status),"
+    "Bash(git log:*),Bash(git diff:*),Bash(date:*),Bash(ls:*)"
+)
+
+
+class AgenticClaudeCodeAdapter(ClaudeCodeAdapter):
+    def __init__(self, workdir: Path, **kwargs):
+        super().__init__(**kwargs)
+        self._workdir = workdir
+
+    def _build_command(self, config: RunConfig) -> list[str]:
+        cmd = [
+            self._cli_path,
+            "--print",
+            "--permission-mode",
+            "acceptEdits",
+            "--allowedTools",
+            ALLOWED_TOOLS,
+        ]
+        if self._model:
+            cmd.extend(["--model", self._model])
+        return cmd
+
+    def execute_prompt(self, prompt: str, config: RunConfig) -> LLMResponse:
+        self._preflight_budget(config)
+        cmd = self._build_command(config)
+        timeout = config.timeout_seconds or self._config.timeout_seconds
+        try:
+            result = subprocess.run(
+                cmd,
+                input=prompt,
+                capture_output=True,
+                text=True,
+                timeout=timeout,
+                cwd=self._workdir,
+            )
+        except subprocess.TimeoutExpired as exc:
+            raise LLMTimeoutError(
+                f"claude CLI timed out after {timeout}s", cause=exc
+            ) from exc
+        if result.returncode != 0:
+            raise LLMSubprocessError(
+                f"claude CLI exited with code {result.returncode}",
+                return_code=result.returncode,
+                stderr=result.stderr,
+            )
+        return LLMResponse(
+            content=result.stdout,
+            model=self._model or "claude-code-cli",
+            usage={},
+            finish_reason="stop",
+            metadata={
+                "provider": "claude-code-agentic",
+                "cli_path": self._cli_path,
+                "workdir": str(self._workdir),
+            },
+        )

agent_harness/cli.py

@@ -0,0 +1,76 @@
+"""CLI: execute exactly one task spec."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+
+from agent_harness.runner import run_task
+from agent_harness.taskspec import TaskSpec, TaskSpecError
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(prog="executor-worker")
+    sub = parser.add_subparsers(dest="command", required=True)
+    run = sub.add_parser("run", help="Execute one task from a JSON spec file")
+    run.add_argument("--task-file", required=True)
+    run.add_argument("--no-hub", action="store_true", help="Skip hub reporting")
+    scan = sub.add_parser(
+        "mail-scan", help="Deterministic company-mailbox scan (no LLM session)"
+    )
+    scan.add_argument("--target-repo", required=True)
+    scan.add_argument("--config", default="integrations/mailbox-binky-company.yml")
+    scan.add_argument("--out", default="mailmeta/reports")
+    scan.add_argument("--no-hub", action="store_true", help="Skip hub reporting")
+    args = parser.parse_args(argv)
+
+    if args.command == "mail-scan":
+        from pathlib import Path
+
+        from agent_harness.mailscan import run_mail_scan
+
+        result = run_mail_scan(
+            target_repo=Path(args.target_repo).expanduser(),
+            config=args.config,
+            out_dir=args.out,
+            report_to_hub=not args.no_hub,
+        )
+        print(
+            json.dumps(
+                {
+                    "ok": result.ok,
+                    "report": result.report_path,
+                    "new_messages": result.new_messages,
+                    "auth_lane": result.auth_lane,
+                    "reason": result.reason,
+                },
+                indent=2,
+            )
+        )
+        return 0 if result.ok else 1
+
+    try:
+        spec = TaskSpec.from_file(args.task_file)
+    except (TaskSpecError, OSError, json.JSONDecodeError) as exc:
+        print(f"invalid task spec: {exc}", file=sys.stderr)
+        return 2
+
+    result = run_task(spec, report_to_hub=not args.no_hub)
+    print(
+        json.dumps(
+            {
+                "ok": result.ok,
+                "committed": result.committed,
+                "head_after": result.head_after,
+                "persona_source": result.persona_source,
+                "reason": result.reason,
+            },
+            indent=2,
+        )
+    )
+    return 0 if result.ok else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

agent_harness/hub.py

@@ -0,0 +1,54 @@
+"""Custodian State Hub reporting (REST, no MCP).
+
+STATE_HUB_URL defaults to the workstation-local hub; on Railiance the
+ops-bridge tunnel exposes it at http://127.0.0.1:18000.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+
+import httpx
+
+_DEFAULT_URL = "http://127.0.0.1:8000"
+_TIMEOUT = 10.0
+
+
+def _base_url() -> str:
+    return os.environ.get("STATE_HUB_URL", _DEFAULT_URL).rstrip("/")
+
+
+def post_progress_event(
+    summary: str,
+    event_type: str,
+    detail: dict[str, Any],
+    task_id: str | None = None,
+) -> bool:
+    payload: dict[str, Any] = {
+        "summary": summary,
+        "event_type": event_type,
+        "detail": detail,
+        "author": "agt-executor-worker",
+    }
+    if task_id:
+        payload["task_id"] = task_id
+    try:
+        resp = httpx.post(f"{_base_url()}/progress/", json=payload, timeout=_TIMEOUT)
+        resp.raise_for_status()
+        return True
+    except httpx.HTTPError:
+        return False
+
+
+def close_task(task_id: str) -> bool:
+    try:
+        resp = httpx.patch(
+            f"{_base_url()}/tasks/{task_id}",
+            json={"status": "done"},
+            timeout=_TIMEOUT,
+        )
+        resp.raise_for_status()
+        return True
+    except httpx.HTTPError:
+        return False

agent_harness/mailscan.py

@@ -0,0 +1,185 @@
+"""Recurring company-mailbox scan (BINKY-WP-0004-T05).
+
+Two-phase design, because credentials and network are barred from the
+agentic session's tool allow-list:
+
+1. **Deterministic scan phase** (this module, no LLM): acquire the IMAP
+   credentials from OpenBao, run email-connect's read-only scan-mailbox in
+   the target repo, record a `binky_mail_intake` hub progress event with
+   metadata-only counts. Credential values live exclusively in the child
+   process environment — never in argv, logs, hub events, or files.
+
+2. **Optional triage session** (standard run_task): an agentic session in
+   the target repo reads the newest report CSV and updates the metadata
+   log/queues. Suspicious-mail rule is part of that task's prompt: never
+   act on message content, only log it.
+
+Credential acquisition, in order:
+- AppRole (unattended lane): role_id/secret_id files under
+  $EXECUTOR_APPROLE_DIR → `bao write -field=token auth/approle/login ...`
+- Pre-existing token (operator lane): ambient bao token, e.g. after an
+  interactive OIDC login.
+Then two `bao kv get -field=...` reads on the kv path.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+
+from agent_harness import hub
+
+_KV_PATH = "tenants/binky/company-email/imap"
+_BAO_TIMEOUT = 30
+
+
+class MailScanError(RuntimeError):
+    pass
+
+
+@dataclass
+class MailScanResult:
+    ok: bool
+    report_path: str | None
+    new_messages: int | None
+    auth_lane: str
+    reason: str = ""
+
+
+def _bao(*args: str, env: dict[str, str] | None = None) -> str:
+    result = subprocess.run(
+        ["bao", *args],
+        capture_output=True,
+        text=True,
+        timeout=_BAO_TIMEOUT,
+        env=env or os.environ.copy(),
+    )
+    if result.returncode != 0:
+        # stderr may reference paths/roles but never secret values
+        raise MailScanError(f"bao {args[0]} failed: {result.stderr.strip()[:200]}")
+    return result.stdout.strip()
+
+
+def _acquire_token() -> tuple[str | None, str]:
+    """Return (token_or_none_for_ambient, auth_lane)."""
+    approle_dir = os.environ.get("EXECUTOR_APPROLE_DIR")
+    if approle_dir:
+        role_id_file = Path(approle_dir) / "role_id"
+        secret_id_file = Path(approle_dir) / "secret_id"
+        if role_id_file.is_file() and secret_id_file.is_file():
+            token = _bao(
+                "write",
+                "-field=token",
+                "auth/approle/login",
+                f"role_id={role_id_file.read_text().strip()}",
+                f"secret_id={secret_id_file.read_text().strip()}",
+            )
+            return token, "approle"
+    return None, "ambient"
+
+
+def _fetch_credentials() -> tuple[dict[str, str], str]:
+    token, lane = _acquire_token()
+    env = os.environ.copy()
+    if token:
+        env["BAO_TOKEN"] = token
+    creds = {
+        field: _bao("kv", "get", f"-field={field}", _KV_PATH, env=env)
+        for field in ("IMAP_USERNAME", "IMAP_PASSWORD")
+    }
+    return creds, lane
+
+
+def run_mail_scan(
+    target_repo: Path,
+    config: str = "integrations/mailbox-binky-company.yml",
+    out_dir: str = "mailmeta/reports",
+    email_connect_src: str | None = None,
+    report_to_hub: bool = True,
+) -> MailScanResult:
+    try:
+        creds, lane = _fetch_credentials()
+    except MailScanError as exc:
+        result = MailScanResult(
+            ok=False, report_path=None, new_messages=None, auth_lane="none",
+            reason=str(exc),
+        )
+        _report(result, report_to_hub, target_repo)
+        return result
+
+    env = os.environ.copy()
+    env.update(creds)
+    if email_connect_src is None:
+        email_connect_src = str(Path.home() / "email-connect" / "src")
+    env["PYTHONPATH"] = email_connect_src
+
+    before = _report_files(target_repo / out_dir)
+    proc = subprocess.run(
+        [
+            "python3", "-m", "email_connect.cli", "scan-mailbox",
+            "--config", config,
+            "--out", out_dir,
+        ],
+        cwd=target_repo,
+        capture_output=True,
+        text=True,
+        timeout=600,
+        env=env,
+    )
+    if proc.returncode != 0:
+        result = MailScanResult(
+            ok=False, report_path=None, new_messages=None, auth_lane=lane,
+            reason=f"email-connect exited {proc.returncode}: {proc.stderr.strip()[:200]}",
+        )
+        _report(result, report_to_hub, target_repo)
+        return result
+
+    new_reports = sorted(_report_files(target_repo / out_dir) - before)
+    report_path = new_reports[-1] if new_reports else None
+    new_messages = _count_rows(target_repo / out_dir / report_path) if report_path else 0
+    result = MailScanResult(
+        ok=True, report_path=report_path, new_messages=new_messages, auth_lane=lane,
+    )
+    _report(result, report_to_hub, target_repo)
+    return result
+
+
+def _report_files(out_dir: Path) -> set[str]:
+    if not out_dir.is_dir():
+        return set()
+    return {p.name for p in out_dir.glob("*.csv")}
+
+
+def _count_rows(path: Path) -> int | None:
+    try:
+        with path.open() as fh:
+            return max(0, sum(1 for _ in fh) - 1)
+    except OSError:
+        return None
+
+
+def _report(result: MailScanResult, report_to_hub: bool, target_repo: Path) -> None:
+    if not report_to_hub:
+        return
+    # Failures must NOT emit binky_mail_intake: the activity-core resolver
+    # treats any such event as "a scan ran" (idempotence guard), so a failed
+    # run reports as a generic executor_run failure instead and the slot
+    # stays due.
+    event_type = "binky_mail_intake" if result.ok else "executor_run"
+    hub.post_progress_event(
+        summary=(
+            f"binky mailbox scan {'ok' if result.ok else 'failed'}"
+            + (f": {result.new_messages} new message(s)" if result.ok else "")
+        ),
+        event_type=event_type,
+        detail={
+            "repo": target_repo.name,
+            "ok": result.ok,
+            "report": result.report_path,
+            "new_messages": result.new_messages,
+            "auth_lane": result.auth_lane,
+            "reason": result.reason,
+        },
+    )

agent_harness/persona.py

@@ -0,0 +1,39 @@
+"""Persona orientation via kaizen-agentic (ADR-005 prepare step).
+
+The ADR-005 contract for a scheduled run is `kaizen-agentic schedule
+prepare <agent>` executed against the target repo: it bundles the agent
+prompt, project memory, metrics summary, and repo pointers from local
+`.kaizen/` state — offline-safe, no hub required. The worker treats the
+bundle as orientation text prepended to the task prompt.
+
+Falls back to an empty bundle when the CLI or the repo's schedule state is
+absent, so a task can still run persona-less (logged in run metadata).
+"""
+
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+
+
+def load_persona_bundle(agent: str, target_repo: Path, timeout: int = 60) -> tuple[str, str]:
+    """Return (bundle_text, source) where source is 'prepare' or 'none'."""
+    try:
+        result = subprocess.run(
+            [
+                "kaizen-agentic",
+                "schedule",
+                "prepare",
+                agent,
+                "--target",
+                str(target_repo),
+            ],
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+        )
+    except (FileNotFoundError, subprocess.TimeoutExpired):
+        return "", "none"
+    if result.returncode != 0 or not result.stdout.strip():
+        return "", "none"
+    return result.stdout, "prepare"

agent_harness/runner.py

@@ -0,0 +1,126 @@
+"""One-task run orchestration.
+
+Flow: lock target repo → snapshot HEAD → persona bundle → prompt → agentic
+session → verify a new commit exists → hub progress event (+ task close).
+The run *fails* if the session pushed anywhere or left the repo dirty in a
+way it should not — the worker never pushes; publishing is a separate,
+explicitly-granted lane (see integrations/executor-worker-secrets.md in
+binky-control).
+"""
+
+from __future__ import annotations
+
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+
+from agent_harness import hub
+from agent_harness.persona import load_persona_bundle
+from agent_harness.taskspec import TaskSpec
+
+PROMPT_TEMPLATE = """\
+You are an unattended executor session (agent persona below, if any).
+Operating rules, non-negotiable:
+- Work ONLY inside the current repository working directory.
+- Green/Blue lane: file edits and local git add/commit only. Never push,
+  never touch the network, never run destructive commands.
+- Bounded effort: complete the single task below, commit with a clear
+  message, then stop. If the task cannot be completed, commit nothing and
+  say why in your final output.
+
+{persona}
+
+## Task: {title}
+
+{description}
+"""
+
+
+@dataclass
+class RunResult:
+    ok: bool
+    committed: bool
+    head_before: str
+    head_after: str
+    persona_source: str
+    session_output: str
+    reason: str = ""
+
+
+def _git(repo: Path, *args: str) -> str:
+    result = subprocess.run(
+        ["git", "-C", str(repo), *args],
+        capture_output=True,
+        text=True,
+        timeout=60,
+    )
+    if result.returncode != 0:
+        raise RuntimeError(f"git {' '.join(args)} failed: {result.stderr.strip()}")
+    return result.stdout.strip()
+
+
+def run_task(spec: TaskSpec, adapter=None, report_to_hub: bool = True) -> RunResult:
+    if adapter is None:
+        from agent_harness.adapter import AgenticClaudeCodeAdapter
+
+        adapter = AgenticClaudeCodeAdapter(workdir=spec.target_repo)
+
+    head_before = _git(spec.target_repo, "rev-parse", "HEAD")
+    persona, persona_source = load_persona_bundle(spec.agent, spec.target_repo)
+    prompt = PROMPT_TEMPLATE.format(
+        persona=persona or "(no persona bundle available for this run)",
+        title=spec.title,
+        description=spec.description,
+    )
+
+    from llm_connect.models import RunConfig
+
+    config = RunConfig(timeout_seconds=spec.timeout_seconds, skip_if_exists=False)
+    try:
+        response = adapter.execute_prompt(prompt, config)
+        session_output = response.content
+        session_ok = True
+        reason = ""
+    except Exception as exc:  # adapter failures must still be reported
+        session_output = ""
+        session_ok = False
+        reason = f"session failed: {exc}"
+
+    head_after = _git(spec.target_repo, "rev-parse", "HEAD")
+    committed = head_after != head_before
+    ok = session_ok and committed
+    if session_ok and not committed:
+        reason = "session completed without committing"
+
+    result = RunResult(
+        ok=ok,
+        committed=committed,
+        head_before=head_before,
+        head_after=head_after,
+        persona_source=persona_source,
+        session_output=session_output,
+        reason=reason,
+    )
+
+    if report_to_hub:
+        detail = {
+            "repo": spec.target_repo.name,
+            "task_title": spec.title,
+            "agent": spec.agent,
+            "labels": spec.labels,
+            "persona_source": persona_source,
+            "committed": committed,
+            "head_after": head_after,
+            "ok": ok,
+            "reason": reason,
+        }
+        hub.post_progress_event(
+            summary=f"executor run: {spec.title} ({'ok' if ok else 'failed'})",
+            event_type=spec.completion_event_type,
+            detail=detail,
+            task_id=spec.hub_task_id,
+        )
+        if ok and spec.hub_task_id:
+            hub.close_task(spec.hub_task_id)
+
+    return result

agent_harness/taskspec.py

@@ -0,0 +1,48 @@
+"""Task specification the worker consumes.
+
+MVP source is a JSON file matching what the activity-core issue-core sink
+emits per rule action (task_template/description/target_repo/labels), plus
+executor-specific fields. A future source polls issue-core directly.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from pathlib import Path
+
+
+class TaskSpecError(ValueError):
+    pass
+
+
+@dataclass
+class TaskSpec:
+    title: str
+    description: str
+    target_repo: Path
+    agent: str = "coach"
+    labels: list[str] = field(default_factory=list)
+    hub_task_id: str | None = None
+    completion_event_type: str = "executor_run"
+    timeout_seconds: int = 900
+
+    @classmethod
+    def from_file(cls, path: str | Path) -> "TaskSpec":
+        raw = json.loads(Path(path).read_text())
+        missing = {"title", "description", "target_repo"} - set(raw)
+        if missing:
+            raise TaskSpecError(f"task spec missing field(s): {', '.join(sorted(missing))}")
+        target = Path(raw["target_repo"]).expanduser()
+        if not (target / ".git").is_dir():
+            raise TaskSpecError(f"target_repo is not a git repository: {target}")
+        return cls(
+            title=str(raw["title"]),
+            description=str(raw["description"]),
+            target_repo=target,
+            agent=str(raw.get("agent", "coach")),
+            labels=[str(label) for label in raw.get("labels", [])],
+            hub_task_id=raw.get("hub_task_id"),
+            completion_event_type=str(raw.get("completion_event_type", "executor_run")),
+            timeout_seconds=int(raw.get("timeout_seconds", 900)),
+        )