feat(api): GET/PATCH /issues for worker poll and claim

Expose list/get/claim/close for agent-harness intake against the same
auth as POST /issues/ ingestion (HARNESS-WP-0001-T03).

SCOPE.md

@@ -39,6 +39,8 @@ which explains *why*; this file states *what* and *what not*.
 - **CLI** (`issue` / `issue-core`) for humans and agents on a shell.
 - **REST** (`POST /issues/`) for automation — primarily activity-core's
   `IssueSink`, but open to any well-authenticated client.
+- **REST worker surface** (`GET/PATCH /issues/`) for executors (agent-harness):
+  list open issues by label/state, claim (`in_progress` + assignee), close.
 - **NATS subscriber** (design stub only — implementation deferred until
   activity-core migrates from REST to NATS, see `docs/nats-task-ingestion.md`).
 

issue_core/api/app.py

@@ -6,6 +6,7 @@ from fastapi import FastAPI
 
 from .. import __version__
 from .ingest import router as ingest_router
+from .query import router as query_router
 
 
 def create_app() -> FastAPI:
@@ -13,11 +14,13 @@ def create_app() -> FastAPI:
         title="issue-core",
         description=(
             "Authoritative task lifecycle manager for the Coulomb org. "
-            "POST /issues/ is the ingestion surface for activity-core's IssueSink."
+            "POST /issues/ is the ingestion surface for activity-core's IssueSink; "
+            "GET/PATCH /issues/ is the worker poll/claim surface for agent-harness."
         ),
         version=__version__,
     )
     app.include_router(ingest_router)
+    app.include_router(query_router)
 
     @app.get("/healthz", tags=["meta"])
     async def healthz() -> dict:

issue_core/api/query.py

