Advance dependency roots with guarded source implementations

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a10391-6355-7b41-b0fa-3e14e874f091

scripts/owner_return_handoff.py

@@ -0,0 +1,149 @@
+"""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"}
+
+
+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")
+    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
+        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, "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)]