diff --git a/docs/owner-return-pickup-activation-plan.md b/docs/owner-return-pickup-activation-plan.md index 0cea396..a61851d 100644 --- a/docs/owner-return-pickup-activation-plan.md +++ b/docs/owner-return-pickup-activation-plan.md @@ -13,6 +13,8 @@ STATE-WP-0094-T03. Nothing below has been applied. | Repository grant | v1, `allowed_paths: [workplans]`, commits 1..1, `publish: false` | | Adapter | `activity_core.owner_return_pickup.submit_owner_returns(session, candidates, collection_complete=...)` | | Gate | env `ACTIVITY_CORE_OWNER_RETURN_PICKUP_ENABLED` (unset = refuse before any write) | +| Caller | `activity_core.owner_return_runner.run_owner_return_pickup`, bounded operation `owner_return_pickup` (source `state-hub-owner-return` / `pickup`), ACTIVITY-WP-0043-T06 | +| Planner | State Hub `scripts/owner_return_handoff.py`, vendored byte-identically from state-hub `150cb0b` as `activity_core/vendored/state_hub_owner_return_handoff.py` (drift test when state-hub is checked out) | ## Preconditions (all required before activation) @@ -25,8 +27,77 @@ STATE-WP-0094-T03. Nothing below has been applied. (ACTIVITY-WP-0043-T05); the C-25 caller must pass the scanned commit SHA. 3. The caller runs inside the existing supported runner (Custodian dependency monitor / established Activity Core worker); no new schedule is introduced. + **Built at source level (ACTIVITY-WP-0043-T06)**, see "Caller wiring" below; + not deployed. 4. Founder review of this plan and of the definition/grant above. +## Caller wiring (precondition 3) + +The caller is the code-registered bounded operation `owner_return_pickup` +executed by the existing `RunActivityWorkflow` → `execute_bounded_operation` +activity on `orchestrator-tq` (single Temporal attempt, 300 s). It is exposed +like every other operator-run bounded operation: the context source sits on a +definition and the operator fires it with +`POST /ops/automations/{id}/trigger` + `confirm_side_effect: true` (the +`owner_return_pickup` side-effect marker forces the confirmation). Manual +trigger does not require `enabled: true`; the definition stays disabled, so +there is no schedule. + +Context-source stanza (exactly one repository; params validated at parse): + +```yaml +context_sources: + - type: state-hub-owner-return + query: pickup + operation: owner_return_pickup + required: true + bind_to: context.owner_return + params: + target_repo: + report_file: activity-core://owner-return/-c25.json # worker-mounted + report_sha256: + submit: false # true only under the activation flag + # optional: expected_source_revision: <40-hex> + # optional: predecessor_repos: {"STATE-WP-": state-hub} +``` + +How the C-25 report and SHA are obtained: **operator-supplied artefact.** The +checker reads workplans from the repository's registered `local_path` and gets +`source_revision` from `git` in that tree; the worker has no checkouts +(ADR-002), no checker image, and Repo Manager's C-rule engine has no C-25. So +the operator runs `consistency_check.py --json` for the repository (clean +`workplans/` so `source_revision` is a full SHA), places the JSON at the mounted +path (a `--all` list is accepted; exactly one object must match `target_repo`), and pins its sha256 in the reviewed params. The caller then refuses +unless: the digest matches; `repo_slug` equals `target_repo`; +`source_revision` (report and every C-25 finding) is the same full 40-hex SHA +(`null` refuses with `source_revision_reason`); `result != error`; no C-25 +"reply collection ... is incomplete" finding exists; and `build_current_tasks` +confirms via Forge that the SHA is still the default-branch HEAD. Planner +candidates must echo the same repo/revision. Only `reviewable` candidates reach +`submit_owner_returns`; existing ops runs are not fed to the planner, so +deduplication stays server-side and a re-run reports `duplicate` receipts. + +Evidence (bound to `context.owner_return`, IDs/counts only): planner input +(report result, revision, finding/message counts, task record IDs, +counterparts, completeness, `current_tasks` schema/revision/count, HEAD +verified), planner output (dispositions, stop reasons, candidate IDs) and +receipts (`triggering_event_id`, `idempotency_key`, `status`, `ops_run_id`, +`state`). Tests: `tests/test_owner_return_runner.py`. + +Remaining gaps before the first trial: + +- No in-cluster report producer: the artefact is produced on the workstation + and mounted by the operator. A Forge-content C-25 (Repo Manager C-rule engine + or a checker mode reading `ref=`) would remove that step. +- Completeness is inferred from the absence of the C-25 "incomplete" WARN + text; the checker JSON has no structured per-counterpart reply count / last + page field, which the evidence section below asks for. +- Forge reads are unauthenticated (public repositories only); a private target + needs a reviewed read-credential route. +- The definition does not yet carry the stanza: adding it (with the chosen + repository and artefact digest) and mounting the artefact are part of the + founder-reviewed activation, as is the flag below. + ## Exactly what would be enabled 1. Set `ACTIVITY_CORE_OWNER_RETURN_PICKUP_ENABLED=true` on the single process diff --git a/src/activity_core/activities.py b/src/activity_core/activities.py index 536ce43..5e07ef7 100644 --- a/src/activity_core/activities.py +++ b/src/activity_core/activities.py @@ -258,6 +258,13 @@ async def execute_bounded_operation(payload: dict[str, Any]) -> dict[str, Any]: from activity_core.context_resolvers.repo_manager import repo_manager_freshness result = repo_manager_freshness(params) + elif spec.operation_id == "owner_return_pickup": + from activity_core.owner_return_runner import run_owner_return_pickup + + result = await run_owner_return_pickup( + params, + session_factory=_get_session_factory() if params.get("submit") is True else None, + ) else: raise ApplicationError( f"bounded operation {spec.operation_id!r} uses a dedicated activity", diff --git a/src/activity_core/bounded_operations.py b/src/activity_core/bounded_operations.py index 5e1c974..960f204 100644 --- a/src/activity_core/bounded_operations.py +++ b/src/activity_core/bounded_operations.py @@ -73,6 +73,21 @@ _SPECS = ( credential_route="repo-manager-freshness-runner", evidence_mode="repo-manager-freshness-receipt", ), + # ACTIVITY-WP-0043-T06: single-repo owner-return pickup (C-25 artefact -> + # Forge-bound eligibility -> vendored State Hub planner -> adapter). + # Operator-triggered only; submission additionally gated by + # ACTIVITY_CORE_OWNER_RETURN_PICKUP_ENABLED. + BoundedOperationSpec( + operation_id="owner_return_pickup", + source_type="state-hub-owner-return", + query="pickup", + resolve_before_execute=False, + max_timeout_seconds=300, + temporal_max_attempts=1, + idempotency="single Temporal attempt; ops_runs.idempotency_key unique (ON CONFLICT DO NOTHING)", + credential_route="none (public Forge reads; State Hub decision reads)", + evidence_mode="owner-return-pickup-receipts", + ), ) BOUNDED_OPERATION_REGISTRY = {spec.operation_id: spec for spec in _SPECS} @@ -177,6 +192,12 @@ def normalize_operation_result(operation_id: str, raw: Any) -> dict[str, Any]: "outcomes": outcomes, "exit_code": exit_code, } + if operation_id == "owner_return_pickup": + # The runner already projects to counts/identities; keep only its keys. + keys = ("kind", "mode", "target_repo", "source_revision", "report_sha256", + "observed_on", "planner_input", "planner_output", "receipts", + "receipt_counts", "submitted_count") + return {key: result.get(key) for key in keys} | {"kind": operation_id} raise ValueError(f"no result projection for bounded operation {operation_id!r}") @@ -235,6 +256,10 @@ def validate_bounded_operations( _validate_backup(params) elif spec.operation_id == "repo_manager_freshness": _validate_repo_manager_freshness(params) + elif spec.operation_id == "owner_return_pickup": + from activity_core.owner_return_runner import validate_params + + validate_params(params) def _validate_timeout(spec: BoundedOperationSpec, params: dict[str, Any]) -> None: diff --git a/src/activity_core/ops_console.py b/src/activity_core/ops_console.py index f42a08f..8f34aba 100644 --- a/src/activity_core/ops_console.py +++ b/src/activity_core/ops_console.py @@ -41,6 +41,7 @@ SIDE_EFFECT_MARKERS = ( "forgejo_package_prune", "cnpg_option_a_backup", "repo_manager_freshness", + "owner_return_pickup", "apply: true", '"apply": true', "'apply': true", diff --git a/src/activity_core/owner_return_eligibility.py b/src/activity_core/owner_return_eligibility.py index 704299e..b8c033f 100644 --- a/src/activity_core/owner_return_eligibility.py +++ b/src/activity_core/owner_return_eligibility.py @@ -328,6 +328,16 @@ def build_current_tasks( if report is not None: if report.get("repo_slug") != target_repo: raise EligibilityError("C-25 report is for a different repository") + # State Hub 150cb0b: every checker report carries source_revision (a + # full SHA, or null + source_revision_reason). A report without the + # key predates the contract and is unbound -- refuse it. + if "source_revision" not in report: + raise EligibilityError("C-25 report has no source_revision (unbound report)") + if report["source_revision"] is None: + raise EligibilityError( + "C-25 report has no source revision: " + f"{report.get('source_revision_reason') or 'reason not given'}" + ) for key in ("source_revision", "commit"): if key in report and report[key] != scan_revision: raise EligibilityError("C-25 report revision does not match scan_revision") diff --git a/src/activity_core/owner_return_runner.py b/src/activity_core/owner_return_runner.py new file mode 100644 index 0000000..1ddbab3 --- /dev/null +++ b/src/activity_core/owner_return_runner.py @@ -0,0 +1,324 @@ +"""Single-repository owner-return pickup caller (ACTIVITY-WP-0043-T06). + +Precondition 3 of ``docs/owner-return-pickup-activation-plan.md``: the C-25 → +planner → adapter chain runs inside the existing Activity Core worker as the +code-registered bounded operation ``owner_return_pickup`` (ACT-ADR-007), +operator-triggered only (``POST /ops/automations/{id}/trigger`` with +``confirm_side_effect``). No schedule exists; the definition stays disabled. + +Report acquisition (why an operator-supplied artefact) +------------------------------------------------------ +State Hub's C-25 checker (``scripts/consistency_check.py``) reads workplan files +from the repository's registered ``local_path`` and derives ``source_revision`` +from ``git`` in that working tree. The worker has no such checkouts (ADR-002) +and no checker image, and Repo Manager's C-rule engine does not implement C-25 +yet. So the worker cannot produce the report itself. It accepts a checker JSON +artefact supplied by the operator at a worker-mounted path, pinned by sha256 in +the reviewed definition params, and binds it to its revision: + +1. ``report.repo_slug`` must equal ``target_repo``; ``report.source_revision`` + must be a full 40-hex SHA (``null`` + ``source_revision_reason`` refuses); + every C-25 finding must carry the same ``repo_slug``/``source_revision``. +2. ``build_current_tasks`` re-reads the workplans from immutable Forge content + at that SHA and refuses unless it is still the default-branch HEAD, so a + stale or forged revision cannot be planned from. +3. Collection must be complete: a failed checker run or any C-25 + "reply collection ... is incomplete" finding refuses. + +The State Hub planner is vendored byte-identically +(``activity_core.vendored.state_hub_owner_return_handoff``). Existing ops-run +rows are deliberately *not* passed to it: deduplication stays server-side in +``submit_owner_returns`` (unique idempotency key), so a re-run reports +``duplicate`` receipts rather than trusting a snapshot. + +Evidence carries counts and identities only (event/message/ops-run IDs, task +record IDs, dispositions) -- never message bodies or finding text. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from collections import Counter +from datetime import date, datetime, timezone +from pathlib import Path +from typing import Any, Callable, Mapping + +import httpx + +from activity_core import owner_return_pickup as orp +from activity_core.owner_return_eligibility import ( + DEFAULT_FORGE_API, + build_current_tasks, + forge_head_status_resolver, + state_hub_decision_status, +) +from activity_core.vendored import state_hub_owner_return_handoff as planner + +OPERATION_ID = "owner_return_pickup" +MAX_REPORT_BYTES = 5 * 1024 * 1024 +MAX_EVIDENCE_ROWS = 200 + +_SHA = re.compile(r"^[0-9a-f]{40}$") +_SHA256 = re.compile(r"^[0-9a-f]{64}$") +_REPO = re.compile(r"^[a-z0-9][a-z0-9._-]{0,127}$") +_PREFIX = re.compile(r"^[A-Z][A-Z0-9]*-(?:WP-)?$") +_INCOMPLETE = re.compile(r"reply collection from \S+ is incomplete") + + +class OwnerReturnPickupRefused(ValueError): + """A precondition failed; nothing was submitted.""" + + +def validate_params(params: Mapping[str, Any]) -> None: + """Admission checks shared by definition parsing and execution.""" + repo = params.get("target_repo") + if not isinstance(repo, str) or not _REPO.fullmatch(repo): + raise ValueError("bounded operation owner_return_pickup requires one target_repo") + if not isinstance(params.get("report_file"), str) or not params["report_file"].strip(): + raise ValueError("bounded operation owner_return_pickup requires report_file") + digest = params.get("report_sha256") + if not isinstance(digest, str) or not _SHA256.fullmatch(digest): + raise ValueError("bounded operation owner_return_pickup requires report_sha256 (64 hex)") + if not isinstance(params.get("submit"), bool): + raise ValueError("bounded operation owner_return_pickup must declare boolean submit explicitly") + expected = params.get("expected_source_revision") + if expected is not None and (not isinstance(expected, str) or not _SHA.fullmatch(expected)): + raise ValueError("expected_source_revision must be a full 40-hex SHA") + mapping = params.get("predecessor_repos", {}) + if not isinstance(mapping, dict) or any( + not isinstance(k, str) or not _PREFIX.fullmatch(k) + or not isinstance(v, str) or not _REPO.fullmatch(v) + for k, v in mapping.items() + ): + raise ValueError("predecessor_repos must map ID prefixes (e.g. 'STATE-WP-') to repo slugs") + for forbidden in ("target_repos", "repos", "command", "argv", "ops_runs"): + if forbidden in params: + raise ValueError(f"bounded operation owner_return_pickup must not declare {forbidden}") + + +def load_report(path: Path, expected_sha256: str) -> Any: + try: + raw = path.read_bytes() + except OSError as exc: + raise OwnerReturnPickupRefused(f"report artefact unreadable: {path}") from exc + if len(raw) > MAX_REPORT_BYTES: + raise OwnerReturnPickupRefused("report artefact exceeds size bound") + if hashlib.sha256(raw).hexdigest() != expected_sha256: + raise OwnerReturnPickupRefused("report artefact does not match pinned report_sha256") + try: + report = json.loads(raw) + except json.JSONDecodeError as exc: + raise OwnerReturnPickupRefused("report artefact is not JSON") from exc + if not isinstance(report, (dict, list)): + raise OwnerReturnPickupRefused("report artefact must be a checker JSON object or list") + return report + + +def select_repo_report(artefact: Any, target_repo: str) -> dict[str, Any]: + """Pick the single repository object (``--repo`` object or ``--all`` list).""" + if isinstance(artefact, dict): + return artefact + matches = [r for r in artefact if isinstance(r, dict) and r.get("repo_slug") == target_repo] + if len(matches) != 1: + raise OwnerReturnPickupRefused( + f"report artefact holds {len(matches)} objects for {target_repo}; exactly one required" + ) + return matches[0] + + +def bind_report( + report: Mapping[str, Any], target_repo: str, expected_revision: str | None = None +) -> dict[str, Any]: + """Return the planner-input summary, refusing unbound or incomplete reports.""" + if report.get("repo_slug") != target_repo: + raise OwnerReturnPickupRefused("report is for a different repository") + revision = report.get("source_revision") + if not isinstance(revision, str) or not _SHA.fullmatch(revision): + reason = report.get("source_revision_reason") + raise OwnerReturnPickupRefused( + "report source_revision is not a full 40-hex SHA" + + (f" (reason: {reason})" if isinstance(reason, str) else "") + ) + if expected_revision is not None and revision != expected_revision: + raise OwnerReturnPickupRefused("report source_revision differs from expected_source_revision") + if report.get("result") == "error": + raise OwnerReturnPickupRefused("checker collection failed (result=error)") + issues = report.get("issues") + if not isinstance(issues, list): + raise OwnerReturnPickupRefused("report issues must be a list") + findings = 0 + owner_returns: list[Mapping[str, Any]] = [] + incomplete: list[str] = [] + for issue in issues: + if not isinstance(issue, Mapping) or issue.get("check_id") != "C-25": + continue + findings += 1 + if issue.get("repo_slug") != target_repo or issue.get("source_revision") != revision: + raise OwnerReturnPickupRefused("C-25 finding is not bound to the report repository/revision") + if "owner_return" in issue: + owner_returns.append(issue) + elif _INCOMPLETE.search(str(issue.get("message") or "")): + incomplete.append(str(issue.get("file_value") or "")) + if incomplete: + raise OwnerReturnPickupRefused( + f"owner-return reply collection incomplete for {len(incomplete)} counterpart(s)" + ) + tasks: list[str] = [] + messages = 0 + counterparts: set[str] = set() + for issue in owner_returns: + context = issue["owner_return"] + if isinstance(context, Mapping): + task = context.get("task_record_id") + if isinstance(task, str) and task not in tasks: + tasks.append(task) + ids = context.get("message_ids") + messages += len(ids) if isinstance(ids, list) else 0 + marker = str(issue.get("file_value") or "") + if marker.startswith("message-from:"): + counterparts.add(marker.removeprefix("message-from:")) + return { + "report_result": report.get("result"), + "source_revision": revision, + "c25_finding_count": findings, + "owner_return_finding_count": len(owner_returns), + "task_record_ids": tasks, + "message_id_count": messages, + "counterparts": sorted(counterparts), + "collection_complete": True, + "incomplete_counterpart_count": 0, + } + + +def _candidate_evidence(candidates: list[dict[str, Any]]) -> dict[str, Any]: + return { + "candidate_count": len(candidates), + "dispositions": dict(Counter(c["disposition"] for c in candidates)), + "stop_reasons": dict(Counter(c["stop_reason"] for c in candidates if c["stop_reason"])), + "candidates": [ + {k: c.get(k) for k in ("triggering_event_id", "task_record_id", "message_id", + "disposition", "stop_reason")} + for c in candidates[:MAX_EVIDENCE_ROWS] + ], + } + + +def _receipt_evidence(receipts: list[dict[str, Any]]) -> list[dict[str, Any]]: + keys = ("triggering_event_id", "idempotency_key", "task_record_id", "message_id", + "status", "ops_run_id", "state") + return [{k: r.get(k) for k in keys} for r in receipts[:MAX_EVIDENCE_ROWS]] + + +async def run_owner_return_pickup( + params: Mapping[str, Any], + *, + session_factory: Callable[[], Any] | None = None, + client: httpx.Client | None = None, + state_hub_url: str | None = None, + forge_api: str = DEFAULT_FORGE_API, + today: date | None = None, + resolve_path: Callable[[str], Path] | None = None, +) -> dict[str, Any]: + """Plan (and, when ``submit`` and the gate allow, submit) one repository.""" + validate_params(params) + target_repo: str = params["target_repo"] + submit: bool = params["submit"] + # Fail fast before any read when submission is requested but not gated on. + if submit and not orp.pickup_enabled(): + raise OwnerReturnPickupRefused(f"{orp.ENABLE_ENV} is not enabled") + if submit and session_factory is None: + raise OwnerReturnPickupRefused("submission requires a database session factory") + + if resolve_path is None: + from activity_core.runtime_paths import resolve_runtime_path as resolve_path + report = select_repo_report( + load_report(resolve_path(params["report_file"]), params["report_sha256"]), target_repo + ) + planner_input = bind_report(report, target_repo, params.get("expected_source_revision")) + revision = planner_input["source_revision"] + observed_on = today or datetime.now(tz=timezone.utc).date() + + evidence: dict[str, Any] = { + "kind": OPERATION_ID, + "mode": "submit" if submit else "plan-only", + "target_repo": target_repo, + "source_revision": revision, + "report_sha256": params["report_sha256"], + "observed_on": observed_on.isoformat(), + "planner_input": planner_input, + "planner_output": _candidate_evidence([]), + "receipts": [], + "receipt_counts": {"inserted": 0, "duplicate": 0, "skipped": 0}, + "submitted_count": 0, + } + if not planner_input["task_record_ids"]: + return evidence + + own_client = client is None + http = client or httpx.Client(timeout=30.0) + try: + hub = (state_hub_url or _state_hub_url()).rstrip("/") + mapping = params.get("predecessor_repos") or {} + built = build_current_tasks( + target_repo=target_repo, + scan_revision=revision, + task_record_ids=planner_input["task_record_ids"], + client=http, + decision_status=state_hub_decision_status(http, hub), + external_status=forge_head_status_resolver(http, mapping, forge_api=forge_api) + if mapping else None, + require_current_head=True, + report=report, + forge_api=forge_api, + ) + except (ValueError, httpx.HTTPError) as exc: + raise OwnerReturnPickupRefused(f"eligibility refused: {exc}") from exc + finally: + if own_client: + http.close() + if built.get("source_revision") != revision: + raise OwnerReturnPickupRefused("current_tasks are bound to a different revision") + planner_input["current_tasks_schema"] = built.get("schema") + planner_input["current_tasks_source_revision"] = built["source_revision"] + planner_input["current_tasks_count"] = len(built["current_tasks"]) + planner_input["head_verified"] = True + + try: + candidates = planner.plan_owner_returns( + report, + activity_definition_id=orp.ACTIVITY_DEFINITION_ID, + current_tasks=built["current_tasks"], + observed_on=observed_on, + ) + except ValueError as exc: + raise OwnerReturnPickupRefused(f"planner refused: {exc}") from exc + for candidate in candidates: + if candidate.get("target_repo") != target_repo or candidate.get( + "source_revision", revision + ) != revision: + raise OwnerReturnPickupRefused("planner candidate is bound to a different repository/revision") + evidence["planner_output"] = _candidate_evidence(candidates) + + reviewable = [c for c in candidates if c["disposition"] == "reviewable"] + if not submit or not reviewable: + return evidence + async with session_factory() as session: + async with session.begin(): + receipts = await orp.submit_owner_returns( + session, reviewable, collection_complete=planner_input["collection_complete"] + ) + evidence["submitted_count"] = len(reviewable) + evidence["receipts"] = _receipt_evidence(receipts) + counts = Counter(r["status"] for r in receipts) + evidence["receipt_counts"] = {k: counts.get(k, 0) for k in ("inserted", "duplicate", "skipped")} + return evidence + + +def _state_hub_url() -> str: + import os + + from activity_core.automation_status import DEFAULT_STATE_HUB_URL + + return os.environ.get("STATE_HUB_URL", DEFAULT_STATE_HUB_URL) diff --git a/src/activity_core/vendored/__init__.py b/src/activity_core/vendored/__init__.py new file mode 100644 index 0000000..c0e550f --- /dev/null +++ b/src/activity_core/vendored/__init__.py @@ -0,0 +1,5 @@ +"""Verbatim copies of small upstream modules the worker cannot import at runtime. + +Each file is a byte-identical copy; drift is detected by tests when the +upstream checkout is present. Never edit a vendored file locally -- re-copy it. +""" diff --git a/src/activity_core/vendored/state_hub_owner_return_handoff.py b/src/activity_core/vendored/state_hub_owner_return_handoff.py new file mode 100644 index 0000000..a610798 --- /dev/null +++ b/src/activity_core/vendored/state_hub_owner_return_handoff.py @@ -0,0 +1,159 @@ +"""Offline C-25 → existing Activity Core ops-run handoff planning. + +No network, scheduling, dispatch or task mutation. An admitted ActivityDefinition +and owner harness are prerequisites for a caller to use these candidate IDs. +""" +from __future__ import annotations + +import hashlib +import json +import re +import uuid +from collections.abc import Mapping, Sequence +from datetime import date + +_REPO = re.compile(r"^[a-z0-9][a-z0-9-]*$") +_TASK = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") +_STATES = {"open", "claimed", "succeeded", "failed", "expired"} +_SHA = re.compile(r"^[0-9a-f]{40}$") + + +def _uuid(value: object) -> str: + try: + return str(uuid.UUID(str(value))) + except (ValueError, TypeError, AttributeError): + raise ValueError("invalid ops-run/message UUID") from None + + +def _eligibility_stop(current: Mapping, *, repo: str, task: str, + marker: object, observed_on: date) -> str | None: + required = {"target_repo", "task_record_id", "status", "workplan_status", + "workplan_flavor", "needs_human", "blocked_on", "depends_on", + "predecessor_statuses", "wait_until", "decision_id", "decision_status"} + if not isinstance(current, Mapping) or not required <= current.keys(): + raise ValueError("missing authoritative task eligibility fields") + if current["target_repo"] != repo or current["task_record_id"] != task: + return "stale_task_identity" + if type(current["needs_human"]) is not bool: + raise ValueError("needs_human must be an authoritative boolean") + if current["needs_human"]: + return "human_gate" + if current["decision_id"] is not None and current["decision_status"] != "resolved": + return "decision_not_resolved_or_unknown" + if current["workplan_flavor"] == "residual": + return "unpromoted_residual" + if current["workplan_status"] not in {"proposed", "ready", "active", "blocked", "backlog"}: + return "workplan_not_open" + if current["status"] != "wait": + return "task_not_waiting" + if not isinstance(marker, str) or not re.fullmatch(r"message-from:[a-z0-9][a-z0-9-]*", marker): + raise ValueError("missing owner-return marker in finding") + if current["blocked_on"] != marker: + return "stale_owner_return_marker" + dependencies = current["depends_on"] + statuses = current["predecessor_statuses"] + if not isinstance(dependencies, list) or not isinstance(statuses, Mapping): + raise ValueError("invalid authoritative predecessor eligibility") + if any(not isinstance(item, str) or not item for item in dependencies): + raise ValueError("invalid predecessor identity") + if any(statuses.get(item) not in {"done", "cancel", "finished", "archived"} + for item in dependencies): + return "predecessor_not_terminal_or_unknown" + wait_until = current["wait_until"] + if wait_until is not None: + if type(wait_until) is date: + until = wait_until + elif isinstance(wait_until, str) and re.fullmatch(r"\d{4}-\d{2}-\d{2}", wait_until): + until = date.fromisoformat(wait_until) + else: + raise ValueError("invalid authoritative wait_until date") + if until > observed_on: + return "future_wait_window" + return None + + +def plan_owner_returns( + report: Mapping, *, activity_definition_id: str, current_tasks: Mapping, + observed_on: date, ops_runs: Sequence[Mapping] = () +) -> list[dict]: + """Deduplicate by repository, task and message against correlated ops runs. + + Existing open/claimed rows are pending, succeeded rows are completed, and + failed/expired rows require owner review. None is dispatched again here. + A succeeded automation receipt does not itself close a workplan task. + Callers must build current_tasks from fresh authoritative source, resolving + every predecessor; checker findings alone contain insufficient safety data. + """ + if not isinstance(current_tasks, Mapping) or type(observed_on) is not date: + raise ValueError("authoritative eligibility and observation date required") + definition = _uuid(activity_definition_id) + repo = report.get("repo_slug") + if not isinstance(repo, str) or not _REPO.fullmatch(repo): + raise ValueError("invalid repository identity") + if report.get("result") == "error": + raise ValueError("cannot plan from failed checker collection") + # Passed through unchanged so candidates stay bound to the scanned commit; + # None (unknown/dirty scan) travels as None and the eligibility builder + # refuses it. Never substituted or guessed here. + revision = report.get("source_revision") + if revision is not None and (not isinstance(revision, str) or not _SHA.fullmatch(revision)): + raise ValueError("source_revision must be a full 40-hex commit SHA or null") + issues = report.get("issues") + if not isinstance(issues, list): + raise ValueError("checker issues must be a list") + candidates = {} + for issue in issues: + if not isinstance(issue, Mapping): + raise ValueError("invalid checker issue") + if issue.get("check_id") != "C-25" or "owner_return" not in issue: + continue + if issue.get("repo_slug", repo) != repo or issue.get("source_revision", revision) != revision: + raise ValueError("C-25 finding is bound to a different repository or revision") + context = issue["owner_return"] + if not isinstance(context, Mapping): + raise ValueError("invalid owner-return context") + task = context.get("task_record_id") + messages = context.get("message_ids") + if not isinstance(task, str) or not _TASK.fullmatch(task): + raise ValueError("missing task identity") + if not isinstance(messages, list) or not messages: + raise ValueError("missing message identities") + if task not in current_tasks: + raise ValueError("missing authoritative task eligibility") + stop = _eligibility_stop(current_tasks[task], repo=repo, task=task, + marker=issue.get("file_value"), observed_on=observed_on) + for raw_message in messages: + message = _uuid(raw_message) + encoded = json.dumps([repo, task, message], separators=(",", ":")).encode() + event = "owner-return:" + hashlib.sha256(encoded).hexdigest() + candidates[event] = { + "target_repo": repo, "source_revision": revision, + "task_record_id": task, "message_id": message, + "activity_definition_id": definition, "triggering_event_id": event, + "disposition": "stopped" if stop else "reviewable", + "stop_reason": stop, "ops_run_id": None, + } + matched = set() + for row in ops_runs: + if not isinstance(row, Mapping): + raise ValueError("invalid ops-run receipt") + event = row.get("triggering_event_id") + if event not in candidates or row.get("target_repo") != repo: + continue + if _uuid(row.get("activity_definition_id")) != definition: + continue + if event in matched: + raise ValueError("multiple correlated ops runs require owner reconciliation") + matched.add(event) + state = row.get("state") + if state not in _STATES: + raise ValueError("invalid correlated ops-run state") + candidate = candidates[event] + candidate["ops_run_id"] = _uuid(row.get("id")) + if candidate["disposition"] == "stopped": + continue + candidate["disposition"] = { + "open": "pending", "claimed": "pending", "succeeded": "completed", + "failed": "review_required", "expired": "review_required", + }[state] + return [candidates[key] for key in sorted(candidates)] diff --git a/tests/test_bounded_operations.py b/tests/test_bounded_operations.py index b09ce87..31b4b0f 100644 --- a/tests/test_bounded_operations.py +++ b/tests/test_bounded_operations.py @@ -21,6 +21,8 @@ def test_registry_names_only_the_accepted_operations() -> None: "cnpg_option_a_backup", # ACTIVITY-WP-0044 / RMGR-WP-0020-T03 (draft definition, disabled) "repo_manager_freshness", + # ACTIVITY-WP-0043-T06 (operator-triggered single-repo pickup) + "owner_return_pickup", } diff --git a/tests/test_owner_return_eligibility.py b/tests/test_owner_return_eligibility.py index ed875e8..7a4235a 100644 --- a/tests/test_owner_return_eligibility.py +++ b/tests/test_owner_return_eligibility.py @@ -196,6 +196,18 @@ def test_refuses_unknown_commit(): _build(transport=_transport(_files(), sha=OLD, head=SHA)) +def test_refuses_report_without_source_revision_key(): + with pytest.raises(ore.EligibilityError, match="unbound report"): + _build(report={"repo_slug": REPO, "issues": []}) + + +def test_refuses_null_source_revision_with_reason(): + report = {"repo_slug": REPO, "source_revision": None, + "source_revision_reason": "dirty_workplans", "issues": []} + with pytest.raises(ore.EligibilityError, match="dirty_workplans"): + _build(report=report) + + def test_refuses_report_revision_mismatch(): report = {"repo_slug": REPO, "source_revision": OLD, "issues": []} with pytest.raises(ore.EligibilityError, match="revision"): @@ -288,7 +300,7 @@ def test_output_is_accepted_by_state_hub_planner(): planner = importlib.util.module_from_spec(spec) spec.loader.exec_module(planner) msg = "11111111-2222-3333-4444-555555555555" - report = {"repo_slug": REPO, "result": "warn", "issues": [{ + report = {"repo_slug": REPO, "source_revision": SHA, "result": "warn", "issues": [{ "check_id": "C-25", "file_value": "message-from:state-hub", "owner_return": {"task_record_id": "WARDEN-WP-0041-T03", "message_ids": [msg]}, }]} diff --git a/tests/test_owner_return_runner.py b/tests/test_owner_return_runner.py new file mode 100644 index 0000000..0cb9145 --- /dev/null +++ b/tests/test_owner_return_runner.py @@ -0,0 +1,354 @@ +"""Tests for the single-repo owner-return pickup caller (ACTIVITY-WP-0043-T06). + +Forge/State Hub HTTP is served by ``httpx.MockTransport``; the ops-run insert +is replaced by an in-memory unique-key store standing in for the DB. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import uuid +from contextlib import asynccontextmanager +from datetime import date +from pathlib import Path +from unittest.mock import MagicMock +from urllib.parse import unquote + +import httpx +import pytest + +from activity_core import owner_return_pickup as orp +from activity_core import owner_return_runner as runner +from activity_core.activities import execute_bounded_operation +from activity_core.bounded_operations import ( + BOUNDED_OPERATION_REGISTRY, + normalize_operation_result, + validate_bounded_operations, +) +from activity_core.owner_return_eligibility import DEFAULT_FORGE_API + +REPO = "ops-warden" +SHA = "a" * 40 +TASK = "WARDEN-WP-0041-T03" +MSG1 = "11111111-2222-3333-4444-555555555555" +MSG2 = "66666666-7777-8888-9999-000000000000" +BASE = f"{DEFAULT_FORGE_API}/repos/coulomb/{REPO}" +WP = f"""--- +id: WARDEN-WP-0041 +status: active +--- + +```task +id: {TASK} +status: wait +needs_human: false +blocked_on: "message-from:state-hub" +``` + +```task +id: WARDEN-WP-0041-T04 +status: wait +needs_human: true +blocked_on: "message-from:state-hub" +``` +""" +FILES = {"workplans/WARDEN-WP-0041-owner.md": WP} +VENDORED = Path(runner.__file__).parent / "vendored" / "state_hub_owner_return_handoff.py" +UPSTREAM = Path("/home/worsch/state-hub/scripts/owner_return_handoff.py") + + +def _transport(head=SHA, seen=None): + def handler(request: httpx.Request) -> httpx.Response: + path = unquote(str(request.url).split("?")[0]) + ref = request.url.params.get("ref") + if seen is not None: + seen.append(path) + if path == BASE: + return httpx.Response(200, json={"default_branch": "main"}) + if path == f"{BASE}/branches/main": + return httpx.Response(200, json={"commit": {"id": head}}) + if path.startswith(f"{BASE}/git/commits/"): + return httpx.Response(200, json={"sha": path.rsplit("/", 1)[1]}) + if path == f"{BASE}/contents/workplans" and ref == SHA: + return httpx.Response(200, json=[ + {"type": "file", "name": p.rsplit("/", 1)[1], "path": p} for p in FILES]) + if path.startswith(f"{BASE}/raw/") and ref == SHA: + name = path[len(f"{BASE}/raw/"):] + if name in FILES: + return httpx.Response(200, text=FILES[name]) + return httpx.Response(404) + + return httpx.MockTransport(handler) + + +def _issue(task, messages, **over): + issue = { + "severity": "WARN", "check_id": "C-25", "message": "Owner-return task ...", + "file_value": "message-from:state-hub", "repo_slug": REPO, + "source_revision": SHA, "source_revision_reason": None, + "owner_return": {"task_record_id": task, "message_ids": messages}, + } + issue.update(over) + return issue + + +def _report(**over): + report = { + "repo_slug": REPO, "repo_path": "/x", "source_revision": SHA, + "source_revision_reason": None, "result": "warn", + "issues": [_issue(TASK, [MSG1, MSG2]), _issue("WARDEN-WP-0041-T04", [MSG1]), + {"severity": "WARN", "check_id": "C-04", "message": "other"}], + } + report.update(over) + return report + + +def _write(tmp_path, report): + raw = json.dumps(report).encode() + path = tmp_path / "c25.json" + path.write_bytes(raw) + return {"target_repo": REPO, "report_file": str(path), + "report_sha256": hashlib.sha256(raw).hexdigest(), "submit": True} + + +class _Store: + """Stand-in for the ops_runs unique index (ON CONFLICT DO NOTHING).""" + + def __init__(self): + self.rows: dict[str, uuid.UUID] = {} + self.last_conflict: str | None = None + + async def create(self, session, spec, **_kw): + from activity_core.ops_run_queue import build_idempotency_key + + key = build_idempotency_key(spec) + if key in self.rows: + self.last_conflict = key + return None + self.rows[key] = uuid.uuid4() + return self.rows[key] + + def session_factory(self): + store = self + + @asynccontextmanager + async def begin(): + yield + + @asynccontextmanager + async def factory(): + session = MagicMock() + session.begin = begin + + async def execute(_stmt): + result = MagicMock() + result.first.return_value = (store.rows[store.last_conflict], "open") + return result + + session.execute = execute + yield session + + return factory + + +@pytest.fixture +def store(monkeypatch): + s = _Store() + monkeypatch.setattr(orp, "create_ops_run_from_spec", s.create) + monkeypatch.setenv(orp.ENABLE_ENV, "true") + monkeypatch.delenv("OPS_RUN_QUEUE_ENABLED", raising=False) + return s + + +def _run(params, store=None, head=SHA, seen=None): + return asyncio.run(runner.run_owner_return_pickup( + params, + session_factory=store.session_factory() if store else None, + client=httpx.Client(transport=_transport(head=head, seen=seen)), + state_hub_url="http://hub", + today=date(2026, 10, 4), + resolve_path=Path, + )) + + +def test_submits_only_reviewable_and_rerun_yields_only_duplicates(tmp_path, store): + params = _write(tmp_path, _report()) + first = _run(params, store) + out = first["planner_output"] + assert out["candidate_count"] == 3 + assert out["dispositions"] == {"reviewable": 2, "stopped": 1} + assert out["stop_reasons"] == {"human_gate": 1} + assert first["submitted_count"] == 2 + assert first["receipt_counts"] == {"inserted": 2, "duplicate": 0, "skipped": 0} + assert {r["message_id"] for r in first["receipts"]} == {MSG1, MSG2} + assert all(r["task_record_id"] == TASK for r in first["receipts"]) + pin = first["planner_input"] + assert pin["source_revision"] == pin["current_tasks_source_revision"] == SHA + assert pin["collection_complete"] is True and pin["head_verified"] is True + assert pin["message_id_count"] == 3 and pin["counterparts"] == ["state-hub"] + + again = _run(params, store) + assert again["receipt_counts"] == {"inserted": 0, "duplicate": 2, "skipped": 0} + assert [r["ops_run_id"] for r in again["receipts"]] == [ + r["ops_run_id"] for r in first["receipts"]] + assert len(store.rows) == 2 + # Evidence is IDs/counts only: no finding or message text survives. + blob = json.dumps(again) + assert "Owner-return task" not in blob and "other" not in blob + + +@pytest.mark.parametrize("revision,reason", [ + (None, "dirty_workplans"), ("a" * 39, None), ("main", None), ("A" * 40, None)]) +def test_refuses_missing_or_short_sha(tmp_path, store, revision, reason): + params = _write(tmp_path, _report(source_revision=revision, source_revision_reason=reason)) + with pytest.raises(runner.OwnerReturnPickupRefused, match="full 40-hex SHA"): + _run(params, store) + assert store.rows == {} + + +def test_refuses_revision_mismatch_in_finding(tmp_path, store): + report = _report() + report["issues"][0]["source_revision"] = "b" * 40 + with pytest.raises(runner.OwnerReturnPickupRefused, match="not bound"): + _run(_write(tmp_path, report), store) + assert store.rows == {} + + +def test_refuses_finding_without_revision_binding(tmp_path, store): + report = _report() + del report["issues"][0]["source_revision"] + with pytest.raises(runner.OwnerReturnPickupRefused, match="not bound"): + _run(_write(tmp_path, report), store) + + +def test_refuses_when_scanned_revision_is_no_longer_head(tmp_path, store): + with pytest.raises(runner.OwnerReturnPickupRefused, match="default-branch head"): + _run(_write(tmp_path, _report()), store, head="c" * 40) + assert store.rows == {} + + +def test_refuses_expected_revision_mismatch(tmp_path, store): + params = _write(tmp_path, _report()) | {"expected_source_revision": "b" * 40} + with pytest.raises(runner.OwnerReturnPickupRefused, match="expected_source_revision"): + _run(params, store) + + +def test_refuses_other_repository(tmp_path, store): + with pytest.raises(runner.OwnerReturnPickupRefused, match="different repository"): + _run(_write(tmp_path, _report(repo_slug="state-hub")), store) + + +def test_refuses_incomplete_collection(tmp_path, store): + report = _report() + report["issues"].append({ + "severity": "WARN", "check_id": "C-25", "repo_slug": REPO, "source_revision": SHA, + "file_value": "message-from:state-hub", + "message": "Owner-return reply collection from state-hub is incomplete (10000 " + "message(s) read; hub ignored paging or the 50-page bound was reached)", + }) + seen: list = [] + with pytest.raises(runner.OwnerReturnPickupRefused, match="incomplete"): + _run(_write(tmp_path, report), store, seen=seen) + assert seen == [] and store.rows == {} + + +def test_refuses_failed_checker_collection(tmp_path, store): + with pytest.raises(runner.OwnerReturnPickupRefused, match="result=error"): + _run(_write(tmp_path, _report(result="error")), store) + + +def test_refuses_when_flag_unset_before_any_read(tmp_path, store, monkeypatch): + monkeypatch.delenv(orp.ENABLE_ENV) + seen: list = [] + with pytest.raises(runner.OwnerReturnPickupRefused, match=orp.ENABLE_ENV): + _run(_write(tmp_path, _report()), store, seen=seen) + assert seen == [] and store.rows == {} + + +def test_plan_only_needs_no_flag_and_writes_nothing(tmp_path, store, monkeypatch): + monkeypatch.delenv(orp.ENABLE_ENV) + params = _write(tmp_path, _report()) | {"submit": False} + out = _run(params) + assert out["mode"] == "plan-only" + assert out["planner_output"]["dispositions"] == {"reviewable": 2, "stopped": 1} + assert out["receipts"] == [] and store.rows == {} + + +def test_all_list_selects_the_single_repo_object(tmp_path, store): + other = _report(repo_slug="state-hub", issues=[]) + out = _run(_write(tmp_path, [other, _report()]) | {"submit": False}, store) + assert out["planner_output"]["candidate_count"] == 3 + with pytest.raises(runner.OwnerReturnPickupRefused, match="exactly one"): + _run(_write(tmp_path, [_report(), _report()]), store) + with pytest.raises(runner.OwnerReturnPickupRefused, match="exactly one"): + _run(_write(tmp_path, [other]), store) + + +def test_null_revision_refusal_names_reason(tmp_path, store): + params = _write(tmp_path, _report(source_revision=None, + source_revision_reason="changed_during_scan")) + with pytest.raises(runner.OwnerReturnPickupRefused, match="changed_during_scan"): + _run(params, store) + + +def test_refuses_tampered_artefact(tmp_path, store): + params = _write(tmp_path, _report()) + Path(params["report_file"]).write_text(json.dumps(_report(result="pass"))) + with pytest.raises(runner.OwnerReturnPickupRefused, match="report_sha256"): + _run(params, store) + + +def test_no_owner_return_findings_skips_forge(tmp_path, store): + seen: list = [] + out = _run(_write(tmp_path, _report(issues=[])), store, seen=seen) + assert seen == [] and out["planner_output"]["candidate_count"] == 0 + + +@pytest.mark.parametrize("bad", [ + {"target_repo": None}, {"report_sha256": "x"}, {"submit": "yes"}, + {"repos": ["a", "b"]}, {"predecessor_repos": {"bad": "x"}}, + {"expected_source_revision": "abc"}]) +def test_param_admission(tmp_path, bad): + params = _write(tmp_path, _report()) | bad + with pytest.raises(ValueError): + runner.validate_params(params) + + +def test_registered_as_operator_bounded_operation(tmp_path, store, monkeypatch): + spec = BOUNDED_OPERATION_REGISTRY["owner_return_pickup"] + assert spec.temporal_max_attempts == 1 + params = _write(tmp_path, _report()) | {"submit": False} + source = {"type": spec.source_type, "query": spec.query, + "operation": "owner_return_pickup", "params": params, + "bind_to": "context.owner_return"} + validate_bounded_operations([source], []) + with pytest.raises(ValueError): + validate_bounded_operations([source | {"params": {"submit": True}}], []) + + async def fake(params, **kw): + assert kw["session_factory"] is None + return {"kind": "owner_return_pickup", "mode": "plan-only", "secret": "x"} + + monkeypatch.setattr(runner, "run_owner_return_pickup", fake) + patch = asyncio.run(execute_bounded_operation({"source": source, "run_id": "r"})) + assert patch["owner_return"]["mode"] == "plan-only" + assert "secret" not in patch["owner_return"] + assert normalize_operation_result("owner_return_pickup", {})["kind"] == "owner_return_pickup" + + +def test_definition_stays_disabled_and_is_side_effect_marked(): + from activity_core.definition_parser import parse_file + from activity_core.ops_console import SIDE_EFFECT_MARKERS + + d = parse_file(Path(__file__).parent.parent / "activity-definitions" / "owner-return-pickup.md") + assert d.enabled is False + assert "owner_return_pickup" in SIDE_EFFECT_MARKERS + + +@pytest.mark.skipif(not UPSTREAM.is_file(), reason="state-hub planner not checked out") +def test_vendored_planner_matches_upstream(): + assert VENDORED.read_bytes() == UPSTREAM.read_bytes(), ( + "re-copy state-hub/scripts/owner_return_handoff.py into activity_core/vendored/" + ) diff --git a/workplans/ACTIVITY-WP-0043-state-hub-owner-return-pickup.md b/workplans/ACTIVITY-WP-0043-state-hub-owner-return-pickup.md index 0456d2d..f86fa74 100644 --- a/workplans/ACTIVITY-WP-0043-state-hub-owner-return-pickup.md +++ b/workplans/ACTIVITY-WP-0043-state-hub-owner-return-pickup.md @@ -85,8 +85,9 @@ blocking_reason: "Activation is a separately reviewed founder action" state_hub_task_id: "254526ce-5a76-57b3-b7d5-a874518c7b40" ``` -Plan: `docs/owner-return-pickup-activation-plan.md`. Requires T03 and T05 -(eligibility-source binding of fresh authoritative `current_tasks`). +Plan: `docs/owner-return-pickup-activation-plan.md`. Requires T03, T05 and T06 +(eligibility-source binding of fresh authoritative `current_tasks`; single-repo +caller in the existing worker). ## Eligibility input builder bound to the scan revision @@ -109,3 +110,28 @@ pinned per foreign repo); unresolved targets stay `None`. Decision status comes from an injected resolver (`state_hub_decision_status`); failure stays unknown. Tests: `tests/test_owner_return_eligibility.py` (HTTP via MockTransport, including acceptance by the State Hub planner). + +## Single-repo caller inside the existing worker (precondition 3) + +```task +id: ACTIVITY-WP-0043-T06 +status: done +priority: high +``` + +`src/activity_core/owner_return_runner.py`, registered as bounded operation +`owner_return_pickup` (`state-hub-owner-return`/`pickup`) and executed by the +existing `execute_bounded_operation` activity; operator-triggered via +`/ops/automations/{id}/trigger` with `confirm_side_effect`, no schedule, +definition still `enabled: false`. Reads an operator-supplied C-25 JSON +artefact pinned by sha256 (the checker cannot run in the worker: it needs a +local checkout), requires a full-SHA `source_revision` on the report and every +C-25 finding, refuses incomplete collection/failed checker, binds +`build_current_tasks` to the SHA (Forge HEAD verified), runs the vendored State +Hub planner, and submits only `reviewable` candidates under +`ACTIVITY_CORE_OWNER_RETURN_PICKUP_ENABLED`. Evidence is IDs/counts only. +Planner vendored from state-hub `150cb0b` (C-25 source-revision contract); +`build_current_tasks` now also refuses a report lacking `source_revision` and +names `source_revision_reason` when it is null. Gaps (in-cluster report producer, structured completeness proof, private-repo +Forge reads) are listed in the activation plan. Tests: +`tests/test_owner_return_runner.py`.