@@ -0,0 +1,211 @@
+"""
+Worker-facing issue query and claim endpoints.
+
+Complements POST /issues/ (ingestion) so executors like agent-harness can:
+
+  GET  /issues/              — list open (or filtered) issues
+  GET  /issues/{issue_id}    — fetch one issue with ingestion metadata
+  PATCH /issues/{issue_id}   — claim (in_progress) or close
+
+Auth: same shared ISSUE_CORE_API_KEY as ingestion.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from pydantic import BaseModel, Field
+
+from ..core.interfaces import IssueFilter
+from ..core.models import Issue, IssueState, Label, User
+from .auth import require_api_key
+from .ingest import _resolve_backend
+
+router = APIRouter()
+
+
+class IssueOut(BaseModel):
+    """Stable JSON shape for list/get (harness maps this to TaskSpec)."""
+
+    issue_id: str
+    number: int
+    title: str
+    description: str
+    state: str
+    labels: List[str] = Field(default_factory=list)
+    target_repo: Optional[str] = None
+    priority: Optional[str] = None
+    source_type: Optional[str] = None
+    source_id: Optional[str] = None
+    triggering_event_id: Optional[str] = None
+    activity_definition_id: Optional[str] = None
+    assignee: Optional[str] = None
+    created_at: Optional[str] = None
+    updated_at: Optional[str] = None
+    issue_url: Optional[str] = None
+
+
+class IssuePatch(BaseModel):
+    state: Optional[str] = Field(
+        default=None,
+        description="open | in_progress | closed | blocked",
+    )
+    assignee: Optional[str] = Field(
+        default=None,
+        description="Username to assign (claim); empty string clears",
+    )
+
+
+def _issue_to_out(issue: Issue) -> IssueOut:
+    labels = [label.name for label in (issue.labels or [])]
+    ingestion: Dict[str, Any] = {}
+    if issue.sync_metadata and isinstance(issue.sync_metadata, dict):
+        raw = issue.sync_metadata.get("ingestion") or {}
+        if isinstance(raw, dict):
+            ingestion = raw
+    priority = None
+    for name in labels:
+        if name.startswith("priority:"):
+            priority = name.split(":", 1)[1]
+            break
+    assignee = None
+    if issue.assignees:
+        assignee = issue.assignees[0].username
+    issue_url = None
+    if issue.sync_metadata:
+        issue_url = issue.sync_metadata.get("url") or issue.sync_metadata.get("html_url")
+    issue_id = issue.id or str(issue.number)
+    return IssueOut(
+        issue_id=issue_id,
+        number=issue.number,
+        title=issue.title,
+        description=issue.description or "",
+        state=issue.state.value if isinstance(issue.state, IssueState) else str(issue.state),
+        labels=labels,
+        target_repo=ingestion.get("target_repo"),
+        priority=priority,
+        source_type=ingestion.get("source_type"),
+        source_id=ingestion.get("source_id"),
+        triggering_event_id=ingestion.get("triggering_event_id"),
+        activity_definition_id=ingestion.get("activity_definition_id"),
+        assignee=assignee,
+        created_at=issue.created_at.isoformat() if issue.created_at else None,
+        updated_at=issue.updated_at.isoformat() if issue.updated_at else None,
+        issue_url=issue_url,
+    )
+
+
+def _with_backend():
+    backend, _backend_type = _resolve_backend()
+    return backend
+
+
+@router.get(
+    "/issues/",
+    response_model=List[IssueOut],
+    dependencies=[Depends(require_api_key)],
+    summary="List issues for workers (filter by state and labels).",
+)
+async def list_issues(
+    state: Optional[str] = Query(default="open"),
+    label: Optional[List[str]] = Query(default=None),
+    limit: int = Query(default=50, ge=1, le=200),
+) -> List[IssueOut]:
+    backend = _with_backend()
+    try:
+        # Fetch a wider page when label-filtering client-side (local backend
+        # does not always apply label predicates in SQL).
+        want_labels = list(label) if label else []
+        fetch_limit = max(limit * 5, limit) if want_labels else limit
+        criteria = IssueFilter(
+            state=state if state and state != "all" else None,
+            labels=want_labels or None,
+            limit=fetch_limit,
+        )
+        issues = backend.list_issues(criteria)
+        if want_labels:
+            required = set(want_labels)
+            issues = [
+                issue
+                for issue in issues
+                if required.issubset({lab.name for lab in (issue.labels or [])})
+            ]
+        return [_issue_to_out(issue) for issue in issues[:limit]]
+    finally:
+        try:
+            backend.disconnect()
+        except Exception:
+            pass
+
+
+@router.get(
+    "/issues/{issue_id}",
+    response_model=IssueOut,
+    dependencies=[Depends(require_api_key)],
+    summary="Get one issue by id.",
+)
+async def get_issue(issue_id: str) -> IssueOut:
+    backend = _with_backend()
+    try:
+        issue = backend.get_issue(issue_id)
+        if issue is None:
+            # Fall back to number lookup for backends that key by number.
+            try:
+                issue = backend.get_issue_by_number(int(issue_id))
+            except (TypeError, ValueError):
+                issue = None
+        if issue is None:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="issue not found")
+        return _issue_to_out(issue)
+    finally:
+        try:
+            backend.disconnect()
+        except Exception:
+            pass
+
+
+@router.patch(
+    "/issues/{issue_id}",
+    response_model=IssueOut,
+    dependencies=[Depends(require_api_key)],
+    summary="Claim (in_progress + assignee) or close an issue.",
+)
+async def patch_issue(issue_id: str, body: IssuePatch) -> IssueOut:
+    backend = _with_backend()
+    try:
+        issue = backend.get_issue(issue_id)
+        if issue is None:
+            try:
+                issue = backend.get_issue_by_number(int(issue_id))
+            except (TypeError, ValueError):
+                issue = None
+        if issue is None:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="issue not found")
+
+        if body.state is not None:
+            try:
+                issue.state = IssueState.from_string(body.state)
+            except Exception as exc:
+                raise HTTPException(
+                    status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+                    detail=f"invalid state: {body.state}",
+                ) from exc
+
+        if body.assignee is not None:
+            if body.assignee == "":
+                issue.assignees = []
+            else:
+                issue.assignees = [
+                    User(id=body.assignee, username=body.assignee)
+                ]
+
+        issue.updated_at = datetime.now(timezone.utc)
+        updated = backend.update_issue(issue)
+        return _issue_to_out(updated)
+    finally:
+        try:
+            backend.disconnect()
+        except Exception:
+            pass

tests/test_api_query.py

@@ -0,0 +1,111 @@
+"""Tests for GET/PATCH /issues/ worker poll/claim surface."""
+
+from __future__ import annotations
+
+import uuid
+
+import pytest
+
+pytest.importorskip("fastapi")
+pytest.importorskip("httpx")
+
+from fastapi.testclient import TestClient
+
+from issue_core.api.app import create_app
+
+API_KEY = "test-key-not-a-real-secret-only-for-pytest"
+
+
+@pytest.fixture
+def tmp_issue_store(monkeypatch, tmp_path):
+    config_dir = tmp_path / "issue-core"
+    config_dir.mkdir()
+    db_path = str(config_dir / "issues.db")
+    configs = {"default": "local", "local": {"type": "local", "db_path": db_path}}
+    monkeypatch.setattr(
+        "issue_core.cli.utils.get_config_dir", lambda: config_dir, raising=True
+    )
+    monkeypatch.setattr(
+        "issue_core.api.ingest.get_config_dir", lambda: config_dir, raising=True
+    )
+    monkeypatch.setattr(
+        "issue_core.cli.utils.load_backend_configs", lambda: configs, raising=True
+    )
+    monkeypatch.setattr(
+        "issue_core.api.ingest.load_backend_configs", lambda: configs, raising=True
+    )
+    return config_dir
+
+
+@pytest.fixture
+def client(monkeypatch, tmp_issue_store):
+    monkeypatch.setenv("ISSUE_CORE_API_KEY", API_KEY)
+    return TestClient(create_app())
+
+
+def _payload(**overrides):
+    base = {
+        "title": "Run Binky daily rhythm",
+        "description": "Queue hygiene + brief",
+        "target_repo": "binky-control",
+        "priority": "medium",
+        "labels": ["binky", "rhythm", "automated"],
+        "source_type": "rule",
+        "source_id": "emit-daily-rhythm-task",
+        "triggering_event_id": str(uuid.uuid4()),
+        "activity_definition_id": "binky-daily-rhythm",
+    }
+    base.update(overrides)
+    return base
+
+
+def _auth():
+    return {"Authorization": f"Bearer {API_KEY}"}
+
+
+def test_list_and_claim_flow(client):
+    created = client.post("/issues/", json=_payload(), headers=_auth())
+    assert created.status_code == 201, created.text
+    issue_id = created.json()["issue_id"]
+
+    listed = client.get("/issues/", params={"state": "open"}, headers=_auth())
+    assert listed.status_code == 200
+    items = listed.json()
+    assert len(items) == 1
+    assert items[0]["issue_id"] == issue_id
+    assert items[0]["target_repo"] == "binky-control"
+    assert "automated" in items[0]["labels"]
+    assert items[0]["activity_definition_id"] == "binky-daily-rhythm"
+
+    filtered = client.get(
+        "/issues/",
+        params=[("state", "open"), ("label", "automated")],
+        headers=_auth(),
+    )
+    assert filtered.status_code == 200
+    assert len(filtered.json()) == 1
+
+    claimed = client.patch(
+        f"/issues/{issue_id}",
+        json={"state": "in_progress", "assignee": "agent-harness"},
+        headers=_auth(),
+    )
+    assert claimed.status_code == 200, claimed.text
+    assert claimed.json()["state"] == "in_progress"
+    assert claimed.json()["assignee"] == "agent-harness"
+
+    open_after = client.get("/issues/", params={"state": "open"}, headers=_auth())
+    assert open_after.json() == []
+
+    closed = client.patch(
+        f"/issues/{issue_id}",
+        json={"state": "closed"},
+        headers=_auth(),
+    )
+    assert closed.status_code == 200
+    assert closed.json()["state"] == "closed"
+
+
+def test_get_issue_not_found(client):
+    resp = client.get("/issues/does-not-exist", headers=_auth())
+    assert resp.status_code == 404