ACTIVITY-WP-0043: owner-return pickup adapter (STATE-WP-0094-T03 owner side)

Disabled reviewed definition, pinned profile/grant, server-side unique
submission adapter with receipts, and opt-in activation plan. No schedule,
deployment or dispatch.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 671493@bnt-lap001
Assistant-Session: 60d7b11f-78f9-4731-9367-918511b2762d

src/activity_core/owner_return_pickup.py

@@ -0,0 +1,195 @@
+"""Owner-return pickup submission/receipt adapter (ACTIVITY-WP-0043).
+
+Owner side of State Hub STATE-WP-0094-T03. State Hub's offline planner
+(``state-hub/scripts/owner_return_handoff.py``) turns C-25 owner-return
+findings plus authoritative current-task eligibility into candidates keyed by
+``triggering_event_id = "owner-return:" + sha256([repo, task, message])``.
+
+This module accepts those candidates and inserts them through the existing
+ops-run queue (``create_ops_run_from_spec``). Deduplication is *server-side*:
+the ``ops_runs.idempotency_key`` unique index with ``ON CONFLICT DO NOTHING``
+decides, never a list snapshot taken earlier. Lease, retry and exact-repeat
+terminal close semantics are those of ``ops_run_queue`` unchanged.
+
+Nothing here schedules, dispatches or enables anything. Submission refuses
+unless ``ACTIVITY_CORE_OWNER_RETURN_PICKUP_ENABLED`` is set (founder-reviewed
+activation, see ``docs/owner-return-pickup-activation-plan.md``) and the caller
+asserts that message collection was paginated to completion.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+import uuid
+from dataclasses import dataclass
+from typing import Any, Mapping, Sequence
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from activity_core.ops_run_queue import (
+    build_idempotency_key,
+    create_ops_run_from_spec,
+    ops_run_queue_enabled,
+)
+from activity_core.orm import OpsRun
+from activity_core.repository_grant import validate_repository_grant
+from activity_core.rules.models import TaskSpec
+
+DEFINITION_SLUG = "owner-return-pickup"
+# Same derivation as sync_activity_definitions._definition_uuid (uuid5 over the
+# file-authored slug); kept literal here so the identity is reviewable.
+ACTIVITY_DEFINITION_ID_NAMESPACE = uuid.uuid5(
+    uuid.NAMESPACE_URL, "activity-core:activity-definition"
+)
+ACTIVITY_DEFINITION_ID = str(
+    uuid.uuid5(ACTIVITY_DEFINITION_ID_NAMESPACE, DEFINITION_SLUG)
+)
+HARNESS_PROFILE_REF = "harness.agent-dev-local@1.0.0"
+REPOSITORY_GRANT: dict[str, Any] = validate_repository_grant(
+    {
+        "version": "1",
+        "allowed_paths": ["workplans"],
+        "commit_count": {"min": 1, "max": 1},
+        "publish": False,
+    }
+)
+SOURCE_TYPE = "state_hub_owner_return"
+ENABLE_ENV = "ACTIVITY_CORE_OWNER_RETURN_PICKUP_ENABLED"
+
+_REPO = re.compile(r"^[a-z0-9][a-z0-9._-]{0,127}$")
+_TASK = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
+_EVENT = re.compile(r"^owner-return:[0-9a-f]{64}$")
+
+
+class OwnerReturnError(ValueError):
+    """A candidate or submission precondition is invalid."""
+
+
+def pickup_enabled() -> bool:
+    return os.getenv(ENABLE_ENV, "").strip().lower() in {"1", "true", "yes", "on"}
+
+
+def owner_return_event_id(repo: str, task: str, message_id: str) -> str:
+    """Recompute State Hub's planner event identity (must stay byte-identical)."""
+    encoded = json.dumps([repo, task, message_id], separators=(",", ":")).encode()
+    return "owner-return:" + hashlib.sha256(encoded).hexdigest()
+
+
+@dataclass(frozen=True)
+class OwnerReturnBinding:
+    """Validated State Hub event/source binding for one owner-return message."""
+
+    target_repo: str
+    task_record_id: str
+    message_id: str
+    triggering_event_id: str
+
+    @classmethod
+    def from_candidate(cls, candidate: Mapping[str, Any]) -> OwnerReturnBinding:
+        repo = candidate.get("target_repo")
+        task = candidate.get("task_record_id")
+        if not isinstance(repo, str) or not _REPO.fullmatch(repo):
+            raise OwnerReturnError("invalid target_repo")
+        if not isinstance(task, str) or not _TASK.fullmatch(task):
+            raise OwnerReturnError("invalid task_record_id")
+        try:
+            message = str(uuid.UUID(str(candidate.get("message_id"))))
+            definition = str(uuid.UUID(str(candidate.get("activity_definition_id"))))
+        except ValueError as exc:
+            raise OwnerReturnError("invalid message or definition identity") from exc
+        if definition != ACTIVITY_DEFINITION_ID:
+            raise OwnerReturnError("candidate planned for a different definition")
+        event = candidate.get("triggering_event_id")
+        if not isinstance(event, str) or not _EVENT.fullmatch(event):
+            raise OwnerReturnError("invalid triggering_event_id")
+        if event != owner_return_event_id(repo, task, message):
+            raise OwnerReturnError("triggering_event_id does not bind repo/task/message")
+        return cls(repo, task, message, event)
+
+    def task_spec(self) -> TaskSpec:
+        return TaskSpec(
+            title=f"Owner return: review {self.task_record_id} in {self.target_repo}",
+            description=(
+                f"An owner reply (State Hub message {self.message_id}) answers the "
+                f"wait on {self.task_record_id}. Re-read the message and the current "
+                "workplan task; prepare a review only. Do not clear the wait marker, "
+                "task status, human gate or any other field without owner review."
+            ),
+            target_repo=self.target_repo,
+            priority="medium",
+            labels=["automated", "owner-return"],
+            source_type=SOURCE_TYPE,
+            source_id=f"{self.target_repo}/{self.task_record_id}",
+            triggering_event_id=self.triggering_event_id,
+            activity_definition_id=ACTIVITY_DEFINITION_ID,
+        )
+
+
+async def submit_owner_returns(
+    session: AsyncSession,
+    candidates: Sequence[Mapping[str, Any]],
+    *,
+    collection_complete: bool,
+) -> list[dict[str, Any]]:
+    """Insert reviewable candidates; return one receipt per candidate.
+
+    Receipt ``status``: ``inserted`` (new row), ``duplicate`` (unique key already
+    held; existing row reported), ``skipped`` (planner disposition not
+    ``reviewable`` or already correlated). Raises before any write when the
+    pickup is not enabled, the queue is disabled, collection was not paginated
+    to completion, or any candidate fails binding validation.
+    """
+    if not pickup_enabled():
+        raise OwnerReturnError(f"{ENABLE_ENV} is not enabled")
+    if not ops_run_queue_enabled():
+        raise OwnerReturnError("ops-run queue is disabled")
+    if collection_complete is not True:
+        raise OwnerReturnError(
+            "message collection must be paginated to completion (C-25 caps at 200)"
+        )
+    bound = [(c, OwnerReturnBinding.from_candidate(c)) for c in candidates]
+    receipts: list[dict[str, Any]] = []
+    for candidate, binding in bound:
+        spec = binding.task_spec()
+        key = build_idempotency_key(spec)
+        receipt: dict[str, Any] = {
+            "triggering_event_id": binding.triggering_event_id,
+            "idempotency_key": key,
+            "target_repo": binding.target_repo,
+            "task_record_id": binding.task_record_id,
+            "message_id": binding.message_id,
+            "ops_run_id": None,
+            "state": None,
+        }
+        if candidate.get("disposition") != "reviewable" or candidate.get("ops_run_id"):
+            receipts.append({**receipt, "status": "skipped",
+                             "disposition": candidate.get("disposition")})
+            continue
+        row_id = await create_ops_run_from_spec(
+            session,
+            spec,
+            harness_profile_ref=HARNESS_PROFILE_REF,
+            execution_refs={
+                "correlation_id": binding.triggering_event_id,
+                "assignment_ref": binding.task_record_id,
+            },
+            repository_grant=REPOSITORY_GRANT,
+        )
+        if row_id is not None:
+            receipts.append({**receipt, "status": "inserted",
+                             "ops_run_id": str(row_id), "state": "open"})
+            continue
+        existing = (
+            await session.execute(
+                select(OpsRun.id, OpsRun.state).where(OpsRun.idempotency_key == key)
+            )
+        ).first()
+        if existing is None:
+            raise OwnerReturnError(f"insert neither created nor found {key}")
+        receipts.append({**receipt, "status": "duplicate",
+                         "ops_run_id": str(existing[0]), "state": existing[1]})
+    return receipts