feat: instance manifest, tool profiles, metrics, and budget enforcement

Land HARNESS-WP-0001 T01/T02/T04/T05: extend ADR-005 schedule.yml with
harness fields, named tool-profile registry, ADR-004 metrics writes, and
BudgetTracker wiring. CLI gains validate/profiles; task-file path kept.

agent_harness/__init__.py

@@ -1,10 +1,10 @@
-"""Thin executor worker (BINKY-WP-0004-T04, DEC-2026-002).
+"""Shared agent runtime (DEC-2026-002 / ADR-001).
 
-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).
+Consuming repos declare instances in `.kaizen/schedule.yml`; this package
+runs them: resolve tool profile + budget, load persona orientation
+(kaizen-agentic schedule prepare), run a bounded agentic session via an
+llm-connect adapter, verify the local commit, write `.kaizen/metrics`,
+and report to the Custodian State Hub.
 """
 
 __version__ = "0.1.0"

agent_harness/adapter.py

@@ -8,9 +8,7 @@ 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.
+- allow-list from a named tool profile (default: green-commit-only)
 """
 
 from __future__ import annotations
@@ -22,17 +20,30 @@ 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:*)"
-)
+from agent_harness.profiles import ToolProfile, get_profile
+
+# Backward-compatible alias for the seed profile allow-list string.
+ALLOWED_TOOLS = get_profile("green-commit-only").allowed_tools
 
 
 class AgenticClaudeCodeAdapter(ClaudeCodeAdapter):
-    def __init__(self, workdir: Path, **kwargs):
+    def __init__(
+        self,
+        workdir: Path,
+        *,
+        tool_profile: str | ToolProfile = "green-commit-only",
+        **kwargs,
+    ):
         super().__init__(**kwargs)
         self._workdir = workdir
+        if isinstance(tool_profile, ToolProfile):
+            self._profile = tool_profile
+        else:
+            self._profile = get_profile(tool_profile)
+
+    @property
+    def tool_profile(self) -> ToolProfile:
+        return self._profile
 
     def _build_command(self, config: RunConfig) -> list[str]:
         cmd = [
@@ -41,7 +52,7 @@ class AgenticClaudeCodeAdapter(ClaudeCodeAdapter):
             "--permission-mode",
             "acceptEdits",
             "--allowedTools",
-            ALLOWED_TOOLS,
+            self._profile.allowed_tools,
         ]
         if self._model:
             cmd.extend(["--model", self._model])
@@ -70,7 +81,7 @@ class AgenticClaudeCodeAdapter(ClaudeCodeAdapter):
                 return_code=result.returncode,
                 stderr=result.stderr,
             )
-        return LLMResponse(
+        response = LLMResponse(
             content=result.stdout,
             model=self._model or "claude-code-cli",
             usage={},
@@ -79,5 +90,8 @@ class AgenticClaudeCodeAdapter(ClaudeCodeAdapter):
                 "provider": "claude-code-agentic",
                 "cli_path": self._cli_path,
                 "workdir": str(self._workdir),
+                "tool_profile": self._profile.name,
             },
         )
+        self._consume_budget(config, response)
+        return response

agent_harness/cli.py

@@ -1,21 +1,86 @@
-"""CLI: execute exactly one task spec."""
+"""CLI: validate instance manifests and execute exactly one task."""
 
 from __future__ import annotations
 
 import argparse
 import json
 import sys
+from pathlib import Path
 
+from agent_harness.manifest import (
+    ManifestError,
+    load_manifest_for_repo,
+    manifest_path,
+    validate_manifest,
+)
+from agent_harness.profiles import list_profiles
 from agent_harness.runner import run_task
 from agent_harness.taskspec import TaskSpec, TaskSpecError
 
 
