Implement REIN-A-0002 ops_run claim loop and approach registry.

Add activity-core ops_run client, approach selection (FI/Binky/mail/agent),
claim-loop worker with lease heartbeat, CLI run --from-ops-run and claim-loop,
install units, and docs demoting issue-core to legacy external tickets.
T05 timer cutover remains operator after five clean cycles.

rein_aharness/approaches.py

@@ -0,0 +1,359 @@
+"""Approach registry: match ops_run → cheapest correct executor (REIN-A-0002-T02).
+
+Prefer deterministic / structured llm-connect adapters before open-ended agent
+sessions. Add rows in APPROACH_RULES (order = priority).
+
+How to add a row:
+  1. Implement a ``run_*`` function or CLI command.
+  2. Append an ApproachRule with match predicates (labels / blob substrings).
+  3. Wire execution in ``execute_approach``.
+  4. Add pure match tests in ``tests/test_approaches.py``.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+from rein_aharness.ops_run_client import OpsRun, OpsRunConfig, resolve_ops_target
+from rein_aharness.taskspec import TaskSpecError
+
+# Approach command names (stable; used in metrics + ops_run.result)
+APPROACH_FI_RESEARCH_BRIEF = "fi-research-brief"
+APPROACH_BRIEF_DAILY = "brief-daily"
+APPROACH_MAIL_SCAN = "mail-scan"
+APPROACH_MAIL_TRIAGE = "mail-triage"
+APPROACH_MAIL_PIPELINE = "mail-scan+triage"
+APPROACH_AGENT_SESSION = "agent-session"
+APPROACH_UNMATCHED = "unmatched"
+
+
+@dataclass(frozen=True)
+class ApproachRule:
+    name: str
+    """Match if any of these labels appear (case-insensitive)."""
+    labels_any: frozenset[str] = frozenset()
+    """Match if all of these labels appear."""
+    labels_all: frozenset[str] = frozenset()
+    """Match if any substring appears in definition id + labels + title + hint."""
+    blob_contains: frozenset[str] = frozenset()
+    """If approach_hint equals this command name, select immediately."""
+    hint_name: str | None = None
+
+
+# Ordered: first match wins. Cheapest / most specific first.
+APPROACH_RULES: tuple[ApproachRule, ...] = (
+    ApproachRule(
+        name=APPROACH_FI_RESEARCH_BRIEF,
+        labels_any=frozenset({"research-brief", "freedom-intelligence"}),
+        blob_contains=frozenset(
+            {"fi-daily", "fi_daily", "fi-research", "freedom intelligence"}
+        ),
+        hint_name=APPROACH_FI_RESEARCH_BRIEF,
+    ),
+    ApproachRule(
+        name=APPROACH_BRIEF_DAILY,
+        labels_any=frozenset({"rhythm", "binky-daily", "daily-brief"}),
+        labels_all=frozenset(),  # rhythm alone is enough for Binky daily
+        blob_contains=frozenset(
+            {"binky-daily", "binky_daily", "daily-rhythm", "daily_brief"}
+        ),
+        hint_name=APPROACH_BRIEF_DAILY,
+    ),
+    ApproachRule(
+        name=APPROACH_MAIL_PIPELINE,
+        labels_any=frozenset({"mail-intake", "mail_intake"}),
+        blob_contains=frozenset({"mail-intake", "mail_intake", "weekly-mail"}),
+        hint_name=APPROACH_MAIL_PIPELINE,
+    ),
+    ApproachRule(
+        name=APPROACH_MAIL_SCAN,
+        labels_any=frozenset({"mail-scan"}),
+        hint_name=APPROACH_MAIL_SCAN,
+    ),
+    ApproachRule(
+        name=APPROACH_MAIL_TRIAGE,
+        labels_any=frozenset({"mail-triage"}),
+        hint_name=APPROACH_MAIL_TRIAGE,
+    ),
+    ApproachRule(
+        name=APPROACH_AGENT_SESSION,
+        labels_any=frozenset({"agent-session", "agent_session"}),
+        blob_contains=frozenset({"agent-session"}),
+        hint_name=APPROACH_AGENT_SESSION,
+    ),
+)
+
+
+def _match_blob(run: OpsRun) -> str:
+    parts = [
+        run.activity_definition_id or "",
+        run.title or "",
+        run.approach_hint or "",
+        " ".join(run.labels or []),
+        run.source_id or "",
+        run.target_repo or "",
+    ]
+    return " ".join(parts).lower()
+
+
+def select_approach(run: OpsRun) -> str:
+    """Return approach command name for this ops_run (pure)."""
+    hint = (run.approach_hint or "").strip().lower()
+    if hint:
+        for rule in APPROACH_RULES:
+            if rule.hint_name and hint in {
+                rule.hint_name,
+                rule.name,
+                rule.hint_name.replace("-", "_"),
+            }:
+                return rule.name
+        # Explicit known command as hint
+        known = {
+            APPROACH_FI_RESEARCH_BRIEF,
+            APPROACH_BRIEF_DAILY,
+            APPROACH_MAIL_SCAN,
+            APPROACH_MAIL_TRIAGE,
+            APPROACH_MAIL_PIPELINE,
+            APPROACH_AGENT_SESSION,
+        }
+        if hint in known or hint.replace("_", "-") in known:
+            return hint.replace("_", "-")
+
+    labels = {str(x).lower() for x in (run.labels or [])}
+    blob = _match_blob(run)
+
+    for rule in APPROACH_RULES:
+        if rule.labels_all and not rule.labels_all.issubset(labels):
+            continue
+        if rule.labels_any and (rule.labels_any & labels):
+            return rule.name
+        if rule.blob_contains and any(s in blob for s in rule.blob_contains):
+            return rule.name
+
+    return APPROACH_UNMATCHED
+
+
+@dataclass
+class ApproachResult:
+    ok: bool
+    approach: str
+    result: dict[str, Any] = field(default_factory=dict)
+    reason: str = ""
+    reopen: bool = False
+
+
+def execute_approach(
+    run: OpsRun,
+    *,
+    approach: str | None = None,
+    config: OpsRunConfig | None = None,
+    report_to_hub: bool = True,
+    commit: bool = True,
+) -> ApproachResult:
+    """Dispatch to the selected executor. Never raises for business failure."""
+    cfg = config or OpsRunConfig.from_env()
+    name = approach or select_approach(run)
+
+    if name == APPROACH_UNMATCHED:
+        return ApproachResult(
+            ok=False,
+            approach=name,
+            reason=(
+                "no approach matched labels/definition; "
+                f"labels={run.labels!r} def={run.activity_definition_id!r}"
+            ),
+            reopen=False,
+        )
+
+    try:
+        target = resolve_ops_target(run, cfg)
+    except TaskSpecError as exc:
+        return ApproachResult(
+            ok=False,
+            approach=name,
+            reason=str(exc),
+            reopen=False,
+        )
+
+    try:
+        if name == APPROACH_FI_RESEARCH_BRIEF:
+            return _run_fi(target, report_to_hub=report_to_hub, commit=commit)
+        if name == APPROACH_BRIEF_DAILY:
+            return _run_brief_daily(target, report_to_hub=report_to_hub, commit=commit)
+        if name == APPROACH_MAIL_SCAN:
+            return _run_mail_scan(target, report_to_hub=report_to_hub)
+        if name == APPROACH_MAIL_TRIAGE:
+            return _run_mail_triage(target, report_to_hub=report_to_hub, commit=commit)
+        if name == APPROACH_MAIL_PIPELINE:
+            return _run_mail_pipeline(
+                target, report_to_hub=report_to_hub, commit=commit
+            )
+        if name == APPROACH_AGENT_SESSION:
+            return _run_agent_session(run, target, report_to_hub=report_to_hub)
+    except Exception as exc:  # noqa: BLE001 — surface as fail ops_run
+        return ApproachResult(
+            ok=False,
+            approach=name,
+            reason=f"{type(exc).__name__}: {exc}",
+            reopen=True,
+        )
+
+    return ApproachResult(
+        ok=False,
+        approach=name,
+        reason=f"approach not implemented: {name}",
+        reopen=False,
+    )
+
+
+def _run_fi(target: Path, *, report_to_hub: bool, commit: bool) -> ApproachResult:
+    from rein_aharness.fi_research_brief import run_fi_research_brief
+
+    r = run_fi_research_brief(
+        target_repo=target,
+        report_to_hub=report_to_hub,
+        commit=commit,
+    )
+    return ApproachResult(
+        ok=r.ok,
+        approach=APPROACH_FI_RESEARCH_BRIEF,
+        result={
+            "date": r.date,
+            "path": r.path,
+            "wrote": r.wrote,
+            "committed": r.committed,
+            "skipped_existing": r.skipped_existing,
+            "collection_candidates": r.collection_candidates,
+            "head_after": r.head_after,
+        },
+        reason=r.reason,
+        reopen=not r.ok and not r.skipped_existing,
+    )
+
+
+def _run_brief_daily(
+    target: Path, *, report_to_hub: bool, commit: bool
+) -> ApproachResult:
+    from rein_aharness.brief_daily import run_brief_daily
+
+    r = run_brief_daily(
+        target_repo=target,
+        report_to_hub=report_to_hub,
+        commit=commit,
+    )
+    return ApproachResult(
+        ok=r.ok,
+        approach=APPROACH_BRIEF_DAILY,
+        result={
+            "date": r.date,
+            "path": r.path,
+            "wrote": r.wrote,
+            "committed": r.committed,
+            "skipped_existing": r.skipped_existing,
+            "head_after": r.head_after,
+        },
+        reason=r.reason,
+        reopen=not r.ok and not r.skipped_existing,
+    )
+
+
+def _run_mail_scan(target: Path, *, report_to_hub: bool) -> ApproachResult:
+    from rein_aharness.mailscan import run_mail_scan
+
+    r = run_mail_scan(target_repo=target, report_to_hub=report_to_hub)
+    return ApproachResult(
+        ok=r.ok,
+        approach=APPROACH_MAIL_SCAN,
+        result={
+            "report": r.report_path,
+            "new_messages": r.new_messages,
+            "auth_lane": r.auth_lane,
+        },
+        reason=r.reason,
+        reopen=not r.ok,
+    )
+
+
+def _run_mail_triage(
+    target: Path, *, report_to_hub: bool, commit: bool
+) -> ApproachResult:
+    from rein_aharness.mail_triage import run_mail_triage
+
+    r = run_mail_triage(
+        target_repo=target,
+        report_to_hub=report_to_hub,
+        commit=commit,
+    )
+    return ApproachResult(
+        ok=r.ok,
+        approach=APPROACH_MAIL_TRIAGE,
+        result={
+            "report": r.report,
+            "entries_applied": r.entries_applied,
+            "committed": r.committed,
+            "head_after": r.head_after,
+        },
+        reason=r.reason,
+        reopen=not r.ok,
+    )
+
+
+def _run_mail_pipeline(
+    target: Path, *, report_to_hub: bool, commit: bool
+) -> ApproachResult:
+    scan = _run_mail_scan(target, report_to_hub=report_to_hub)
+    if not scan.ok:
+        scan.approach = APPROACH_MAIL_PIPELINE
+        return scan
+    triage = _run_mail_triage(target, report_to_hub=report_to_hub, commit=commit)
+    return ApproachResult(
+        ok=triage.ok,
+        approach=APPROACH_MAIL_PIPELINE,
+        result={"scan": scan.result, "triage": triage.result},
+        reason=triage.reason or scan.reason,
+        reopen=not triage.ok,
+    )
+
+
+def _run_agent_session(
+    run: OpsRun, target: Path, *, report_to_hub: bool
+) -> ApproachResult:
+    from rein_aharness.ops_run_client import ops_run_to_taskspec
+    from rein_aharness.runner import run_task
+
+    # Infer agent/event from labels (reuse intake hints)
+    from rein_aharness.intake import EmittedIssue, infer_agent_and_event
+
+    pseudo = EmittedIssue(
+        issue_id=run.id,
+        title=run.title,
+        description=run.description,
+        labels=list(run.labels),
+        target_repo=run.target_repo,
+        activity_definition_id=run.activity_definition_id,
+    )
+    agent, event = infer_agent_and_event(pseudo)
+    spec = ops_run_to_taskspec(
+        run,
+        agent=agent,
+        completion_event_type=event,
+    )
+    # target already resolved into TaskSpec
+    assert spec.target_repo == target or True
+    r = run_task(spec, report_to_hub=report_to_hub)
+    return ApproachResult(
+        ok=r.ok,
+        approach=APPROACH_AGENT_SESSION,
+        result={
+            "committed": r.committed,
+            "head_after": r.head_after,
+            "tool_profile": r.tool_profile,
+            "tokens_spent": r.tokens_spent,
+        },
+        reason=r.reason,
+        reopen=not r.ok,
+    )
+
+

rein_aharness/claim_loop.py

@@ -0,0 +1,307 @@
+"""Continuous ops_run claim worker (REIN-A-0002-T03).
+
+  rein-aharness claim-loop
+  rein-aharness claim-loop --once
+  rein-aharness poll --source=ops-run
+
+Concurrency default 1. Heartbeats while an approach runs.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import signal
+import threading
+import time
+from dataclasses import dataclass, field
+from typing import Any
+
+from rein_aharness.approaches import (
+    APPROACH_UNMATCHED,
+    ApproachResult,
+    execute_approach,
+    select_approach,
+)
+from rein_aharness.ops_run_client import (
+    ActivityCoreOpsClient,
+    OpsRun,
+    OpsRunConfig,
+    OpsRunError,
+)
+
+logger = logging.getLogger("rein_aharness.claim_loop")
+
+
+@dataclass
+class ProcessResult:
+    claimed: bool
+    empty: bool = False
+    run_id: str | None = None
+    approach: str | None = None
+    ok: bool | None = None
+    reason: str = ""
+    ops_state: str | None = None
+    detail: dict[str, Any] = field(default_factory=dict)
+
+
+def _heartbeat_interval(lease_seconds: int) -> float:
+    # Heartbeat at 1/3 lease, min 30s, max 300s
+    return max(30.0, min(300.0, lease_seconds / 3.0))
+
+
+class _Heartbeat:
+    def __init__(
+        self,
+        client: ActivityCoreOpsClient,
+        run_id: str,
+        lease_seconds: int,
+    ):
+        self._client = client
+        self._run_id = run_id
+        self._lease = lease_seconds
+        self._stop = threading.Event()
+        self._thread: threading.Thread | None = None
+
+    def start(self) -> None:
+        interval = _heartbeat_interval(self._lease)
+
+        def _loop() -> None:
+            while not self._stop.wait(interval):
+                try:
+                    self._client.heartbeat(self._run_id, lease_seconds=self._lease)
+                    logger.info("heartbeat ok run_id=%s", self._run_id)
+                except OpsRunError as exc:
+                    logger.warning("heartbeat failed run_id=%s: %s", self._run_id, exc)
+
+        self._thread = threading.Thread(
+            target=_loop, name=f"ops-hb-{self._run_id[:8]}", daemon=True
+        )
+        self._thread.start()
+
+    def stop(self) -> None:
+        self._stop.set()
+        if self._thread is not None:
+            self._thread.join(timeout=5.0)
+
+
+def process_one(
+    client: ActivityCoreOpsClient | None = None,
+    *,
+    report_to_hub: bool = True,
+    commit: bool = True,
+    dry_run: bool = False,
+) -> ProcessResult:
+    """Claim at most one ops_run, execute approach, complete or fail."""
+    client = client or ActivityCoreOpsClient()
+    cfg = client.config
+
+    try:
+        claimed = client.claim(limit=1)
+    except OpsRunError as exc:
+        return ProcessResult(claimed=False, reason=f"claim error: {exc}")
+
+    if not claimed:
+        return ProcessResult(claimed=False, empty=True, reason="queue empty")
+
+    run = claimed[0]
+    approach = select_approach(run)
+    logger.info(
+        "claimed run_id=%s approach=%s title=%r labels=%s",
+        run.id,
+        approach,
+        run.title,
+        run.labels,
+    )
+
+    if dry_run:
+        try:
+            client.fail(
+                run.id,
+                error="dry-run: not executing",
+                reopen=True,
+                result={"approach": approach, "dry_run": True},
+            )
+        except OpsRunError as exc:
+            return ProcessResult(
+                claimed=True,
+                run_id=run.id,
+                approach=approach,
+                ok=False,
+                reason=f"dry-run reopen failed: {exc}",
+            )
+        return ProcessResult(
+            claimed=True,
+            run_id=run.id,
+            approach=approach,
+            ok=True,
+            reason="dry-run reopened",
+            ops_state="open",
+            detail={"dry_run": True},
+        )
+
+    hb = _Heartbeat(client, run.id, cfg.lease_seconds)
+    hb.start()
+    try:
+        ar: ApproachResult = execute_approach(
+            run,
+            approach=approach,
+            config=cfg,
+            report_to_hub=report_to_hub,
+            commit=commit,
+        )
+    finally:
+        hb.stop()
+
+    payload = {
+        "approach": ar.approach,
+        "ok": ar.ok,
+        "reason": ar.reason,
+        **(ar.result or {}),
+    }
+
+    try:
+        if ar.ok:
+            # skipped_existing still succeeds the ops_run (idempotent day)
+            out = client.complete(run.id, result=payload)
+            state = out.state
+            logger.info("completed run_id=%s approach=%s", run.id, ar.approach)
+        else:
+            reopen = ar.reopen and ar.approach != APPROACH_UNMATCHED
+            out = client.fail(
+                run.id,
+                error=ar.reason or "approach failed",
+                reopen=reopen,
+                result=payload,
+            )
+            state = out.state
+            logger.warning(
+                "failed run_id=%s approach=%s reopen=%s reason=%s",
+                run.id,
+                ar.approach,
+                reopen,
+                ar.reason,
+            )
+    except OpsRunError as exc:
+        return ProcessResult(
+            claimed=True,
+            run_id=run.id,
+            approach=ar.approach,
+            ok=False,
+            reason=f"close ops_run failed: {exc}; approach_ok={ar.ok} {ar.reason}",
+            detail=payload,
+        )
+
+    return ProcessResult(
+        claimed=True,
+        run_id=run.id,
+        approach=ar.approach,
+        ok=ar.ok,
+        reason=ar.reason,
+        ops_state=state,
+        detail=payload,
+    )
+
+
+def poll_peek(client: ActivityCoreOpsClient | None = None) -> list[dict[str, Any]]:
+    """List open ops_runs with selected approach (no claim)."""
+    client = client or ActivityCoreOpsClient()
+    rows = client.list_open()
+    out = []
+    for run in rows:
+        out.append(
+            {
+                "id": run.id,
+                "title": run.title,
+                "state": run.state,
+                "labels": run.labels,
+                "target_repo": run.target_repo,
+                "approach": select_approach(run),
+                "created_at": run.raw.get("created_at"),
+            }
+        )
+    return out
+
+
+def run_claim_loop(
+    *,
+    once: bool = False,
+    interval_seconds: float | None = None,
+    report_to_hub: bool = True,
+    commit: bool = True,
+    dry_run: bool = False,
+    max_iterations: int | None = None,
+) -> int:
+    """Poll forever (or once). Returns process exit code."""
+    if interval_seconds is None:
+        try:
+            interval_seconds = float(
+                os.environ.get("AGENT_HARNESS_CLAIM_INTERVAL", "30")
+            )
+        except ValueError:
+            interval_seconds = 30.0
+    interval_seconds = max(1.0, interval_seconds)
+
+    stop = threading.Event()
+
+    def _handle_sig(*_args: Any) -> None:
+        logger.info("shutdown signal received")
+        stop.set()
+
+    signal.signal(signal.SIGINT, _handle_sig)
+    signal.signal(signal.SIGTERM, _handle_sig)
+
+    client = ActivityCoreOpsClient()
+    logger.info(
+        "claim-loop start worker_id=%s url=%s labels=%s interval=%ss once=%s",
+        client.config.worker_id,
+        client.config.base_url,
+        client.config.claim_labels,
+        interval_seconds,
+        once,
+    )
+
+    iterations = 0
+    exit_code = 0
+    while not stop.is_set():
+        iterations += 1
+        t0 = time.monotonic()
+        try:
+            result = process_one(
+                client,
+                report_to_hub=report_to_hub,
+                commit=commit,
+                dry_run=dry_run,
+            )
+        except Exception as exc:  # noqa: BLE001
+            logger.exception("process_one crashed: %s", exc)
+            result = ProcessResult(claimed=False, reason=str(exc))
+            exit_code = 1
+
+        elapsed = time.monotonic() - t0
+        if result.empty:
+            logger.debug("queue empty (%.2fs)", elapsed)
+        else:
+            logger.info(
+                "cycle claimed=%s run_id=%s ok=%s approach=%s state=%s reason=%s (%.2fs)",
+                result.claimed,
+                result.run_id,
+                result.ok,
+                result.approach,
+                result.ops_state,
+                result.reason,
+                elapsed,
+            )
+            if result.claimed and result.ok is False:
+                exit_code = 1
+
+        if once:
+            break
+        if max_iterations is not None and iterations >= max_iterations:
+            break
+        # Sleep full interval only when empty; short pause after work
+        sleep_for = interval_seconds if result.empty else min(2.0, interval_seconds)
+        stop.wait(sleep_for)
+
+    logger.info("claim-loop stop iterations=%s exit=%s", iterations, exit_code)
+    return 0 if once and exit_code == 0 else exit_code if once else 0

