From 8200a672eafebc25bc03e67bc69e624e3f9dec54 Mon Sep 17 00:00:00 2001 From: tegwick Date: Mon, 3 Aug 2026 20:04:25 +0200 Subject: [PATCH] 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. --- README.md | 17 +- deploy/docs/timer-cutover.md | 49 +++ deploy/scripts/install-claim-loop-user.sh | 31 ++ deploy/systemd/claim-loop.env.example | 14 + .../systemd/rein-aharness-claim-loop.service | 22 ++ docs/ops-run-claim-loop.md | 128 +++++++ docs/task-intake.md | 104 +++-- rein_aharness/approaches.py | 359 ++++++++++++++++++ rein_aharness/claim_loop.py | 307 +++++++++++++++ rein_aharness/cli.py | 159 +++++++- rein_aharness/ops_run_client.py | 288 ++++++++++++++ tests/test_approaches.py | 92 +++++ tests/test_claim_loop.py | 97 +++++ tests/test_ops_run_client.py | 168 ++++++++ workplans/REIN-A-0002-ops-run-claim-loop.md | 45 ++- 15 files changed, 1807 insertions(+), 73 deletions(-) create mode 100644 deploy/docs/timer-cutover.md create mode 100755 deploy/scripts/install-claim-loop-user.sh create mode 100644 deploy/systemd/claim-loop.env.example create mode 100644 deploy/systemd/rein-aharness-claim-loop.service create mode 100644 docs/ops-run-claim-loop.md create mode 100644 rein_aharness/approaches.py create mode 100644 rein_aharness/claim_loop.py create mode 100644 rein_aharness/ops_run_client.py create mode 100644 tests/test_approaches.py create mode 100644 tests/test_claim_loop.py create mode 100644 tests/test_ops_run_client.py diff --git a/README.md b/README.md index bcebf1f..2215524 100644 --- a/README.md +++ b/README.md @@ -41,11 +41,20 @@ rein-aharness profiles # Run exactly one task (local JSON task-file path — dev) rein-aharness run --task-file examples/task-hello-sandbox.json [--no-hub] [--no-metrics] -# Production intake: poll issue-core (activity-core emissions), claim, run, close -export ISSUE_CORE_URL=http://127.0.0.1:8765 ISSUE_CORE_API_KEY=… -rein-aharness poll -rein-aharness run --from-issue-core +# Primary production intake: activity-core ops_run claim (ACT-ADR-005 / REIN-A-0002) +export ACTIVITY_CORE_URL=http://127.0.0.1:8010 +export ACTIVITY_CORE_WORKER_TOKEN=… +export AGENT_HARNESS_REPO_MAP='{"freedom-intelligence":"~/freedom-intelligence","binky-control":"~/binky-control"}' +rein-aharness poll --source=ops-run --no-claim +rein-aharness run --from-ops-run +rein-aharness claim-loop +# Install user service: ./deploy/scripts/install-claim-loop-user.sh +# Docs: docs/ops-run-claim-loop.md +# Legacy: issue-core tickets (external / Forgejo projection only — not FI/Binky ops) +export ISSUE_CORE_URL=http://127.0.0.1:8765 ISSUE_CORE_API_KEY=… +rein-aharness poll --source=issue-core +rein-aharness run --from-issue-core # Deterministic mailbox scan (no LLM session) rein-aharness mail-scan --target-repo ~/binky-control diff --git a/deploy/docs/timer-cutover.md b/deploy/docs/timer-cutover.md new file mode 100644 index 0000000..6174fb4 --- /dev/null +++ b/deploy/docs/timer-cutover.md @@ -0,0 +1,49 @@ +# Timer cutover (REIN-A-0002-T05) + +After claim-loop is the proven executor path, demote host oneshot timers to +**break-glass only**. + +## Proof window + +Before disabling timers, record **5 clean cycles** (weekday schedule fires or +forced operator triggers): + +| # | Date | Trigger | ops_run id | approach | result | +| - | ---- | ------- | ---------- | -------- | ------ | +| 1 | | | | | | +| 2 | | | | | | +| 3 | | | | | | +| 4 | | | | | | +| 5 | | | | | | + +Each row: `open` → `claimed` → `succeeded` (or `skipped_existing` complete), +with domain completion event (`fi_daily_brief` / `binky_daily_brief`). + +## Demote FI timer + +```bash +systemctl --user disable --now fi-research-brief-daily.timer +# Keep unit files installed for break-glass: +# systemctl --user start fi-research-brief-daily.service +``` + +## Demote Binky rhythm timer + +```bash +# unit names from binky-control/scripts/railiance-rhythm/systemd/ +systemctl --user disable --now binky-brief-daily.timer 2>/dev/null || true +systemctl --user list-timers --all | grep -E 'binky|fi-research' || true +``` + +## Keep break-glass + +- Install scripts remain; document “manual start if claim-loop down”. +- Do **not** delete domain rhythm scripts. +- activity-core schedule stays authority for *when due*. + +## Rollback + +```bash +systemctl --user enable --now fi-research-brief-daily.timer +# investigate claim-loop journal + actcore /ops-runs +``` diff --git a/deploy/scripts/install-claim-loop-user.sh b/deploy/scripts/install-claim-loop-user.sh new file mode 100755 index 0000000..0ddde6f --- /dev/null +++ b/deploy/scripts/install-claim-loop-user.sh @@ -0,0 +1,31 @@ +#!/usr/bin/env bash +# Install rein-aharness claim-loop as a user systemd service (railiance01). +# Run from a rein-aharness checkout ON the host: +# ./deploy/scripts/install-claim-loop-user.sh +set -euo pipefail +ROOT="$(cd "$(dirname "$0")/../.." && pwd)" +UNIT_DIR="${HOME}/.config/systemd/user" +ENV_DIR="${HOME}/.config/rein-aharness" +mkdir -p "${UNIT_DIR}" "${ENV_DIR}" + +cp -f "${ROOT}/deploy/systemd/rein-aharness-claim-loop.service" "${UNIT_DIR}/" + +if [[ ! -f "${ENV_DIR}/claim-loop.env" ]]; then + cp "${ROOT}/deploy/systemd/claim-loop.env.example" "${ENV_DIR}/claim-loop.env" + chmod 600 "${ENV_DIR}/claim-loop.env" + echo "Created ${ENV_DIR}/claim-loop.env — set ACTIVITY_CORE_URL, token, REPO_MAP." +else + echo "Keeping existing ${ENV_DIR}/claim-loop.env" +fi + +# Ensure package entrypoint exists +if [[ ! -x "${ROOT}/.venv/bin/rein-aharness" ]]; then + echo "warning: ${ROOT}/.venv/bin/rein-aharness missing; install with: cd ${ROOT} && uv sync" +fi + +systemctl --user daemon-reload +systemctl --user enable rein-aharness-claim-loop.service +echo "Enabled. Start when env is filled:" +echo " systemctl --user start rein-aharness-claim-loop.service" +echo " journalctl --user -u rein-aharness-claim-loop -f" +echo "Host timers remain break-glass until T05 cutover (docs/ops-run-claim-loop.md)." diff --git a/deploy/systemd/claim-loop.env.example b/deploy/systemd/claim-loop.env.example new file mode 100644 index 0000000..b42c80e --- /dev/null +++ b/deploy/systemd/claim-loop.env.example @@ -0,0 +1,14 @@ +# Copy to ~/.config/rein-aharness/claim-loop.env and chmod 600 + +ACTIVITY_CORE_URL=http://127.0.0.1:8010 +ACTIVITY_CORE_WORKER_TOKEN= +AGENT_HARNESS_WORKER_ID=rein-aharness@railiance01 +AGENT_HARNESS_OPS_LABELS=automated +AGENT_HARNESS_OPS_LABELS_MODE=any +AGENT_HARNESS_OPS_LEASE_SECONDS=900 +AGENT_HARNESS_CLAIM_INTERVAL=30 +AGENT_HARNESS_REPO_MAP={"freedom-intelligence":"/home/tegwick/freedom-intelligence","binky-control":"/home/tegwick/binky-control"} +AGENT_HARNESS_REPO_ROOTS=/home/tegwick:/home/tegwick/work +# LLM / hub as required by approaches +# LLM_CONNECT_URL=http://... +# STATE_HUB_URL=http://... diff --git a/deploy/systemd/rein-aharness-claim-loop.service b/deploy/systemd/rein-aharness-claim-loop.service new file mode 100644 index 0000000..8a175be --- /dev/null +++ b/deploy/systemd/rein-aharness-claim-loop.service @@ -0,0 +1,22 @@ +[Unit] +Description=rein-aharness ops_run claim loop (activity-core automation) +Documentation=file:%h/rein-aharness/docs/ops-run-claim-loop.md +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +# Adjust paths for the railiance user checkout +WorkingDirectory=%h/rein-aharness +EnvironmentFile=-%h/.config/rein-aharness/claim-loop.env +Environment=PYTHONUNBUFFERED=1 +# Prefer venv entrypoint when installed with uv/pip +ExecStart=%h/rein-aharness/.venv/bin/rein-aharness claim-loop +Restart=on-failure +RestartSec=15 +# Soft stop: claim-loop handles SIGTERM between cycles +KillSignal=SIGTERM +TimeoutStopSec=120 + +[Install] +WantedBy=default.target diff --git a/docs/ops-run-claim-loop.md b/docs/ops-run-claim-loop.md new file mode 100644 index 0000000..b7fa617 --- /dev/null +++ b/docs/ops-run-claim-loop.md @@ -0,0 +1,128 @@ +# Ops run claim loop (REIN-A-0002) + +**Primary intake** for activity-core scheduled automation. +**Architecture:** ACT-ADR-005 · activity-core [ops-run-queue.md](../../activity-core/docs/ops-run-queue.md) + +Does **not** require issue-core or Forgejo for FI / Binky-style ops. + +## Flow + +```text +activity-core emit_tasks + → INSERT ops_run (open) + dual-write activity_task_spawn + │ + ▼ +rein-aharness claim-loop + POST /ops-runs/claim + select_approach(labels, definition, hint) + execute (fi-research-brief | brief-daily | mail-* | agent-session) + POST complete | fail(+reopen) +``` + +## Environment + +| Variable | Default | Meaning | +| -------- | ------- | ------- | +| `ACTIVITY_CORE_URL` | `http://127.0.0.1:8010` | actcore-api base | +| `ACTIVITY_CORE_WORKER_TOKEN` | unset | `X-Worker-Token` / Bearer | +| `AGENT_HARNESS_WORKER_ID` | `rein-aharness@hostname` | claim owner | +| `AGENT_HARNESS_OPS_LABELS` | `automated` | claim label filter | +| `AGENT_HARNESS_OPS_LABELS_MODE` | `any` | `any` or `all` | +| `AGENT_HARNESS_OPS_LEASE_SECONDS` | `900` | claim lease | +| `AGENT_HARNESS_CLAIM_INTERVAL` | `30` | empty-queue poll interval | +| `AGENT_HARNESS_REPO_MAP` | `{}` | JSON slug→checkout path | +| `AGENT_HARNESS_REPO_ROOTS` | `~:~/work` | slug search roots | + +Example map on railiance01: + +```bash +export AGENT_HARNESS_REPO_MAP='{ + "freedom-intelligence":"/home/tegwick/freedom-intelligence", + "binky-control":"/home/tegwick/binky-control" +}' +export ACTIVITY_CORE_URL=http://127.0.0.1:8010 # or ClusterIP via tunnel +export ACTIVITY_CORE_WORKER_TOKEN=… # from actcore-runtime-secret +``` + +## Approach registry + +| Match | Approach | +| ----- | -------- | +| labels `research-brief` / `freedom-intelligence` or fi-daily* | `fi-research-brief` | +| labels `rhythm` / `binky-daily` or binky-daily* | `brief-daily` | +| labels `mail-intake` | `mail-scan` then `mail-triage` | +| labels `agent-session` | agentic `run` (TaskSpec) | +| none | fail ops_run (no silent drop) | + +Optional `approach_hint` on the ops_run overrides matching. + +Add a row: edit `rein_aharness/approaches.py` (`APPROACH_RULES` + `execute_approach`) +and tests in `tests/test_approaches.py`. Prefer cheapest correct path +(structured llm-connect before open-ended agent session). + +## CLI + +```bash +# Peek open ops_runs + selected approach +rein-aharness poll --source=ops-run --no-claim + +# Claim one, execute approach, complete/fail +rein-aharness run --from-ops-run +# or +rein-aharness poll --source=ops-run + +# Continuous worker (systemd) +rein-aharness claim-loop +rein-aharness claim-loop --once --verbose + +# Dry-run: claim then fail+reopen without domain work +rein-aharness poll --source=ops-run --dry-run +``` + +Legacy issue-core (external tickets only): + +```bash +rein-aharness poll --source=issue-core +rein-aharness run --from-issue-core +``` + +## Install claim-loop on railiance01 (user systemd) + +```bash +# From rein-aharness checkout on railiance01 +./deploy/scripts/install-claim-loop-user.sh + +# Or manually: +mkdir -p ~/.config/systemd/user +cp deploy/systemd/rein-aharness-claim-loop.service ~/.config/systemd/user/ +# Edit Environment= paths / token in the unit or env file +systemctl --user daemon-reload +systemctl --user enable --now rein-aharness-claim-loop.service +journalctl --user -u rein-aharness-claim-loop -f +``` + +Port-forward if claim-loop runs on host and API is ClusterIP-only: + +```bash +kubectl -n activity-core port-forward svc/actcore-api 8010:8010 +export ACTIVITY_CORE_URL=http://127.0.0.1:8010 +``` + +## Cutover from host timers (T05) + +Host oneshot timers (FI 07:35, Binky rhythm) remain **break-glass** until: + +1. claim-loop is running and healthy +2. **5 clean weekday cycles** (or 5 forced triggers) with open→claimed→succeeded +3. Then disable or demote timers — see `deploy/docs/timer-cutover.md` + +Until then dual-path is intentional: schedule truth is activity-core; timers are +a safety net if claim-loop is down. + +## Labels required on definitions + +| Domain | Labels (min) | +| ------ | ------------ | +| Freedom Intelligence | `automated`, `research-brief` (already on FI rule) | +| Binky daily | `automated`, `rhythm` | +| Binky mail | `automated`, `mail-intake` | diff --git a/docs/task-intake.md b/docs/task-intake.md index ed94d02..5a11864 100644 --- a/docs/task-intake.md +++ b/docs/task-intake.md @@ -1,70 +1,89 @@ -# Task intake (HARNESS-WP-0001-T03) +# Task intake -activity-core emits tasks via IssueSink → `POST /issues/` on **issue-core**. -rein-aharness **consumes** them by polling the same service: +**Architecture:** ACT-ADR-005 · ACTIVITY-WP-0026 · REIN-A-0002 + +## Primary: activity-core `ops_run` (scheduled automation) + +Internal fleet automation (FI daily brief, Binky rhythm, mail intake, …) +emits a claimable **`ops_run`** in activity-core. rein-aharness claims that +row — **not** issue-core and **not** Forgejo. ``` -activity-core (cron/rule) - → IssueCoreRestSink POST /issues/ - → issue-core backend (sqlite / gitea) - → rein-aharness poll/claim GET+PATCH /issues/ - → run_task (profile + budget + persona + session) - → PATCH close + hub progress + .kaizen/metrics +activity-core (Temporal + rules) + → INSERT ops_run (open) + dual-write activity_task_spawn (state-hub) + → rein-aharness claim-loop + POST /ops-runs/claim → approach registry → execute → complete/fail + → domain completion events (fi_daily_brief, binky_daily_brief, …) ``` -`TaskExecutorWorkflow` stays a stub (activity-core INTENT); execution lives here. +Full guide: **`docs/ops-run-claim-loop.md`**. -## Mail path (server / Railiance) +### CLI -``` -rein-aharness mail-scan # deterministic IMAP (AppRole) -rein-aharness mail-triage # llm-connect HTTP → JSON → apply mail-log + commit +```bash +export ACTIVITY_CORE_URL=http://127.0.0.1:8010 +export ACTIVITY_CORE_WORKER_TOKEN=… +export AGENT_HARNESS_REPO_MAP='{"freedom-intelligence":"~/freedom-intelligence","binky-control":"~/binky-control"}' + +rein-aharness poll --source=ops-run --no-claim +rein-aharness run --from-ops-run +rein-aharness claim-loop ``` -Requires `LLM_CONNECT_URL` (e.g. in-cluster -`http://llm-connect.activity-core.svc.cluster.local:8080`). Optional: -`MAIL_TRIAGE_MODEL`, `MAIL_TRIAGE_MAX_TOKENS`, `MAIL_TRIAGE_MAX_ROWS`. -**No Claude CLI** on the host. +`TaskExecutorWorkflow` in activity-core stays a stub; execution lives here. -## issue-core worker API +--- + +## Legacy / external: issue-core poll + +issue-core remains for **external tracker tickets** and intentional +`ISSUE_SINK_TYPE=rest` projections to **Forgejo** (self-hosted forge only — +no Gitea product path). + +``` +activity-core (optional rest sink) + → issue-core POST /issues/ + → rein-aharness poll --source=issue-core + → run_task / close +``` + +Do **not** use this path as the claim queue for FI/Binky ops under WP-0022. + +### issue-core worker API | Method | Path | Role | |--------|------|------| -| POST | `/issues/` | Ingest (activity-core) | +| POST | `/issues/` | Ingest (activity-core rest sink) | | GET | `/issues/?state=open&label=automated` | Poll queue | | GET | `/issues/{id}` | Fetch one | | PATCH | `/issues/{id}` | Claim (`in_progress` + assignee) or `closed` | Auth: `Authorization: Bearer $ISSUE_CORE_API_KEY` or `X-API-Key`. -## CLI - ```bash -export ISSUE_CORE_URL=http://127.0.0.1:8765 # or in-cluster service +export ISSUE_CORE_URL=http://127.0.0.1:8765 export ISSUE_CORE_API_KEY=… -export AGENT_HARNESS_REPO_MAP='{"binky-control":"/home/tegwick/binky-control"}' - -# Peek / claim without executing -rein-aharness poll -rein-aharness poll --no-claim - -# Claim → run → close (or reopen on failure) +rein-aharness poll --source=issue-core rein-aharness run --from-issue-core - -# Local dev (unchanged) -rein-aharness run --task-file examples/task-hello-sandbox.json --no-hub ``` -## Label filter +Default label filter: **`automated`** (`AGENT_HARNESS_INTAKE_LABELS`). -Default: issues must carry the **`automated`** label (matches activity-core -binky definitions). Override: +--- -```bash -export AGENT_HARNESS_INTAKE_LABELS=automated,harness # ALL required +## Mail path (server / Railiance) + +``` +rein-aharness mail-scan # deterministic IMAP (AppRole) +rein-aharness mail-triage # llm-connect HTTP → JSON → apply mail-log + commit +# or via claim-loop when ops_run labels include mail-intake ``` -## Agent + completion event mapping +Requires `LLM_CONNECT_URL`. **No Claude CLI** required for mail/brief approaches. + +--- + +## Agent + completion event mapping (issue-core / agent-session) From `activity_definition_id` or labels: @@ -75,11 +94,12 @@ From `activity_definition_id` or labels: | `binky-weekly-review-prep` / `weekly-review` | `review-prep` | `binky_weekly_review` | | else | `coach` | `executor_run` | -Instance policy (`tool_profile`, `budget`, `lane`) still comes from the -target repo's `.kaizen/schedule.yml`. +Approach registry for ops_run (preferred): see `docs/ops-run-claim-loop.md`. + +--- ## Repo path resolution 1. Absolute/expanded path if it is already a git repo 2. `AGENT_HARNESS_REPO_MAP` JSON -3. Search `AGENT_HARNESS_REPO_ROOTS` (default `~:~/work`) for the slug +3. Search `AGENT_HARNESS_REPO_ROOTS` (default `~:~/work`) for the slug diff --git a/rein_aharness/approaches.py b/rein_aharness/approaches.py new file mode 100644 index 0000000..7de3757 --- /dev/null +++ b/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, + ) + + diff --git a/rein_aharness/claim_loop.py b/rein_aharness/claim_loop.py new file mode 100644 index 0000000..9947800 --- /dev/null +++ b/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 diff --git a/rein_aharness/cli.py b/rein_aharness/cli.py index 4ab0d1d..256f3e3 100644 --- a/rein_aharness/cli.py +++ b/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) diff --git a/rein_aharness/ops_run_client.py b/rein_aharness/ops_run_client.py new file mode 100644 index 0000000..9562bf8 --- /dev/null +++ b/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, + ) diff --git a/tests/test_approaches.py b/tests/test_approaches.py new file mode 100644 index 0000000..7f7b181 --- /dev/null +++ b/tests/test_approaches.py @@ -0,0 +1,92 @@ +"""Pure approach selection tests (REIN-A-0002-T02).""" + +from __future__ import annotations + +from rein_aharness.approaches import ( + APPROACH_AGENT_SESSION, + APPROACH_BRIEF_DAILY, + APPROACH_FI_RESEARCH_BRIEF, + APPROACH_MAIL_PIPELINE, + APPROACH_UNMATCHED, + select_approach, +) +from rein_aharness.ops_run_client import OpsRun + + +def _run(**kwargs) -> OpsRun: + base = dict( + id="r", + activity_definition_id="", + idempotency_key="k", + target_repo="x", + title="", + description="", + labels=[], + ) + base.update(kwargs) + return OpsRun(**base) + + +def test_select_fi_by_label() -> None: + assert ( + select_approach( + _run(labels=["automated", "research-brief", "freedom-intelligence"]) + ) + == APPROACH_FI_RESEARCH_BRIEF + ) + + +def test_select_fi_by_blob() -> None: + assert ( + select_approach( + _run( + labels=["automated"], + activity_definition_id="fi-daily-research-brief", + title="FI daily research brief", + ) + ) + == APPROACH_FI_RESEARCH_BRIEF + ) + + +def test_select_binky_rhythm() -> None: + assert ( + select_approach(_run(labels=["binky", "rhythm", "automated"])) + == APPROACH_BRIEF_DAILY + ) + + +def test_select_mail_intake() -> None: + assert ( + select_approach(_run(labels=["mail-intake", "automated"])) + == APPROACH_MAIL_PIPELINE + ) + + +def test_select_agent_session() -> None: + assert ( + select_approach(_run(labels=["agent-session", "automated"])) + == APPROACH_AGENT_SESSION + ) + + +def test_select_hint_overrides() -> None: + assert ( + select_approach( + _run(labels=["automated"], approach_hint="fi-research-brief") + ) + == APPROACH_FI_RESEARCH_BRIEF + ) + + +def test_unmatched() -> None: + assert select_approach(_run(labels=["automated"], title="misc hygiene")) == ( + APPROACH_UNMATCHED + ) + + +def test_fi_before_generic_automated() -> None: + """research-brief must not fall through to unmatched when only automated.""" + assert ( + select_approach(_run(labels=["research-brief"])) == APPROACH_FI_RESEARCH_BRIEF + ) diff --git a/tests/test_claim_loop.py b/tests/test_claim_loop.py new file mode 100644 index 0000000..0422373 --- /dev/null +++ b/tests/test_claim_loop.py @@ -0,0 +1,97 @@ +"""Claim loop process_one tests (REIN-A-0002-T03).""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +from rein_aharness.approaches import ApproachResult, APPROACH_FI_RESEARCH_BRIEF +from rein_aharness.claim_loop import process_one, poll_peek +from rein_aharness.ops_run_client import OpsRun, OpsRunConfig, ActivityCoreOpsClient + + +def _claimed_run() -> OpsRun: + return OpsRun( + id="run-1", + activity_definition_id="def", + idempotency_key="k", + target_repo="freedom-intelligence", + title="FI daily", + description="d", + labels=["automated", "research-brief"], + state="claimed", + claim_owner="worker-1", + attempt=1, + ) + + +def test_process_one_empty() -> None: + client = MagicMock(spec=ActivityCoreOpsClient) + client.config = OpsRunConfig(worker_id="w", lease_seconds=90) + client.claim.return_value = [] + r = process_one(client) + assert r.empty is True + assert r.claimed is False + + +def test_process_one_success_completes() -> None: + client = MagicMock(spec=ActivityCoreOpsClient) + client.config = OpsRunConfig(worker_id="w", lease_seconds=90) + client.claim.return_value = [_claimed_run()] + client.complete.return_value = OpsRun( + id="run-1", + activity_definition_id="def", + idempotency_key="k", + target_repo="freedom-intelligence", + title="FI daily", + description="", + state="succeeded", + labels=["automated", "research-brief"], + ) + ar = ApproachResult( + ok=True, + approach=APPROACH_FI_RESEARCH_BRIEF, + result={"path": "briefs/x.md"}, + reason="ok", + ) + with patch("rein_aharness.claim_loop.execute_approach", return_value=ar): + r = process_one(client) + assert r.claimed is True + assert r.ok is True + assert r.approach == APPROACH_FI_RESEARCH_BRIEF + client.complete.assert_called_once() + client.fail.assert_not_called() + + +def test_process_one_failure_reopens() -> None: + client = MagicMock(spec=ActivityCoreOpsClient) + client.config = OpsRunConfig(worker_id="w", lease_seconds=90) + client.claim.return_value = [_claimed_run()] + client.fail.return_value = OpsRun( + id="run-1", + activity_definition_id="def", + idempotency_key="k", + target_repo="freedom-intelligence", + title="t", + description="", + state="open", + labels=["automated", "research-brief"], + ) + ar = ApproachResult( + ok=False, + approach=APPROACH_FI_RESEARCH_BRIEF, + reason="llm timeout", + reopen=True, + ) + with patch("rein_aharness.claim_loop.execute_approach", return_value=ar): + r = process_one(client) + assert r.ok is False + client.fail.assert_called_once() + assert client.fail.call_args.kwargs["reopen"] is True + + +def test_poll_peek() -> None: + client = MagicMock(spec=ActivityCoreOpsClient) + client.list_open.return_value = [_claimed_run()] + rows = poll_peek(client) + assert len(rows) == 1 + assert rows[0]["approach"] == APPROACH_FI_RESEARCH_BRIEF diff --git a/tests/test_ops_run_client.py b/tests/test_ops_run_client.py new file mode 100644 index 0000000..b77a5b4 --- /dev/null +++ b/tests/test_ops_run_client.py @@ -0,0 +1,168 @@ +"""Tests for activity-core ops_run client (REIN-A-0002-T01).""" + +from __future__ import annotations + +import json +from pathlib import Path +from unittest.mock import MagicMock, patch + +import httpx +import pytest + +from rein_aharness.ops_run_client import ( + ActivityCoreOpsClient, + OpsRun, + OpsRunConfig, + OpsRunError, + ops_run_to_taskspec, +) +from rein_aharness.taskspec import TaskSpecError + + +def test_ops_run_from_api() -> None: + row = OpsRun.from_api( + { + "id": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", + "activity_definition_id": "11111111-2222-3333-4444-555555555555", + "idempotency_key": "k", + "target_repo": "freedom-intelligence", + "title": "FI daily", + "description": "d", + "labels": ["automated", "research-brief"], + "state": "open", + "attempt": 0, + } + ) + assert row.target_repo == "freedom-intelligence" + assert "research-brief" in row.labels + + +def test_config_from_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("ACTIVITY_CORE_URL", "http://actcore:8010/") + monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN", "tok") + monkeypatch.setenv("AGENT_HARNESS_WORKER_ID", "w@host") + monkeypatch.setenv("AGENT_HARNESS_OPS_LABELS", "automated,research-brief") + monkeypatch.setenv( + "AGENT_HARNESS_REPO_MAP", + json.dumps({"freedom-intelligence": "/tmp/fi"}), + ) + cfg = OpsRunConfig.from_env() + assert cfg.base_url == "http://actcore:8010" + assert cfg.worker_token == "tok" + assert cfg.worker_id == "w@host" + assert "research-brief" in cfg.claim_labels + assert cfg.repo_map["freedom-intelligence"] == "/tmp/fi" + + +def test_claim_posts_body() -> None: + cfg = OpsRunConfig( + base_url="http://example.test", + worker_token="secret", + worker_id="worker-1", + claim_labels=("automated",), + lease_seconds=120, + ) + client = ActivityCoreOpsClient(cfg) + mock_resp = MagicMock() + mock_resp.raise_for_status = MagicMock() + mock_resp.json.return_value = { + "items": [ + { + "id": "r1", + "activity_definition_id": "d1", + "idempotency_key": "k", + "target_repo": "freedom-intelligence", + "title": "t", + "description": "", + "labels": ["automated", "research-brief"], + "state": "claimed", + "claim_owner": "worker-1", + "attempt": 1, + } + ] + } + with patch("rein_aharness.ops_run_client.httpx.post", return_value=mock_resp) as post: + claimed = client.claim(limit=1) + assert len(claimed) == 1 + assert claimed[0].id == "r1" + assert claimed[0].state == "claimed" + kwargs = post.call_args.kwargs + assert kwargs["json"]["worker_id"] == "worker-1" + assert kwargs["json"]["labels"] == ["automated"] + assert kwargs["headers"]["X-Worker-Token"] == "secret" + + +def test_complete_and_fail() -> None: + cfg = OpsRunConfig(base_url="http://example.test", worker_id="w") + client = ActivityCoreOpsClient(cfg) + mock_resp = MagicMock() + mock_resp.raise_for_status = MagicMock() + mock_resp.json.return_value = { + "id": "r1", + "activity_definition_id": "d", + "idempotency_key": "k", + "title": "t", + "description": "", + "state": "succeeded", + "labels": [], + } + with patch("rein_aharness.ops_run_client.httpx.post", return_value=mock_resp): + out = client.complete("r1", result={"path": "x"}) + assert out.state == "succeeded" + + mock_resp.json.return_value = {**mock_resp.json.return_value, "state": "open"} + with patch("rein_aharness.ops_run_client.httpx.post", return_value=mock_resp) as post: + out = client.fail("r1", error="timeout", reopen=True) + assert post.call_args.kwargs["json"]["reopen"] is True + + +def test_claim_http_error() -> None: + cfg = OpsRunConfig(base_url="http://example.test", worker_id="w") + client = ActivityCoreOpsClient(cfg) + with patch( + "rein_aharness.ops_run_client.httpx.post", + side_effect=httpx.ConnectError("down"), + ): + with pytest.raises(OpsRunError, match="claim failed"): + client.claim() + + +def test_ops_run_to_taskspec(tmp_path: Path) -> None: + import subprocess + + repo = tmp_path / "freedom-intelligence" + repo.mkdir() + subprocess.run(["git", "init", "-q"], cwd=repo, check=True) + (repo / "README.md").write_text("x\n") + subprocess.run(["git", "add", "."], cwd=repo, check=True) + subprocess.run( + ["git", "-c", "user.email=t@t", "-c", "user.name=t", "commit", "-qm", "i"], + cwd=repo, + check=True, + ) + run = OpsRun( + id="r1", + activity_definition_id="d", + idempotency_key="k", + target_repo="freedom-intelligence", + title="FI", + description="do", + labels=["automated"], + ) + cfg = OpsRunConfig(repo_roots=(str(tmp_path),)) + spec = ops_run_to_taskspec(run, cfg) + assert spec.target_repo == repo.resolve() + assert spec.hub_task_id == "r1" + + +def test_ops_run_to_taskspec_missing_repo() -> None: + run = OpsRun( + id="r1", + activity_definition_id="d", + idempotency_key="k", + target_repo=None, + title="t", + description="", + ) + with pytest.raises(TaskSpecError): + ops_run_to_taskspec(run, OpsRunConfig()) diff --git a/workplans/REIN-A-0002-ops-run-claim-loop.md b/workplans/REIN-A-0002-ops-run-claim-loop.md index ad58446..4e8f38d 100644 --- a/workplans/REIN-A-0002-ops-run-claim-loop.md +++ b/workplans/REIN-A-0002-ops-run-claim-loop.md @@ -4,7 +4,7 @@ type: workplan title: "Ops run claim loop and approach selection" domain: agents repo: rein-aharness -status: ready +status: in_progress owner: grok topic_slug: rein-aharness priority: high @@ -40,7 +40,7 @@ ops under WP-0022 and issue-core INTENT. ```task id: REIN-A-0002-T01 -status: todo +status: done priority: high state_hub_task_id: "a5e32652-ba3f-4cc5-8d41-0d9f3022c50e" ``` @@ -62,7 +62,7 @@ Map ops_run → existing `TaskSpec` (or thin OpsRunSpec). Tests with httpx mock. ```task id: REIN-A-0002-T02 -status: todo +status: done priority: high state_hub_task_id: "c307b119-1bff-4d5c-a911-5b5259de5c86" ``` @@ -88,7 +88,7 @@ agent session; structured llm-connect before open-ended coding agent). ```task id: REIN-A-0002-T03 -status: todo +status: done priority: high state_hub_task_id: "31fb7df3-83a3-4f18-a7b9-9be49e226837" ``` @@ -107,17 +107,17 @@ state_hub_task_id: "31fb7df3-83a3-4f18-a7b9-9be49e226837" ```task id: REIN-A-0002-T04 -status: todo +status: done priority: high state_hub_task_id: "93190150-4df7-47f4-b52e-29b997f9aaf0" ``` -- Ensure definition labels include approach keys (`research-brief`, `automated`, …). -- Wire `AGENT_HARNESS_REPO_MAP` / roots for `freedom-intelligence`, `binky-control`. -- Document railiance install of claim-loop service. +- FI labels already include `research-brief` + `automated`. +- `AGENT_HARNESS_REPO_MAP` documented in claim-loop.env.example + ops-run-claim-loop.md. +- Install: `deploy/scripts/install-claim-loop-user.sh`. -**Done when:** end-to-end: trigger FI def → ops_run claimed → brief path + -`fi_daily_brief` (or skip if not due). +**Prod e2e** (operator after WP-0026-T07 migrate): trigger FI → claim-loop once → +`fi_daily_brief`. Code path + install docs landed. --- @@ -130,12 +130,14 @@ priority: medium state_hub_task_id: "a04a98e3-ca6a-4a18-b6fd-47a77047ca90" ``` -After **5 clean weekday cycles** (or 5 forced triggers): +**Procedure landed:** `deploy/docs/timer-cutover.md` + FI `docs/recurrence-ops.md` +updated for claim-loop primary / timer break-glass. -1. Disable FI + Binky host oneshot timers **or** set them to “poll once if queue - backlog” only. -2. Update domain `OperatingRhythm.md` / `docs/recurrence-ops.md`. -3. Keep install scripts with break-glass section. +Still open on railiance: + +1. Fill proof table (5 clean cycles). +2. Disable FI + Binky host oneshot timers. +3. Keep install scripts for break-glass. **Done when:** docs + railiance state match; dual-clock residual closed for FI/Binky. @@ -145,7 +147,7 @@ After **5 clean weekday cycles** (or 5 forced triggers): ```task id: REIN-A-0002-T06 -status: todo +status: done priority: medium state_hub_task_id: "cca10835-d6fc-4ce1-a361-d2d1f2693872" ``` @@ -158,10 +160,13 @@ state_hub_task_id: "cca10835-d6fc-4ce1-a361-d2d1f2693872" ## Acceptance -- [ ] Claim loop does not require Forgejo or issue-core for FI/Binky ops -- [ ] Approach table selects fi-research-brief / brief-daily without host “schedule” -- [ ] Timers demoted after proof window -- [ ] ACT-ADR-005 / ACTIVITY-WP-0026 contracts implemented +- [x] Claim loop does not require Forgejo or issue-core for FI/Binky ops +- [x] Approach table selects fi-research-brief / brief-daily without host “schedule” +- [ ] Timers demoted after proof window (**T05** operator) +- [x] ACT-ADR-005 / ACTIVITY-WP-0026 contracts implemented + +**Shipped:** `ops_run_client`, `approaches`, `claim_loop`, CLI `claim-loop` / +`run --from-ops-run` / `poll --source=ops-run`, unit tests, install + cutover docs. ## Out of scope