+def _cmd_validate(args: argparse.Namespace) -> int:
+    target = Path(args.target).expanduser().resolve()
+    path = manifest_path(target)
+    if not path.is_file():
+        print(f"error: no instance manifest at {path}", file=sys.stderr)
+        return 2
+    try:
+        manifest = load_manifest_for_repo(target)
+    except ManifestError as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 2
+
+    errors = validate_manifest(
+        manifest, require_harness_fields=bool(args.strict)
+    )
+    if errors:
+        print(f"invalid: {path}", file=sys.stderr)
+        for err in errors:
+            print(f"  - {err}", file=sys.stderr)
+        return 1
+
+    enabled = manifest.enabled_agents()
+    print(f"ok: {path}")
+    print(f"  version: {manifest.version}")
+    if manifest.timezone:
+        print(f"  timezone: {manifest.timezone}")
+    if manifest.harness is not None:
+        print(f"  harness major: {manifest.harness}")
+    print(f"  agents: {len(manifest.agents)} ({len(enabled)} enabled)")
+    for agent in manifest.agents:
+        flags = []
+        if not agent.enabled:
+            flags.append("disabled")
+        if agent.tool_profile:
+            flags.append(f"profile={agent.tool_profile}")
+        if agent.lane:
+            flags.append(f"lane={agent.lane}")
+        if agent.budget is not None:
+            flags.append(f"budget={agent.budget}")
+        suffix = f" [{', '.join(flags)}]" if flags else ""
+        print(f"    - {agent.name}: {agent.cadence}{suffix}")
+    return 0
+
+
+def _cmd_profiles(_args: argparse.Namespace) -> int:
+    for profile in list_profiles():
+        print(f"{profile.name}\tlane={profile.lane}\t{profile.description}")
+    return 0
+
+
 def main(argv: list[str] | None = None) -> int:
-    parser = argparse.ArgumentParser(prog="executor-worker")
+    parser = argparse.ArgumentParser(prog="agent-harness")
     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")
+    run.add_argument(
+        "--no-metrics",
+        action="store_true",
+        help="Skip writing .kaizen/metrics in the target repo",
+    )
+
     scan = sub.add_parser(
         "mail-scan", help="Deterministic company-mailbox scan (no LLM session)"
     )
@@ -23,11 +88,33 @@ def main(argv: list[str] | None = None) -> int:
     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")
+
+    validate = sub.add_parser(
+        "validate",
+        help="Validate instance manifest (.kaizen/schedule.yml + harness fields)",
+    )
+    validate.add_argument(
+        "--target",
+        default=".",
+        help="Consuming repo root (default: cwd)",
+    )
+    validate.add_argument(
+        "--strict",
+        action="store_true",
+        help="Require lane, tool_profile, and harness pin on every enabled agent",
+    )
+
+    sub.add_parser("profiles", help="List named tool profiles")
+
     args = parser.parse_args(argv)
 
-    if args.command == "mail-scan":
-        from pathlib import Path
+    if args.command == "validate":
+        return _cmd_validate(args)
 
