state-hub/scripts/owner_return_handoff.py

"""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)]