rein_aharness/cli.py

@@ -69,6 +69,48 @@ def _cmd_profiles(_args: argparse.Namespace) -> int:
 
 
 def _cmd_poll(args: argparse.Namespace) -> int:
+    source = getattr(args, "source", "issue-core") or "issue-core"
+    if source in {"ops-run", "ops_run", "ops"}:
+        from rein_aharness.claim_loop import process_one, poll_peek
+        from rein_aharness.ops_run_client import OpsRunError
+
+        try:
+            if args.no_claim:
+                rows = poll_peek()
+                print(
+                    json.dumps(
+                        {"source": "ops-run", "queue": "empty" if not rows else "open", "items": rows},
+                        indent=2,
+                    )
+                )
+                return 0
+            result = process_one(
+                dry_run=bool(getattr(args, "dry_run", False)),
+                report_to_hub=False if getattr(args, "no_hub", False) else True,
+            )
+        except OpsRunError as exc:
+            print(f"ops-run error: {exc}", file=sys.stderr)
+            return 2
+        print(
+            json.dumps(
+                {
+                    "source": "ops-run",
+                    "claimed": result.claimed,
+                    "empty": result.empty,
+                    "run_id": result.run_id,
+                    "approach": result.approach,
+                    "ok": result.ok,
+                    "ops_state": result.ops_state,
+                    "reason": result.reason,
+                    "detail": result.detail,
+                },
+                indent=2,
+            )
+        )
+        if result.empty:
+            return 0
+        return 0 if result.ok else 1
+
     from rein_aharness.intake import IntakeError, IssueCoreClient, poll_next
 
     try:
@@ -78,12 +120,13 @@ def _cmd_poll(args: argparse.Namespace) -> int:
         print(f"intake error: {exc}", file=sys.stderr)
         return 2
     if result is None:
-        print(json.dumps({"queue": "empty"}, indent=2))
+        print(json.dumps({"source": "issue-core", "queue": "empty"}, indent=2))
         return 0
     issue, spec = result
     print(
         json.dumps(
             {
+                "source": "issue-core",
                 "issue_id": issue.issue_id,
                 "state": issue.state,
                 "title": issue.title,
@@ -99,12 +142,62 @@ def _cmd_poll(args: argparse.Namespace) -> int:
     return 0
 
 
+def _cmd_claim_loop(args: argparse.Namespace) -> int:
+    import logging
+
+    from rein_aharness.claim_loop import run_claim_loop
+
+    logging.basicConfig(
+        level=logging.DEBUG if args.verbose else logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+    return run_claim_loop(
+        once=bool(args.once),
+        interval_seconds=args.interval,
+        report_to_hub=not args.no_hub,
+        commit=not args.no_commit,
+        dry_run=bool(args.dry_run),
+        max_iterations=args.max_iterations,
+    )
+
+
 def _cmd_run(args: argparse.Namespace) -> int:
     from rein_aharness.intake import IntakeError, IssueCoreClient, poll_next
 
     issue_id: str | None = None
     client: IssueCoreClient | None = None
 
+    if getattr(args, "from_ops_run", False):
+        from rein_aharness.claim_loop import process_one
+        from rein_aharness.ops_run_client import OpsRunError
+
+        try:
+            result = process_one(
+                report_to_hub=not args.no_hub,
+                commit=not getattr(args, "no_commit", False),
+            )
+        except OpsRunError as exc:
+            print(f"ops-run error: {exc}", file=sys.stderr)
+            return 2
+        print(
+            json.dumps(
+                {
+                    "source": "ops-run",
+                    "ok": bool(result.ok) if result.claimed else True,
+                    "empty": result.empty,
+                    "run_id": result.run_id,
+                    "approach": result.approach,
+                    "ops_state": result.ops_state,
+                    "reason": result.reason,
+                    "detail": result.detail,
+                },
+                indent=2,
+            )
+        )
+        if result.empty:
+            return 0
+        return 0 if result.ok else 1
+
     if args.from_issue_core:
         try:
             client = IssueCoreClient()
@@ -113,14 +206,14 @@ def _cmd_run(args: argparse.Namespace) -> int:
             print(f"intake error: {exc}", file=sys.stderr)
             return 2
         if polled is None:
-            print(json.dumps({"ok": True, "queue": "empty"}, indent=2))
+            print(json.dumps({"ok": True, "source": "issue-core", "queue": "empty"}, indent=2))
             return 0
         issue, spec = polled
         issue_id = issue.issue_id
     else:
         if not args.task_file:
             print(
-                "error: provide --task-file or --from-issue-core",
+                "error: provide --task-file, --from-ops-run, or --from-issue-core",
                 file=sys.stderr,
             )
             return 2
@@ -160,6 +253,7 @@ def _cmd_run(args: argparse.Namespace) -> int:
         json.dumps(
             {
                 "ok": result.ok,
+                "source": "issue-core" if issue_id else "task-file",
                 "committed": result.committed,
                 "head_after": result.head_after,
                 "persona_source": result.persona_source,
@@ -184,16 +278,26 @@ def main(argv: list[str] | None = None) -> int:
 
     run = sub.add_parser(
         "run",
-        help="Execute one task from a JSON file or next issue-core emission",
+        help="Execute one task: ops_run (primary), issue-core (legacy), or task file",
     )
     run_src = run.add_mutually_exclusive_group(required=True)
     run_src.add_argument("--task-file", help="Local JSON task-spec (dev path)")
+    run_src.add_argument(
+        "--from-ops-run",
+        action="store_true",
+        help="Claim one activity-core ops_run, select approach, execute, complete/fail",
+    )
     run_src.add_argument(
         "--from-issue-core",
         action="store_true",
-        help="Poll issue-core for one open harness-labeled task, claim, run, close",
+        help="Legacy: poll issue-core for one open harness-labeled task, claim, run, close",
     )
     run.add_argument("--no-hub", action="store_true", help="Skip hub reporting")
+    run.add_argument(
+        "--no-commit",
+        action="store_true",
+        help="With --from-ops-run: approaches that support it skip git commit",
+    )
     run.add_argument(
         "--no-metrics",
         action="store_true",
@@ -329,13 +433,51 @@ def main(argv: list[str] | None = None) -> int:
 
     poll = sub.add_parser(
         "poll",
-        help="Peek/claim next open issue-core task labeled for the harness (no execute)",
+        help="Peek/claim next task (default source=ops-run; issue-core is legacy)",
+    )
+    poll.add_argument(
+        "--source",
+        choices=["ops-run", "issue-core"],
+        default="ops-run",
+        help="ops-run = activity-core claim queue (primary); issue-core = legacy tickets",
     )
     poll.add_argument(
         "--no-claim",
         action="store_true",
-        help="List/map only; do not set in_progress",
+        help="List/map only; do not claim",
     )
+    poll.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="With ops-run: claim then fail+reopen without executing",
+    )
+    poll.add_argument("--no-hub", action="store_true", help="Skip hub on execute")
+
+    claim_loop = sub.add_parser(
+        "claim-loop",
+        help="Continuously claim ops_runs, select approach, execute, complete/fail",
+    )
+    claim_loop.add_argument(
+        "--once",
+        action="store_true",
+        help="Process at most one claim cycle then exit",
+    )
+    claim_loop.add_argument(
+        "--interval",
+        type=float,
+        default=None,
+        help="Seconds between empty-queue polls (default env AGENT_HARNESS_CLAIM_INTERVAL or 30)",
+    )
+    claim_loop.add_argument(
+        "--max-iterations",
+        type=int,
+        default=None,
+        help="Stop after N cycles (tests / bounded runs)",
+    )
+    claim_loop.add_argument("--dry-run", action="store_true")
+    claim_loop.add_argument("--no-hub", action="store_true")
+    claim_loop.add_argument("--no-commit", action="store_true")
+    claim_loop.add_argument("-v", "--verbose", action="store_true")
 
     args = parser.parse_args(argv)
 
@@ -348,6 +490,9 @@ def main(argv: list[str] | None = None) -> int:
     if args.command == "poll":
         return _cmd_poll(args)
 
+    if args.command == "claim-loop":
+        return _cmd_claim_loop(args)
+
     if args.command == "run":
         return _cmd_run(args)
 

rein_aharness/ops_run_client.py

@@ -0,0 +1,288 @@
+"""activity-core ops_run claim client (REIN-A-0002 / ACT-ADR-005).
+
+Primary intake for scheduled automation. Does **not** use issue-core or Forgejo.
+
+Environment:
+
+  ACTIVITY_CORE_URL              default http://127.0.0.1:8010
+  ACTIVITY_CORE_WORKER_TOKEN     X-Worker-Token / Bearer (optional if API open)
+  AGENT_HARNESS_WORKER_ID        claim owner (default: rein-aharness@hostname)
+  AGENT_HARNESS_OPS_LABELS       comma labels for claim filter (default: automated)
+  AGENT_HARNESS_OPS_LABELS_MODE  any|all (default: any)
+  AGENT_HARNESS_OPS_LEASE_SECONDS  claim lease (default: 900)
+  AGENT_HARNESS_REPO_MAP / AGENT_HARNESS_REPO_ROOTS  shared with intake
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import socket
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+import httpx
+
+from rein_aharness.intake import resolve_target_repo
+from rein_aharness.taskspec import TaskSpec, TaskSpecError
+
+DEFAULT_ACTIVITY_CORE_URL = "http://127.0.0.1:8010"
+DEFAULT_OPS_LABELS = ("automated",)
+DEFAULT_REPO_ROOTS = ("~", "~/work")
+
+
+class OpsRunError(RuntimeError):
+    pass
+
+
+@dataclass
+class OpsRun:
+    """Normalized ops_run row from actcore-api."""
+
+    id: str
+    activity_definition_id: str
+    idempotency_key: str
+    target_repo: str | None
+    title: str
+    description: str
+    labels: list[str] = field(default_factory=list)
+    priority: str = "medium"
+    state: str = "open"
+    claim_owner: str | None = None
+    lease_until: str | None = None
+    attempt: int = 0
+    source_type: str = "rule"
+    source_id: str = ""
+    triggering_event_id: str = ""
+    approach_hint: str | None = None
+    result: dict[str, Any] = field(default_factory=dict)
+    raw: dict[str, Any] = field(default_factory=dict)
+
+    @classmethod
+    def from_api(cls, data: dict[str, Any]) -> "OpsRun":
+        return cls(
+            id=str(data.get("id") or ""),
+            activity_definition_id=str(data.get("activity_definition_id") or ""),
+            idempotency_key=str(data.get("idempotency_key") or ""),
+            target_repo=data.get("target_repo"),
+            title=str(data.get("title") or ""),
+            description=str(data.get("description") or ""),
+            labels=[str(x) for x in (data.get("labels") or [])],
+            priority=str(data.get("priority") or "medium"),
+            state=str(data.get("state") or "open"),
+            claim_owner=data.get("claim_owner"),
+            lease_until=data.get("lease_until"),
+            attempt=int(data.get("attempt") or 0),
+            source_type=str(data.get("source_type") or "rule"),
+            source_id=str(data.get("source_id") or ""),
+            triggering_event_id=str(data.get("triggering_event_id") or ""),
+            approach_hint=data.get("approach_hint"),
+            result=dict(data.get("result") or {}),
+            raw=data,
+        )
+
+
+@dataclass
+class OpsRunConfig:
+    base_url: str = DEFAULT_ACTIVITY_CORE_URL
+    worker_token: str = ""
+    worker_id: str = "rein-aharness"
+    claim_labels: tuple[str, ...] = DEFAULT_OPS_LABELS
+    labels_mode: str = "any"
+    lease_seconds: int = 900
+    repo_map: dict[str, str] = field(default_factory=dict)
+    repo_roots: tuple[str, ...] = DEFAULT_REPO_ROOTS
+    timeout: float = 30.0
+
+    @classmethod
+    def from_env(cls) -> "OpsRunConfig":
+        labels_raw = os.environ.get("AGENT_HARNESS_OPS_LABELS", "automated")
+        labels = tuple(p.strip() for p in labels_raw.split(",") if p.strip())
+        roots_raw = os.environ.get("AGENT_HARNESS_REPO_ROOTS", "~:~/work")
+        roots = tuple(p.strip() for p in roots_raw.split(":") if p.strip())
+        repo_map: dict[str, str] = {}
+        map_raw = os.environ.get("AGENT_HARNESS_REPO_MAP", "").strip()
+        if map_raw:
+            repo_map = {str(k): str(v) for k, v in json.loads(map_raw).items()}
+        host = socket.gethostname().split(".")[0]
+        default_worker = f"rein-aharness@{host}"
+        try:
+            lease = max(30, int(os.environ.get("AGENT_HARNESS_OPS_LEASE_SECONDS", "900")))
+        except ValueError:
+            lease = 900
+        mode = (os.environ.get("AGENT_HARNESS_OPS_LABELS_MODE") or "any").strip().lower()
+        if mode not in {"any", "all"}:
+            mode = "any"
+        return cls(
+            base_url=os.environ.get(
+                "ACTIVITY_CORE_URL", DEFAULT_ACTIVITY_CORE_URL
+            ).rstrip("/"),
+            worker_token=(
+                os.environ.get("ACTIVITY_CORE_WORKER_TOKEN")
+                or os.environ.get("AGENT_HARNESS_WORKER_TOKEN")
+                or ""
+            ).strip(),
+            worker_id=os.environ.get("AGENT_HARNESS_WORKER_ID", default_worker).strip()
+            or default_worker,
+            claim_labels=labels or DEFAULT_OPS_LABELS,
+            labels_mode=mode,
+            lease_seconds=lease,
+            repo_map=repo_map,
+            repo_roots=roots or DEFAULT_REPO_ROOTS,
+        )
+
+
+class ActivityCoreOpsClient:
+    """REST client for POST /ops-runs/claim|heartbeat|complete|fail."""
+
+    def __init__(self, config: OpsRunConfig | None = None):
+        self.config = config or OpsRunConfig.from_env()
+
+    def _headers(self) -> dict[str, str]:
+        headers = {"Accept": "application/json", "Content-Type": "application/json"}
+        if self.config.worker_token:
+            headers["X-Worker-Token"] = self.config.worker_token
+            headers["Authorization"] = f"Bearer {self.config.worker_token}"
+        return headers
+
+    def list_open(self, *, limit: int = 50) -> list[OpsRun]:
+        try:
+            resp = httpx.get(
+                f"{self.config.base_url}/ops-runs",
+                params={"state": "open", "limit": str(limit)},
+                headers=self._headers(),
+                timeout=self.config.timeout,
+            )
+            resp.raise_for_status()
+        except httpx.HTTPError as exc:
+            raise OpsRunError(f"list ops-runs failed: {exc}") from exc
+        data = resp.json()
+        items = data.get("items") if isinstance(data, dict) else data
+        if not isinstance(items, list):
+            raise OpsRunError(f"unexpected list payload: {type(data)}")
+        return [OpsRun.from_api(item) for item in items if isinstance(item, dict)]
+
+    def claim(
+        self,
+        *,
+        labels: list[str] | None = None,
+        labels_mode: str | None = None,
+        limit: int = 1,
+        lease_seconds: int | None = None,
+    ) -> list[OpsRun]:
+        body = {
+            "worker_id": self.config.worker_id,
+            "labels": list(labels if labels is not None else self.config.claim_labels),
+            "labels_mode": labels_mode or self.config.labels_mode,
+            "limit": max(1, min(limit, 20)),
+            "lease_seconds": lease_seconds or self.config.lease_seconds,
+        }
+        try:
+            resp = httpx.post(
+                f"{self.config.base_url}/ops-runs/claim",
+                json=body,
+                headers=self._headers(),
+                timeout=self.config.timeout,
+            )
+            resp.raise_for_status()
+        except httpx.HTTPError as exc:
+            raise OpsRunError(f"claim failed: {exc}") from exc
+        data = resp.json()
+        items = data.get("items") if isinstance(data, dict) else []
+        return [OpsRun.from_api(item) for item in items if isinstance(item, dict)]
+
+    def heartbeat(
+        self,
+        run_id: str,
+        *,
+        lease_seconds: int | None = None,
+    ) -> OpsRun:
+        body: dict[str, Any] = {"worker_id": self.config.worker_id}
+        if lease_seconds is not None:
+            body["lease_seconds"] = lease_seconds
+        return self._post_run(run_id, "heartbeat", body)
+
+    def complete(
+        self,
+        run_id: str,
+        *,
+        result: dict[str, Any] | None = None,
+    ) -> OpsRun:
+        return self._post_run(
+            run_id,
+            "complete",
+            {"worker_id": self.config.worker_id, "result": result or {}},
+        )
+
+    def fail(
+        self,
+        run_id: str,
+        *,
+        error: str = "",
+        reopen: bool = False,
+        result: dict[str, Any] | None = None,
+    ) -> OpsRun:
+        return self._post_run(
+            run_id,
+            "fail",
+            {
+                "worker_id": self.config.worker_id,
+                "error": error,
+                "reopen": reopen,
+                "result": result or {},
+            },
+        )
+
+    def _post_run(self, run_id: str, action: str, body: dict[str, Any]) -> OpsRun:
+        try:
+            resp = httpx.post(
+                f"{self.config.base_url}/ops-runs/{run_id}/{action}",
+                json=body,
+                headers=self._headers(),
+                timeout=self.config.timeout,
+            )
+            resp.raise_for_status()
+        except httpx.HTTPError as exc:
+            raise OpsRunError(f"{action} ops_run {run_id} failed: {exc}") from exc
+        return OpsRun.from_api(resp.json())
+
+
+def ops_run_to_taskspec(
+    run: OpsRun,
+    config: OpsRunConfig | None = None,
+    *,
+    agent: str = "coach",
+    completion_event_type: str = "executor_run",
+    timeout_seconds: int = 900,
+) -> TaskSpec:
+    """Map claimed ops_run to TaskSpec for agent-session approach."""
+    cfg = config or OpsRunConfig.from_env()
+    if not run.target_repo:
+        raise TaskSpecError(f"ops_run {run.id} missing target_repo")
+    target = resolve_target_repo(
+        run.target_repo,
+        repo_map=cfg.repo_map,
+        repo_roots=cfg.repo_roots,
+    )
+    return TaskSpec(
+        title=run.title or "(untitled)",
+        description=run.description or "",
+        target_repo=target,
+        agent=agent,
+        labels=list(run.labels),
+        hub_task_id=run.id,
+        completion_event_type=completion_event_type,
+        timeout_seconds=timeout_seconds,
+    )
+
+
+def resolve_ops_target(run: OpsRun, config: OpsRunConfig | None = None) -> Path:
+    cfg = config or OpsRunConfig.from_env()
+    if not run.target_repo:
+        raise TaskSpecError(f"ops_run {run.id} missing target_repo")
+    return resolve_target_repo(
+        run.target_repo,
+        repo_map=cfg.repo_map,
+        repo_roots=cfg.repo_roots,
+    )