coulomb/rein-aharness
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: issue-core task intake poll/claim/run (HARNESS-WP-0001-T03)

Browse source 
Add intake client mapping emissions to TaskSpec, CLI poll and
run --from-issue-core, keep --task-file for local dev. Completes
HARNESS-WP-0001 workplan.
This commit is contained in:
tegwick 2026-07-18 11:30:40 +02:00
parent 154e535695
commit 66ccd4fa01
7 changed files with 705 additions and 34 deletions
Download patch file Download diff file Expand all files Collapse all files

161
agent_harness/cli.py
View file

@ -68,12 +68,123 @@ def _cmd_profiles(_args: argparse.Namespace) -> int:
return 0
def _cmd_poll(args: argparse.Namespace) -> int:
from agent_harness.intake import IntakeError, IssueCoreClient, poll_next
try:
client = IssueCoreClient()
result = poll_next(client, claim=not args.no_claim)
except IntakeError as exc:
print(f"intake error: {exc}", file=sys.stderr)
return 2
if result is None:
print(json.dumps({"queue": "empty"}, indent=2))
return 0
issue, spec = result
print(
json.dumps(
{
"issue_id": issue.issue_id,
"state": issue.state,
"title": issue.title,
"agent": spec.agent,
"target_repo": str(spec.target_repo),
"completion_event_type": spec.completion_event_type,
"labels": spec.labels,
"claimed": not args.no_claim,
},
indent=2,
)
)
return 0
def _cmd_run(args: argparse.Namespace) -> int:
from agent_harness.intake import IntakeError, IssueCoreClient, poll_next
issue_id: str | None = None
client: IssueCoreClient | None = None
if args.from_issue_core:
try:
client = IssueCoreClient()
polled = poll_next(client, claim=True)
except IntakeError as exc:
print(f"intake error: {exc}", file=sys.stderr)
return 2
if polled is None:
print(json.dumps({"ok": True, "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",
file=sys.stderr,
)
return 2
try:
spec = TaskSpec.from_file(args.task_file)
except (TaskSpecError, OSError, json.JSONDecodeError) as exc:
print(f"invalid task spec: {exc}", file=sys.stderr)
return 2
result = run_task(
spec,
report_to_hub=not args.no_hub,
write_metrics=not args.no_metrics,
)
closed = False
close_error = ""
if client is not None and issue_id:
try:
if result.ok:
client.close(issue_id)
closed = True
else:
client.reopen(issue_id)
except IntakeError as exc:
close_error = str(exc)
print(
json.dumps(
{
"ok": result.ok,
"committed": result.committed,
"head_after": result.head_after,
"persona_source": result.persona_source,
"tool_profile": result.tool_profile,
"budget_tokens": result.budget_tokens,
"tokens_spent": result.tokens_spent,
"execution_time_s": round(result.execution_time_s, 3),
"reason": result.reason,
"issue_id": issue_id,
"issue_closed": closed,
"issue_close_error": close_error,
},
indent=2,
)
)
return 0 if result.ok else 1
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(prog="agent-harness")
sub = parser.add_subparsers(dest="command", required=True)
run = sub.add_parser("run", help="Execute one task from a JSON spec file")
run.add_argument("--task-file", required=True)
run = sub.add_parser(
"run",
help="Execute one task from a JSON file or next issue-core emission",
)
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-issue-core",
action="store_true",
help="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-metrics",
@ -127,6 +238,16 @@ def main(argv: list[str] | None = None) -> int:
smoke.add_argument("--no-hub", action="store_true", help="Skip hub reporting")
smoke.add_argument("--no-push", action="store_true", help="Skip git push after commit")
poll = sub.add_parser(
"poll",
help="Peek/claim next open issue-core task labeled for the harness (no execute)",
)
poll.add_argument(
"--no-claim",
action="store_true",
help="List/map only; do not set in_progress",
)
args = parser.parse_args(argv)
if args.command == "validate":
@ -135,6 +256,12 @@ def main(argv: list[str] | None = None) -> int:
if args.command == "profiles":
return _cmd_profiles(args)
if args.command == "poll":
return _cmd_poll(args)
if args.command == "run":
return _cmd_run(args)
if args.command == "smoke":
from agent_harness.smoke import ensure_sandbox_clone, run_smoke
@ -201,34 +328,8 @@ def main(argv: list[str] | None = None) -> int:
)
return 0 if result.ok else 1
try:
spec = TaskSpec.from_file(args.task_file)
except (TaskSpecError, OSError, json.JSONDecodeError) as exc:
print(f"invalid task spec: {exc}", file=sys.stderr)
return 2
result = run_task(
spec,
report_to_hub=not args.no_hub,
write_metrics=not args.no_metrics,
)
print(
json.dumps(
{
"ok": result.ok,
"committed": result.committed,
"head_after": result.head_after,
"persona_source": result.persona_source,
"tool_profile": result.tool_profile,
"budget_tokens": result.budget_tokens,
"tokens_spent": result.tokens_spent,
"execution_time_s": round(result.execution_time_s, 3),
"reason": result.reason,
},
indent=2,
)
)
return 0 if result.ok else 1
print(f"unknown command: {args.command}", file=sys.stderr)
return 2
if __name__ == "__main__":

287
agent_harness/intake.py Normal file
View file

@ -0,0 +1,287 @@
"""Task intake from issue-core REST (activity-core emission sink).
Polls open issues labeled for the harness, maps them to TaskSpec, claims
them (in_progress + assignee), and after a successful run closes them.
Environment:
ISSUE_CORE_URL default http://127.0.0.1:8765
ISSUE_CORE_API_KEY required for live poll
AGENT_HARNESS_INTAKE_LABELS comma list; issue must have ALL (default: automated)
AGENT_HARNESS_ASSIGNEE claim assignee (default: agent-harness)
AGENT_HARNESS_REPO_MAP JSON object name→path, e.g. {"binky-control":"~/binky-control"}
AGENT_HARNESS_REPO_ROOTS colon-separated search roots (default: ~:~/work)
Local development still uses `agent-harness run --task-file …`.
"""
from __future__ import annotations
import json
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
import httpx
from agent_harness.taskspec import TaskSpec, TaskSpecError
DEFAULT_ISSUE_CORE_URL = "http://127.0.0.1:8765"
DEFAULT_INTAKE_LABELS = ("automated",)
DEFAULT_ASSIGNEE = "agent-harness"
DEFAULT_REPO_ROOTS = ("~", "~/work")
# activity-definition / label → (agent instance, completion_event_type)
_DEFINITION_HINTS: dict[str, tuple[str, str]] = {
"binky-daily-rhythm": ("coach", "binky_daily_brief"),
"binky-weekly-mail-intake": ("mail-triage", "binky_mail_intake"),
"binky-weekly-review-prep": ("review-prep", "binky_weekly_review"),
"daily_brief": ("coach", "binky_daily_brief"),
"mail_intake": ("mail-triage", "binky_mail_intake"),
"weekly_review": ("review-prep", "binky_weekly_review"),
"rhythm": ("coach", "binky_daily_brief"),
"mail-intake": ("mail-triage", "binky_mail_intake"),
"weekly-review": ("review-prep", "binky_weekly_review"),
}
class IntakeError(RuntimeError):
pass
@dataclass
class EmittedIssue:
"""Normalized issue-core issue for harness consumption."""
issue_id: str
title: str
description: str
labels: list[str] = field(default_factory=list)
target_repo: str | None = None
priority: str | None = None
activity_definition_id: str | None = None
source_id: str | None = None
triggering_event_id: str | None = None
state: str = "open"
number: int = 0
raw: dict[str, Any] = field(default_factory=dict)
@classmethod
def from_api(cls, data: dict[str, Any]) -> "EmittedIssue":
return cls(
issue_id=str(data.get("issue_id") or data.get("id") or ""),
title=str(data.get("title") or ""),
description=str(data.get("description") or ""),
labels=[str(x) for x in (data.get("labels") or [])],
target_repo=data.get("target_repo"),
priority=data.get("priority"),
activity_definition_id=data.get("activity_definition_id"),
source_id=data.get("source_id"),
triggering_event_id=data.get("triggering_event_id"),
state=str(data.get("state") or "open"),
number=int(data.get("number") or 0),
raw=data,
)
@dataclass
class IntakeConfig:
base_url: str = DEFAULT_ISSUE_CORE_URL
api_key: str = ""
require_labels: tuple[str, ...] = DEFAULT_INTAKE_LABELS
assignee: str = DEFAULT_ASSIGNEE
repo_map: dict[str, str] = field(default_factory=dict)
repo_roots: tuple[str, ...] = DEFAULT_REPO_ROOTS
timeout: float = 15.0
@classmethod
def from_env(cls) -> "IntakeConfig":
labels_raw = os.environ.get("AGENT_HARNESS_INTAKE_LABELS", "automated")
labels = tuple(part.strip() for part in labels_raw.split(",") if part.strip())
roots_raw = os.environ.get("AGENT_HARNESS_REPO_ROOTS", "~:~/work")
roots = tuple(part.strip() for part in roots_raw.split(":") if part.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()}
return cls(
base_url=os.environ.get("ISSUE_CORE_URL", DEFAULT_ISSUE_CORE_URL).rstrip("/"),
api_key=os.environ.get("ISSUE_CORE_API_KEY", "").strip(),
require_labels=labels or DEFAULT_INTAKE_LABELS,
assignee=os.environ.get("AGENT_HARNESS_ASSIGNEE", DEFAULT_ASSIGNEE),
repo_map=repo_map,
repo_roots=roots or DEFAULT_REPO_ROOTS,
)
class IssueCoreClient:
"""Thin REST client for issue-core poll/claim/close."""
def __init__(self, config: IntakeConfig | None = None):
self.config = config or IntakeConfig.from_env()
def _headers(self) -> dict[str, str]:
if not self.config.api_key:
raise IntakeError(
"ISSUE_CORE_API_KEY is not set (required to poll issue-core)"
)
return {
"Authorization": f"Bearer {self.config.api_key}",
"Accept": "application/json",
}
def list_open(
self,
*,
labels: tuple[str, ...] | None = None,
limit: int = 50,
) -> list[EmittedIssue]:
want = labels if labels is not None else self.config.require_labels
params: list[tuple[str, str]] = [("state", "open"), ("limit", str(limit))]
for lab in want:
params.append(("label", lab))
try:
resp = httpx.get(
f"{self.config.base_url}/issues/",
params=params,
headers=self._headers(),
timeout=self.config.timeout,
)
resp.raise_for_status()
except httpx.HTTPError as exc:
raise IntakeError(f"list issues failed: {exc}") from exc
data = resp.json()
if not isinstance(data, list):
raise IntakeError(f"unexpected list payload type: {type(data)}")
return [EmittedIssue.from_api(item) for item in data]
def claim(self, issue_id: str) -> EmittedIssue:
return self._patch(
issue_id,
{"state": "in_progress", "assignee": self.config.assignee},
)
def close(self, issue_id: str) -> EmittedIssue:
return self._patch(issue_id, {"state": "closed"})
def reopen(self, issue_id: str) -> EmittedIssue:
return self._patch(issue_id, {"state": "open", "assignee": ""})
def _patch(self, issue_id: str, body: dict[str, Any]) -> EmittedIssue:
try:
resp = httpx.patch(
f"{self.config.base_url}/issues/{issue_id}",
json=body,
headers=self._headers(),
timeout=self.config.timeout,
)
resp.raise_for_status()
except httpx.HTTPError as exc:
raise IntakeError(f"patch issue {issue_id} failed: {exc}") from exc
return EmittedIssue.from_api(resp.json())
def resolve_target_repo(
name: str,
*,
repo_map: dict[str, str] | None = None,
repo_roots: tuple[str, ...] = DEFAULT_REPO_ROOTS,
) -> Path:
"""Map emission target_repo slug/path to a local git checkout."""
raw = (name or "").strip()
if not raw:
raise TaskSpecError("target_repo is empty")
# Absolute or explicit path
direct = Path(raw).expanduser()
if direct.is_dir() and (direct / ".git").is_dir():
return direct.resolve()
# Strip org prefix: coulomb/binky-control → binky-control
slug = raw.split("/")[-1]
mapping = repo_map or {}
if slug in mapping:
path = Path(mapping[slug]).expanduser()
if (path / ".git").is_dir():
return path.resolve()
raise TaskSpecError(f"mapped target_repo not a git repo: {path}")
if raw in mapping:
path = Path(mapping[raw]).expanduser()
if (path / ".git").is_dir():
return path.resolve()
for root in repo_roots:
candidate = Path(root).expanduser() / slug
if (candidate / ".git").is_dir():
return candidate.resolve()
raise TaskSpecError(
f"cannot resolve target_repo '{name}' under {list(repo_roots)}; "
"set AGENT_HARNESS_REPO_MAP or clone the repo"
)
def infer_agent_and_event(issue: EmittedIssue) -> tuple[str, str]:
"""Return (agent_instance_name, completion_event_type)."""
keys: list[str] = []
if issue.activity_definition_id:
keys.append(issue.activity_definition_id)
keys.extend(issue.labels)
blob = " ".join(keys).lower()
for hint, mapped in _DEFINITION_HINTS.items():
if hint.lower() in blob:
return mapped
# explicit agent:name label
for lab in issue.labels:
if lab.startswith("agent:"):
return lab.split(":", 1)[1], "executor_run"
return "coach", "executor_run"
def issue_to_taskspec(issue: EmittedIssue, config: IntakeConfig | None = None) -> TaskSpec:
cfg = config or IntakeConfig.from_env()
if not issue.target_repo:
raise TaskSpecError(
f"issue {issue.issue_id} missing target_repo (ingestion metadata)"
)
target = resolve_target_repo(
issue.target_repo,
repo_map=cfg.repo_map,
repo_roots=cfg.repo_roots,
)
agent, event = infer_agent_and_event(issue)
timeout = 900
return TaskSpec(
title=issue.title,
description=issue.description,
target_repo=target,
agent=agent,
labels=list(issue.labels),
hub_task_id=None,
completion_event_type=event,
timeout_seconds=timeout,
)
def poll_next(
client: IssueCoreClient | None = None,
*,
claim: bool = True,
) -> tuple[EmittedIssue, TaskSpec] | None:
"""Poll one open harness-labeled issue; optionally claim it.
Returns None when the queue is empty.
"""
client = client or IssueCoreClient()
issues = client.list_open()
if not issues:
return None
# Prefer oldest by number when available
issues_sorted = sorted(issues, key=lambda i: (i.number or 0, i.issue_id))
issue = issues_sorted[0]
if claim:
issue = client.claim(issue.issue_id)
spec = issue_to_taskspec(issue, client.config)
return issue, spec