+    if args.command == "profiles":
+        return _cmd_profiles(args)
+
+    if args.command == "mail-scan":
         from agent_harness.mailscan import run_mail_scan
 
         result = run_mail_scan(
@@ -56,7 +143,11 @@ def main(argv: list[str] | None = None) -> int:
         print(f"invalid task spec: {exc}", file=sys.stderr)
         return 2
 
-    result = run_task(spec, report_to_hub=not args.no_hub)
+    result = run_task(
+        spec,
+        report_to_hub=not args.no_hub,
+        write_metrics=not args.no_metrics,
+    )
     print(
         json.dumps(
             {
@@ -64,6 +155,10 @@ def main(argv: list[str] | None = None) -> int:
                 "committed": result.committed,
                 "head_after": result.head_after,
                 "persona_source": result.persona_source,
+                "tool_profile": result.tool_profile,
+                "budget_tokens": result.budget_tokens,
+                "tokens_spent": result.tokens_spent,
+                "execution_time_s": round(result.execution_time_s, 3),
                 "reason": result.reason,
             },
             indent=2,

agent_harness/hub.py

@@ -52,3 +52,48 @@ def close_task(task_id: str) -> bool:
         return True
     except httpx.HTTPError:
         return False
+
+
+def post_token_event(
+    repo: str,
+    tokens: int,
+    *,
+    budget: int | None = None,
+    agent: str | None = None,
+    ok: bool = True,
+    detail: dict[str, Any] | None = None,
+) -> bool:
+    """Best-effort token cost event for the hub Token Cost dashboard.
+
+    Schema is intentionally loose: if the hub rejects the payload we
+    swallow the error so a metrics-schema drift never fails a run.
+    """
+    payload: dict[str, Any] = {
+        "repo": repo,
+        "tokens": tokens,
+        "source": "agent-harness",
+        "ok": ok,
+    }
+    if budget is not None:
+        payload["budget"] = budget
+    if agent is not None:
+        payload["agent"] = agent
+    if detail:
+        payload["detail"] = detail
+    try:
+        resp = httpx.post(
+            f"{_base_url()}/token-events/upsert",
+            json=payload,
+            timeout=_TIMEOUT,
+        )
+        if resp.status_code >= 400:
+            # Fallback shape used by some hub builds
+            resp = httpx.post(
+                f"{_base_url()}/token-events/",
+                json=payload,
+                timeout=_TIMEOUT,
+            )
+        resp.raise_for_status()
+        return True
+    except httpx.HTTPError:
+        return False

agent_harness/manifest.py

@@ -0,0 +1,301 @@
+"""Instance manifest — declarative agent instances in consuming repos.
+
+Extends ADR-005 `.kaizen/schedule.yml` with harness fields. kaizen-agentic
+owns the base keys (version, timezone, agents.<name>.{cadence,cron,enabled});
+agent-harness owns the extension keys (lane, tool_profile, budget, harness,
+optional blueprint). Same file — no sibling manifest unless kaizen owners
+later prefer separation.
+
+See docs/instance-manifest.md for the full contract.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+from agent_harness.profiles import PROFILES, UnknownToolProfileError, get_profile
+
+MANIFEST_RELATIVE_PATH = Path(".kaizen") / "schedule.yml"
+MANIFEST_VERSION = "1"
+VALID_CADENCES = ("daily", "weekly", "monthly")
+VALID_LANES = ("green", "blue")
+# Package major this harness release implements (0.x → 0).
+HARNESS_MAJOR = 0
+
+
+class ManifestError(ValueError):
+    """Structural failure loading or parsing a manifest."""
+
+
+@dataclass
+class AgentInstance:
+    """One agent instance declared in a consuming repo."""
+
+    name: str
+    cadence: str
+    enabled: bool = True
+    cron: str | None = None
+    # Harness extension fields (optional in file; defaults applied at load).
+    blueprint: str | None = None  # defaults to name
+    lane: str | None = None
+    tool_profile: str | None = None
+    budget: int | None = None  # tokens per run; None = unlimited for this field
+    harness: int | None = None  # pinned major; None inherits repo-level
+
+    @property
+    def blueprint_name(self) -> str:
+        return self.blueprint or self.name
+
+
+@dataclass
+class InstanceManifest:
+    """Parsed instance manifest (schedule.yml + harness extensions)."""
+
+    version: str
+    timezone: str | None
+    harness: int | None  # repo-level pinned harness major
+    agents: list[AgentInstance] = field(default_factory=list)
+    source_path: Path | None = None
+
+    def agent_for(self, name: str) -> AgentInstance | None:
+        for agent in self.agents:
+            if agent.name == name:
+                return agent
+        return None
+
+    def enabled_agents(self) -> list[AgentInstance]:
+        return [a for a in self.agents if a.enabled]
+
+    def effective_harness(self, agent: AgentInstance) -> int | None:
+        if agent.harness is not None:
+            return agent.harness
+        return self.harness
+
+
+def manifest_path(project_root: Path) -> Path:
+    return Path(project_root) / MANIFEST_RELATIVE_PATH
+
+
+def _parse_budget(raw: Any, agent_name: str) -> int | None:
+    if raw is None:
+        return None
+    if isinstance(raw, bool) or not isinstance(raw, int):
+        raise ManifestError(
+            f"agent '{agent_name}': budget must be a positive integer (tokens/run)"
+        )
+    if raw <= 0:
+        raise ManifestError(
+            f"agent '{agent_name}': budget must be a positive integer (tokens/run)"
+        )
+    return raw
+
+
+def _parse_harness_major(raw: Any, context: str) -> int | None:
+    if raw is None:
+        return None
+    if isinstance(raw, bool) or not isinstance(raw, int):
+        raise ManifestError(f"{context}: harness must be a non-negative integer major")
+    if raw < 0:
+        raise ManifestError(f"{context}: harness must be a non-negative integer major")
+    return raw
+
+
+def parse_manifest(data: Any, source_path: Path | None = None) -> InstanceManifest:
+    """Parse a raw mapping into InstanceManifest. Structural errors raise."""
+    if not isinstance(data, dict):
+        raise ManifestError("manifest must be a YAML mapping at the top level")
+
+    version = data.get("version")
+    if version is None:
+        raise ManifestError("missing required key: version")
+    version = str(version)
+
+    timezone = data.get("timezone")
+    if timezone is not None and not isinstance(timezone, str):
+        raise ManifestError("timezone must be a string")
+
+    repo_harness = _parse_harness_major(data.get("harness"), "top-level")
+
+    agents_raw = data.get("agents", {})
+    if not isinstance(agents_raw, dict):
+        raise ManifestError("agents must be a mapping of agent-name -> settings")
+
+    agents: list[AgentInstance] = []
+    for name, settings in agents_raw.items():
+        if settings is None:
+            settings = {}
+        if not isinstance(settings, dict):
+            raise ManifestError(f"agent '{name}' settings must be a mapping")
+
+        cron = settings.get("cron")
+        if cron is not None and not isinstance(cron, str):
+            raise ManifestError(f"agent '{name}' cron must be a string")
+
+        blueprint = settings.get("blueprint")
+        if blueprint is not None and not isinstance(blueprint, str):
+            raise ManifestError(f"agent '{name}' blueprint must be a string")
+
+        lane = settings.get("lane")
+        if lane is not None and not isinstance(lane, str):
+            raise ManifestError(f"agent '{name}' lane must be a string")
+
+        tool_profile = settings.get("tool_profile")
+        if tool_profile is not None and not isinstance(tool_profile, str):
+            raise ManifestError(f"agent '{name}' tool_profile must be a string")
+
+        agents.append(
+            AgentInstance(
+                name=str(name),
+                cadence=str(settings.get("cadence", "")),
+                enabled=bool(settings.get("enabled", True)),
+                cron=cron,
+                blueprint=blueprint,
+                lane=lane,
+                tool_profile=tool_profile,
+                budget=_parse_budget(settings.get("budget"), str(name)),
+                harness=_parse_harness_major(
+                    settings.get("harness"), f"agent '{name}'"
+                ),
+            )
+        )
+
+    return InstanceManifest(
+        version=version,
+        timezone=timezone,
+        harness=repo_harness,
+        agents=agents,
+        source_path=source_path,
+    )
+
+
+def load_manifest(path: Path | str) -> InstanceManifest:
+    path = Path(path)
+    if not path.exists():
+        raise ManifestError(f"manifest not found: {path}")
+    try:
+        raw = yaml.safe_load(path.read_text(encoding="utf-8"))
+    except yaml.YAMLError as exc:
+        raise ManifestError(f"invalid YAML in {path}: {exc}") from exc
+    return parse_manifest(raw, source_path=path)
+
+
+def load_manifest_for_repo(project_root: Path | str) -> InstanceManifest:
+    return load_manifest(manifest_path(Path(project_root)))
+
+
+def validate_manifest(
+    manifest: InstanceManifest,
+    *,
+    require_harness_fields: bool = False,
+) -> list[str]:
+    """Return human-readable validation errors (empty == valid).
+
+    When *require_harness_fields* is False (default), ADR-005-only manifests
+    are valid: harness extension fields are checked only when present. When
+    True, every enabled agent must declare lane, tool_profile, and an
+    effective harness pin (repo- or agent-level).
+    """
+    errors: list[str] = []
+
+    if manifest.version != MANIFEST_VERSION:
+        errors.append(
+            f"unsupported version '{manifest.version}' "
+            f"(expected '{MANIFEST_VERSION}')"
+        )
+
+    if not manifest.agents:
+        errors.append("no agents declared under 'agents:'")
+
+    seen: set[str] = set()
+    for agent in manifest.agents:
+        if agent.name in seen:
+            errors.append(f"duplicate agent entry: {agent.name}")
+        seen.add(agent.name)
+
+        if agent.cadence not in VALID_CADENCES:
+            errors.append(
+                f"agent '{agent.name}': invalid cadence '{agent.cadence}' "
+                f"(expected one of {', '.join(VALID_CADENCES)})"
+            )
+
+        if agent.lane is not None and agent.lane not in VALID_LANES:
+            errors.append(
+                f"agent '{agent.name}': invalid lane '{agent.lane}' "
+                f"(expected one of {', '.join(VALID_LANES)})"
+            )
+
+        if agent.tool_profile is not None:
+            try:
+                profile = get_profile(agent.tool_profile)
+            except UnknownToolProfileError as exc:
+                errors.append(f"agent '{agent.name}': {exc}")
+            else:
+                if agent.lane is not None and agent.lane != profile.lane:
+                    errors.append(
+                        f"agent '{agent.name}': lane '{agent.lane}' does not "
+                        f"match tool_profile '{profile.name}' (lane={profile.lane})"
+                    )
+
+        if require_harness_fields and agent.enabled:
+            if agent.lane is None:
+                errors.append(
+                    f"agent '{agent.name}': lane is required for harness runs "
+                    f"(green|blue)"
+                )
+            if agent.tool_profile is None:
+                errors.append(
+                    f"agent '{agent.name}': tool_profile is required for harness runs"
+                )
+            if manifest.effective_harness(agent) is None:
+                errors.append(
+                    f"agent '{agent.name}': harness major pin required "
+                    f"(set top-level harness: or agents.{agent.name}.harness)"
+                )
+
+        effective = manifest.effective_harness(agent)
+        if effective is not None and effective != HARNESS_MAJOR:
+            # Soft pin check: warn-style error so validate fails closed for
+            # mismatched majors (instances must upgrade deliberately).
+            errors.append(
+                f"agent '{agent.name}': pinned harness major {effective} "
+                f"does not match this runtime (major {HARNESS_MAJOR})"
+            )
+
+    return errors
+
+
+def resolve_run_policy(
+    project_root: Path | str,
+    agent_name: str,
+    *,
+    default_profile: str = "green-commit-only",
+) -> tuple[str, int | None, str | None]:
+    """Resolve (tool_profile_name, budget_tokens, lane) for a run.
+
+    If the repo has no manifest or no entry for *agent_name*, returns the
+    default profile with no budget. If the entry names an unknown profile,
+    raises UnknownToolProfileError (refuse to run).
+    """
+    root = Path(project_root)
+    path = manifest_path(root)
+    if not path.exists():
+        get_profile(default_profile)  # validate default exists
+        return default_profile, None, None
+
+    manifest = load_manifest(path)
+    instance = manifest.agent_for(agent_name)
+    if instance is None or not instance.enabled:
+        get_profile(default_profile)
+        return default_profile, None, None
+
+    profile_name = instance.tool_profile or default_profile
+    get_profile(profile_name)  # raises if unknown
+    return profile_name, instance.budget, instance.lane
+
+
+def known_profile_names() -> list[str]:
+    return sorted(PROFILES)

agent_harness/metrics.py

@@ -0,0 +1,160 @@
+"""Write per-run records into the target repo's `.kaizen/metrics` tree.
+
+Follows kaizen-agentic ADR-004 conventions so the optimization loop can
+observe harness-run agents:
+
+  .kaizen/metrics/<agent>/
+    executions.jsonl   # append-only
+    summary.json       # regenerated on write
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+
+@dataclass
+class ExecutionRecord:
+    timestamp: str
+    agent: str
+    success: bool
+    execution_time_s: float = 0.0
+    session_id: str | None = None
+    quality_score: float | None = None
+    primary_metric: dict[str, Any] | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+    # Helix / harness correlation (ADR-004 optional fields)
+    repo: str | None = None
+    tokens: int | None = None
+    committed: bool | None = None
+    head_after: str | None = None
+    reason: str | None = None
+    harness: str = "agent-harness"
+
+    def to_json_line(self) -> str:
+        data = asdict(self)
+        # Drop Nones for a compact record; required fields always present.
+        compact = {k: v for k, v in data.items() if v is not None}
+        return json.dumps(compact, sort_keys=True)
+
+
+def metrics_dir(project_root: Path, agent: str) -> Path:
+    return Path(project_root) / ".kaizen" / "metrics" / agent
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace(
+        "+00:00", "Z"
+    )
+
+
+def _load_executions(path: Path) -> list[dict[str, Any]]:
+    if not path.is_file():
+        return []
+    records: list[dict[str, Any]] = []
+    for line in path.read_text(encoding="utf-8").splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            records.append(json.loads(line))
+        except json.JSONDecodeError:
+            continue
+    return records
+
+
+def _trend(values: list[float]) -> str:
+    if len(values) < 4:
+        return "stable"
+    mid = len(values) // 2
+    early = sum(values[:mid]) / max(1, mid)
+    late = sum(values[mid:]) / max(1, len(values) - mid)
+    if late - early > 0.05:
+        return "up"
+    if early - late > 0.05:
+        return "down"
+    return "stable"
+
+
+def regenerate_summary(agent: str, executions: list[dict[str, Any]]) -> dict[str, Any]:
+    count = len(executions)
+    successes = [e for e in executions if e.get("success")]
+    success_rate = (len(successes) / count) if count else 0.0
+    times = [float(e["execution_time_s"]) for e in executions if "execution_time_s" in e]
+    qualities = [
+        float(e["quality_score"])
+        for e in executions
+        if isinstance(e.get("quality_score"), (int, float))
+    ]
+    last_ts = executions[-1].get("timestamp") if executions else None
+    return {
+        "agent": agent,
+        "execution_count": count,
+        "success_rate": round(success_rate, 3),
+        "avg_quality_score": (
+            round(sum(qualities) / len(qualities), 3) if qualities else None
+        ),
+        "avg_execution_time_s": (
+            round(sum(times) / len(times), 3) if times else None
+        ),
+        "last_execution": last_ts,
+        "trend": {
+            "success_rate": _trend(
+                [1.0 if e.get("success") else 0.0 for e in executions]
+            ),
+            "quality_score": _trend(qualities) if qualities else "stable",
+        },
+    }
+
+
+def record_execution(
+    project_root: Path | str,
+    agent: str,
+    *,
+    success: bool,
+    execution_time_s: float = 0.0,
+    tokens: int | None = None,
+    committed: bool | None = None,
+    head_after: str | None = None,
+    reason: str | None = None,
+    metadata: dict[str, Any] | None = None,
+    session_id: str | None = None,
+) -> Path:
+    """Append one execution record and regenerate summary.json.
+
+    Returns the path to executions.jsonl. Never raises for missing parent
+    dirs (creates them). Callers that must not write should skip this.
+    """
+    root = Path(project_root)
+    directory = metrics_dir(root, agent)
+    directory.mkdir(parents=True, exist_ok=True)
+
+    record = ExecutionRecord(
+        timestamp=_utc_now(),
+        agent=agent,
+        success=success,
+        execution_time_s=float(execution_time_s),
+        session_id=session_id,
+        metadata=metadata or {},
+        repo=root.name,
+        tokens=tokens,
+        committed=committed,
+        head_after=head_after,
+        reason=reason,
+    )
+
+    executions_path = directory / "executions.jsonl"
+    with executions_path.open("a", encoding="utf-8") as fh:
+        fh.write(record.to_json_line() + "\n")
+
+    all_records = _load_executions(executions_path)
+    summary = regenerate_summary(agent, all_records)
+    (directory / "summary.json").write_text(
+        json.dumps(summary, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+    return executions_path

agent_harness/profiles.py

@@ -0,0 +1,78 @@
+"""Named tool-profile registry.
+
+Instances declare a profile by name in the instance manifest; the harness
+resolves and enforces the allow-list. Instances never enumerate tools.
+Unknown profile names refuse to run.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+class UnknownToolProfileError(ValueError):
+    """Raised when a manifest references a profile that is not registered."""
+
+
+# Claude Code --allowedTools strings. No push, no network, no arbitrary shell.
+_GREEN_COMMIT_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:*)"
+)
+
+# Blue-lane mail triage session: same session tools as green-commit-only.
+# Credentialed IMAP scan is a deterministic pre-step outside the session
+# (see mailscan.py); the session only reads reports and updates queues.
+_BLUE_MAIL_TRIAGE_TOOLS = _GREEN_COMMIT_TOOLS
+
+
+@dataclass(frozen=True)
+class ToolProfile:
+    """A named hard allow-list for agentic sessions."""
+
+    name: str
+    description: str
+    allowed_tools: str
+    lane: str  # green | blue — advisory; enforcement is the allow-list
+
+
+PROFILES: dict[str, ToolProfile] = {
+    "green-commit-only": ToolProfile(
+        name="green-commit-only",
+        description=(
+            "Green-lane local commit: read/edit tools plus git add/commit/"
+            "status/log/diff. No push, no network, no arbitrary shell."
+        ),
+        allowed_tools=_GREEN_COMMIT_TOOLS,
+        lane="green",
+    ),
+    "blue-mail-triage": ToolProfile(
+        name="blue-mail-triage",
+        description=(
+            "Blue-lane mail triage session after a credentialed deterministic "
+            "scan: same local commit tools as green-commit-only. Credentials "
+            "and network stay outside the agentic session."
+        ),
+        allowed_tools=_BLUE_MAIL_TRIAGE_TOOLS,
+        lane="blue",
+    ),
+}
+
+
+def get_profile(name: str) -> ToolProfile:
+    """Resolve a profile by name. Raises UnknownToolProfileError if missing."""
+    key = (name or "").strip()
+    if not key:
+        raise UnknownToolProfileError("tool_profile name is empty")
+    profile = PROFILES.get(key)
+    if profile is None:
+        known = ", ".join(sorted(PROFILES))
+        raise UnknownToolProfileError(
+            f"unknown tool_profile '{key}' (known: {known})"
+        )
+    return profile
+
+
+def list_profiles() -> list[ToolProfile]:
+    return [PROFILES[k] for k in sorted(PROFILES)]

agent_harness/runner.py

@@ -1,21 +1,24 @@
 """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).
+Flow: lock target repo → resolve tool profile / budget from instance
+manifest → snapshot HEAD → persona bundle → prompt → agentic session →
+verify a new commit exists → kaizen metrics + 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.
 """
 
 from __future__ import annotations
 
 import subprocess
+import time
 from dataclasses import dataclass
 from pathlib import Path
 
-from agent_harness import hub
+from agent_harness import hub, metrics
+from agent_harness.manifest import resolve_run_policy
 from agent_harness.persona import load_persona_bundle
+from agent_harness.profiles import UnknownToolProfileError, get_profile
 from agent_harness.taskspec import TaskSpec
 
 PROMPT_TEMPLATE = """\
@@ -24,6 +27,7 @@ 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.
+- Tool profile for this run: {tool_profile} (lane={lane}).
 - 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.
@@ -45,6 +49,10 @@ class RunResult:
     persona_source: str
     session_output: str
     reason: str = ""
+    tool_profile: str = ""
+    budget_tokens: int | None = None
+    tokens_spent: int | None = None
+    execution_time_s: float = 0.0
 
 
 def _git(repo: Path, *args: str) -> str:
@@ -59,11 +67,49 @@ def _git(repo: Path, *args: str) -> str:
     return result.stdout.strip()
 
 
-def run_task(spec: TaskSpec, adapter=None, report_to_hub: bool = True) -> RunResult:
+def run_task(
+    spec: TaskSpec,
+    adapter=None,
+    report_to_hub: bool = True,
+    write_metrics: bool = True,
+) -> RunResult:
+    try:
+        profile_name, budget_tokens, lane = resolve_run_policy(
+            spec.target_repo, spec.agent
+        )
+        profile = get_profile(profile_name)
+    except UnknownToolProfileError as exc:
+        return RunResult(
+            ok=False,
+            committed=False,
+            head_before="",
+            head_after="",
+            persona_source="none",
+            session_output="",
+            reason=f"refused: {exc}",
+            tool_profile="",
+            budget_tokens=None,
+        )
+    except Exception as exc:
+        return RunResult(
+            ok=False,
+            committed=False,
+            head_before="",
+            head_after="",
+            persona_source="none",
+            session_output="",
+            reason=f"manifest resolution failed: {exc}",
+            tool_profile="",
+            budget_tokens=None,
+        )
+
     if adapter is None:
         from agent_harness.adapter import AgenticClaudeCodeAdapter
 
-        adapter = AgenticClaudeCodeAdapter(workdir=spec.target_repo)
+        adapter = AgenticClaudeCodeAdapter(
+            workdir=spec.target_repo,
+            tool_profile=profile,
+        )
 
     head_before = _git(spec.target_repo, "rev-parse", "HEAD")
     persona, persona_source = load_persona_bundle(spec.agent, spec.target_repo)
@@ -71,20 +117,30 @@ def run_task(spec: TaskSpec, adapter=None, report_to_hub: bool = True) -> RunRes
         persona=persona or "(no persona bundle available for this run)",
         title=spec.title,
         description=spec.description,
+        tool_profile=profile.name,
+        lane=lane or profile.lane,
     )
 
-    from llm_connect.models import RunConfig
+    from llm_connect.models import BudgetTracker, RunConfig
 
-    config = RunConfig(timeout_seconds=spec.timeout_seconds, skip_if_exists=False)
+    budget_tracker = BudgetTracker(total=budget_tokens) if budget_tokens else None
+    config = RunConfig(
+        timeout_seconds=spec.timeout_seconds,
+        skip_if_exists=False,
+        budget_tracker=budget_tracker,
+    )
+
+    started = time.monotonic()
     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
+    except Exception as exc:  # adapter / budget failures must still be reported
         session_output = ""
         session_ok = False
         reason = f"session failed: {exc}"
+    execution_time_s = time.monotonic() - started
 
     head_after = _git(spec.target_repo, "rev-parse", "HEAD")
     committed = head_after != head_before
@@ -92,6 +148,8 @@ def run_task(spec: TaskSpec, adapter=None, report_to_hub: bool = True) -> RunRes
     if session_ok and not committed:
         reason = "session completed without committing"
 
+    tokens_spent = budget_tracker.spent if budget_tracker is not None else None
+
     result = RunResult(
         ok=ok,
         committed=committed,
@@ -100,8 +158,33 @@ def run_task(spec: TaskSpec, adapter=None, report_to_hub: bool = True) -> RunRes
         persona_source=persona_source,
         session_output=session_output,
         reason=reason,
+        tool_profile=profile.name,
+        budget_tokens=budget_tokens,
+        tokens_spent=tokens_spent,
+        execution_time_s=execution_time_s,
     )
 
+    if write_metrics:
+        try:
+            metrics.record_execution(
+                spec.target_repo,
+                spec.agent,
+                success=ok,
+                execution_time_s=execution_time_s,
+                tokens=tokens_spent,
+                committed=committed,
+                head_after=head_after,
+                reason=reason or None,
+                metadata={
+                    "task_title": spec.title,
+                    "tool_profile": profile.name,
+                    "labels": list(spec.labels),
+                    "completion_event_type": spec.completion_event_type,
+                },
+            )
+        except OSError:
+            pass  # metrics must not block run completion reporting
+
     if report_to_hub:
         detail = {
             "repo": spec.target_repo.name,
@@ -113,6 +196,10 @@ def run_task(spec: TaskSpec, adapter=None, report_to_hub: bool = True) -> RunRes
             "head_after": head_after,
             "ok": ok,
             "reason": reason,
+            "tool_profile": profile.name,
+            "budget_tokens": budget_tokens,
+            "tokens_spent": tokens_spent,
+            "execution_time_s": round(execution_time_s, 3),
         }
         hub.post_progress_event(
             summary=f"executor run: {spec.title} ({'ok' if ok else 'failed'})",
@@ -120,6 +207,15 @@ def run_task(spec: TaskSpec, adapter=None, report_to_hub: bool = True) -> RunRes
             detail=detail,
             task_id=spec.hub_task_id,
         )
+        if tokens_spent is not None or budget_tokens is not None:
+            hub.post_token_event(
+                repo=spec.target_repo.name,
+                tokens=tokens_spent or 0,
+                budget=budget_tokens,
+                agent=spec.agent,
+                ok=ok,
+                detail={"task_title": spec.title, "tool_profile": profile.name},
+            )
         if ok and spec.hub_task_id:
             hub.close_task(spec.hub_task_id)