fix: exclude epic and child-only containers from allocator selection (Closes #844 )

Epics and parent issues whose body delegates implementation to children are skipped before ranking with structured reason epic_or_child_only_container. Title-only "epic" mentions without body/label evidence remain eligible. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-23 04:56:58 -04:00
6 changed files with 374 additions and 1191 deletions
@@ -53,6 +53,8 @@ OUTCOME_CANDIDATE_SET_DRIFT = "candidate_set_drift"
 SKIP_CLAIMED_BY_OTHER_SESSION = "claimed_by_other_session"
 # #776: controller-supplied pre-rank exclusion.
 SKIP_EXCLUDED_BY_CONTROLLER = "excluded_by_controller"
 # #844: epic / child-only implementation container (pre-rank).
 SKIP_EPIC_OR_CHILD_ONLY_CONTAINER = "epic_or_child_only_container"
 # Ownership verdicts for a live claim on a candidate (#765).
 OWNERSHIP_OWN = "own"
@@ -130,6 +132,39 @@ ROLE_ACTIONS: dict[str, tuple[tuple[str, ...], tuple[str, ...]]] = {
 }
 # Body phrases that prove an issue is an implementation container, not a
 # unit of direct author work (#844). Matched case-insensitively against the
 # issue body. Title alone is never sufficient (ordinary issues may mention
 # "epic" incidentally).
 _CHILD_ONLY_BODY_MARKERS: tuple[str, ...] = (
    "implementation is delivered via child issues only",
    "implementation is delivered through child issues only",
    "implementation is delivered via child issues",
    "implementation is delivered through child issues",
    "do not implement product features in this epic",
    "do not implement product features in this epic issue itself",
    "no product feature implementation is claimed complete solely on this epic",
    "implementable child issues remain independently eligible",
    "owns the product roadmap and linkage",
    "this epic owns the product roadmap",
    "coordination container",
    "child-only container",
    "implementation is delegated to child",
 )
 # Explicit epic / umbrella labels (structured evidence preferred over title).
 _EPIC_LABELS: frozenset[str] = frozenset(
    {
        "type:epic",
        "epic",
        "kind:epic",
        "scope:epic",
        "type:umbrella",
        "umbrella",
    }
 )
@dataclass
 class WorkCandidate:
    """One assignable Gitea issue or PR presented to the allocator."""
@@ -139,6 +174,7 @@ class WorkCandidate:
    state: str = "open"
    labels: tuple[str, ...] = ()
    title: str = ""
    body: str = ""
    priority: int = 0
    head_sha: str | None = None
    # Routing signals (callers derive from Gitea / review feedback).
@@ -158,6 +194,7 @@ class WorkCandidate:
        self.labels = tuple(
            str(x).strip().lower() for x in (self.labels or ()) if str(x).strip()
        )
        self.body = str(self.body or "")
        if self.kind not in WORK_KINDS:
            raise InvalidWorkKindError(
                f"candidate kind '{self.kind}' is not assignable; only "
@@ -171,6 +208,7 @@ class WorkCandidate:
            "state": self.state,
            "labels": list(self.labels),
            "title": self.title,
            "body": self.body,
            "priority": self.priority,
            "head_sha": self.head_sha,
            "request_changes_current_head": self.request_changes_current_head,
@@ -184,6 +222,51 @@ class WorkCandidate:
        }
 def classify_epic_or_child_only_container(
    c: WorkCandidate,
 ) -> tuple[bool, str | None]:
    """Return whether *c* is an epic / child-only implementation container (#844).
    Exclusion uses structured evidence first (labels, body scope language).
    A bare title containing the word "epic" is **not** enough — ordinary
    implementable issues may mention epics incidentally. A title that is
    explicitly prefixed ``Epic:`` only counts when the body also proves
    child-only / no-direct-implementation scope (or an epic label is present).
    PRs are never classified as containers here (they already have a head).
    """
    if c.kind != "issue":
        return False, None
    labels = set(c.labels)
    epic_label = sorted(labels & _EPIC_LABELS)
    body_l = (c.body or "").lower()
    title = (c.title or "").strip()
    title_l = title.lower()
    body_hits = [m for m in _CHILD_ONLY_BODY_MARKERS if m in body_l]
    title_epic_prefix = title_l.startswith("epic:") or title_l.startswith("epic ")
    if epic_label:
        detail = f"label={epic_label[0]}"
        if body_hits:
            detail = f"{detail}; body_marker={body_hits[0]!r}"
        return True, detail
    if body_hits:
        # Body proves child-only / umbrella scope. Title "Epic:" is corroborating
        # but not required — containers without the word still exclude.
        detail = f"body_marker={body_hits[0]!r}"
        if title_epic_prefix:
            detail = f"title_epic_prefix; {detail}"
        return True, detail
    # Title-only "Epic:" without body scope evidence is insufficient (#844 AC:
    # eligibility does not rely solely on the word "Epic" in a title).
    # Similarly, incidental "epic" mid-title without markers stays eligible.
    return False, None
@dataclass
 class SkipRecord:
    kind: str
@@ -850,7 +933,8 @@ def allocate_next_work(
    ownership_defects: list[dict[str, Any]] = []
    controller_excluded: list[dict[str, Any]] = []
-    # #776 AC2: remove excluded numbers *before* ranking / selection / lease.
+    # #776 AC2 + #844: remove excluded numbers *and* epic/child-only containers
    # *before* ranking / selection / lease so they never receive assignments.
    rankable: list[WorkCandidate] = []
    for c in candidates:
        if int(c.number) in exclude_set:
@@ -929,6 +1013,23 @@ def allocate_next_work(
                    },
                }
            continue
        # #844: epics / child-only containers are never direct implement targets.
        is_container, container_detail = classify_epic_or_child_only_container(c)
        if is_container:
            detail = container_detail or "epic or child-only container"
            reason = (
                f"{c.kind}#{c.number} {SKIP_EPIC_OR_CHILD_ONLY_CONTAINER}: "
                f"{detail}; implementation is delegated to child issues"
            )
            skipped.append(
                SkipRecord(
                    c.kind,
                    c.number,
                    reason,
                    SKIP_EPIC_OR_CHILD_ONLY_CONTAINER,
                )
            )
            continue
        rankable.append(c)
    ordered = sort_candidates(rankable)
@@ -1115,12 +1216,6 @@ def allocate_next_work(
            "reasons": [
                "dry-run only (apply=false); no assignment/lease created — "
                "call again with apply=true to reserve via control-plane DB"
                + (
                    "; after apply, the required-role worker consumes via "
                    "gitea_adopt_workflow_lease (#843)"
                    if mode == ALLOCATION_MODE_CROSS_ROLE and expected_role != role_norm
                    else ""
                )
            ],
            "skipped": [s.as_dict() for s in skipped],
            "terminal_pr": terminal_pr,
@@ -1152,9 +1247,6 @@ def allocate_next_work(
    # Atomic reserve via #613 substrate.
    ttl = lease_ttl_seconds if lease_ttl_seconds is not None else None
    try:
        cross_role_handoff = (
            mode == ALLOCATION_MODE_CROSS_ROLE and lease_role != role_norm
        )
        kwargs: dict[str, Any] = {
            "session_id": session_id,
            "role": lease_role,
@@ -1166,8 +1258,7 @@ def allocate_next_work(
            "expected_head_sha": selected.head_sha,
            "allowed_actions": allowed,
            "forbidden_actions": forbidden,
-            # #843: mark cross-role allocations as awaiting independent consume
+            "phase": "allocated",
            "phase": "awaiting_handoff" if cross_role_handoff else "allocated",
        }
        if ttl is not None:
            kwargs["lease_ttl_seconds"] = int(ttl)
@@ -1247,52 +1338,7 @@ def allocate_next_work(
        "lease_role": lease_role,
        "source": "control_plane_db.assign_and_lease",
    }
-    consume_allocation = None
+    return {
    if cross_role_handoff and result.lease_id:
        # Durable handoff marker so independent required-role workers can
        # consume without sharing the controller session (#843).
        handoff_prov = {
            "cross_role_handoff": True,
            "handoff_status": "pending",
            "allocating_session_id": session_id,
            "allocating_role": role_norm,
            "required_role": expected_role,
            "required_profile": selection["required_profile"],
            "required_namespace": selection["required_namespace"],
            "assignment_id": result.assignment_id,
            "lease_id": result.lease_id,
            "allocation_mode": mode,
            "adopted_by_session_id": None,
        }
        try:
            db.attach_lease_provenance(result.lease_id, handoff_prov)
        except ControlPlaneError:
            # Still return assignment evidence; consume path may be unavailable
            handoff_prov["attach_failed"] = True
        consume_allocation = {
            "tool": "gitea_adopt_workflow_lease",
            "lease_id": result.lease_id,
            "assignment_id": result.assignment_id,
            "required_role": expected_role,
            "required_profile": selection["required_profile"],
            "required_namespace": selection["required_namespace"],
            "handoff_status": "pending",
            "controller_session_required": False,
            "instructions": (
                f"From an independent {expected_role} session "
                f"({selection['required_namespace']} / "
                f"{selection['required_profile']}), call "
                f"gitea_adopt_workflow_lease(lease_id={result.lease_id!r}) "
                "to consume this controller allocation. The allocating "
                "controller process does not need to remain alive. Wrong-role "
                "and second-adoption attempts fail closed."
            ),
        }
        lease_proof["cross_role_handoff"] = True
        lease_proof["handoff_status"] = "pending"
        lease_proof["consume_tool"] = "gitea_adopt_workflow_lease"
    out = {
        "success": True,
        "outcome": OUTCOME_ASSIGNED,
        "apply": True,
@@ -1326,17 +1372,8 @@ def allocate_next_work(
            "lease_role": lease_role,
            "lease_proof": lease_proof,
            "selection_policy": SELECTION_POLICY,
            "cross_role_handoff": bool(cross_role_handoff),
        },
-        "next_valid_command": (
+        "next_valid_command": _next_command(lease_role, selected),
            (
                f"consume lease {result.lease_id} via gitea_adopt_workflow_lease "
                f"as {expected_role}, then "
            )
            + _next_command(lease_role, selected)
            if cross_role_handoff
            else _next_command(lease_role, selected)
        ),
        "substrate": "control_plane_db",
        "file_lock_only": False,
        "comment_lease_only": False,
@@ -1351,14 +1388,9 @@ def allocate_next_work(
        "downstream_note": (
            "#612 incident bridge remains downstream of #600; "
            "allocator never assigns raw monitoring incidents; "
-            "controller routes only under cross_role (#840); "
+            "controller routes only under cross_role (#840)"
            "cross-role assignments are consumable by independent "
            "required-role workers via gitea_adopt_workflow_lease (#843)"
        ),
    }
    if consume_allocation is not None:
        out["consume_allocation"] = consume_allocation
    return out
 def _next_command(role: str, c: WorkCandidate) -> str:
@@ -1410,6 +1442,7 @@ def candidate_from_dict(data: dict[str, Any]) -> WorkCandidate:
        state=str(data.get("state") or "open"),
        labels=tuple(data.get("labels") or ()),
        title=str(data.get("title") or ""),
        body=str(data.get("body") or ""),
        priority=priority,
        head_sha=data.get("head_sha"),
        request_changes_current_head=bool(data.get("request_changes_current_head")),
@@ -1637,13 +1637,11 @@ class ControlPlaneDB:
        provenance: dict[str, Any] | None = None,
        lease_ttl_seconds: int = DEFAULT_LEASE_TTL_SECONDS,
    ) -> dict[str, Any]:
-        """Transfer or refresh a lease with provenance (#601 / #843).
+        """Transfer or refresh a lease with provenance (#601).
        * Same owner + active → refresh (owner-resume).
        * Cross-role handoff pending + matching required role → atomic consume
          (even while the allocating controller session still "owns" the lease).
        * Expired/abandoned/released → create new assignment+lease with provenance.
-        * Active foreign (non-handoff) → raise ForeignLeaseError (never silent steal).
+        * Active foreign → raise ForeignLeaseError (never silent steal).
        """
        now = _utc_now()
        now_s = _ts(now)
@@ -1679,35 +1677,7 @@ class ControlPlaneDB:
                status = "expired"
            owner = lease["session_id"]
-            # Parse durable provenance for cross-role handoff consume (#843).
+            if status == "active" and owner != adopter_session_id:
            lease_prov: dict[str, Any] = {}
            if "provenance_json" in lease.keys() and lease["provenance_json"]:
                try:
                    loaded = json.loads(lease["provenance_json"])
                    if isinstance(loaded, dict):
                        lease_prov = loaded
                except (TypeError, json.JSONDecodeError):
                    lease_prov = {}
            handoff_pending = bool(lease_prov.get("cross_role_handoff")) and (
                str(lease_prov.get("handoff_status") or "pending").strip().lower()
                == "pending"
            )
            already_adopted = bool(
                (lease["adopted_by_session_id"] if "adopted_by_session_id" in lease.keys() else None)
                or lease_prov.get("adopted_by_session_id")
            )
            required_role = str(
                lease_prov.get("required_role") or lease["role"] or ""
            ).strip().lower()
            adopter_role = (role or "").strip().lower()
            cross_role_consume = (
                handoff_pending
                and not already_adopted
                and status == "active"
                and owner != adopter_session_id
            )
            if status == "active" and owner != adopter_session_id and not cross_role_consume:
                raise ForeignLeaseError(
                    f"cannot adopt active foreign lease {lease_id} owned by {owner}"
                )
@@ -1791,142 +1761,6 @@ class ControlPlaneDB:
                    "reasons": ["owner-resume: refreshed lease with provenance"],
                }
            # #843: controller→required-role handoff consume (atomic, same lease_id)
            if cross_role_consume:
                if not required_role:
                    raise ControlPlaneError(
                        f"cross-role handoff lease {lease_id} missing required_role"
                    )
                if adopter_role != required_role:
                    raise ForeignLeaseError(
                        f"wrong role for cross-role handoff consume: "
                        f"required={required_role} adopter={adopter_role or 'none'} "
                        f"(fail closed)"
                    )
                # CAS: only transfer if still owned by allocating session and unadopted
                cols = self._lease_columns(conn)
                adopted_col_null = (
                    "(adopted_by_session_id IS NULL OR adopted_by_session_id = '')"
                    if "adopted_by_session_id" in cols
                    else "1=1"
                )
                cas = conn.execute(
                    f"""
                    UPDATE leases
                    SET session_id = ?,
                        heartbeat_at = ?,
                        expires_at = ?,
                        phase = ?,
                        role = ?
                    WHERE lease_id = ?
                      AND status = 'active'
                      AND session_id = ?
                      AND {adopted_col_null}
                    """,
                    (
                        adopter_session_id,
                        now_s,
                        expires,
                        "adopted",
                        required_role,
                        lease_id,
                        owner,
                    ),
                )
                if cas.rowcount != 1:
                    raise ForeignLeaseError(
                        f"cross-role handoff consume lost race for lease {lease_id} "
                        "(already adopted or no longer pending; fail closed)"
                    )
                if "adopted_from_session_id" in cols:
                    conn.execute(
                        """
                        UPDATE leases
                        SET adopted_from_session_id = ?, adopted_by_session_id = ?
                        WHERE lease_id = ?
                        """,
                        (owner, adopter_session_id, lease_id),
                    )
                if "worktree_path" in cols and worktree_path:
                    conn.execute(
                        "UPDATE leases SET worktree_path = ? WHERE lease_id = ?",
                        (worktree_path, lease_id),
                    )
                if "owner_pid" in cols and owner_pid is not None:
                    conn.execute(
                        "UPDATE leases SET owner_pid = ? WHERE lease_id = ?",
                        (owner_pid, lease_id),
                    )
                if "expected_head_sha" in cols and expected_head_sha:
                    conn.execute(
                        "UPDATE leases SET expected_head_sha = ? WHERE lease_id = ?",
                        (expected_head_sha, lease_id),
                    )
                # Merge handoff provenance + caller provenance
                merged = dict(lease_prov)
                merged.update(provenance or {})
                merged["cross_role_handoff"] = True
                merged["handoff_status"] = "adopted"
                merged["adopted_from_session_id"] = owner
                merged["adopted_by_session_id"] = adopter_session_id
                merged["required_role"] = required_role
                if "provenance_json" in cols:
                    conn.execute(
                        "UPDATE leases SET provenance_json = ? WHERE lease_id = ?",
                        (json.dumps(merged), lease_id),
                    )
                # Transfer active assignment ownership atomically
                asn_cas = conn.execute(
                    """
                    UPDATE assignments
                    SET session_id = ?, role = ?
                    WHERE lease_id = ? AND status = 'active' AND session_id = ?
                    """,
                    (adopter_session_id, required_role, lease_id, owner),
                )
                if asn_cas.rowcount < 1:
                    # Fail closed: assignment must move with the lease
                    raise ControlPlaneError(
                        f"cross-role handoff: no active assignment for lease {lease_id} "
                        f"owned by {owner}"
                    )
                lease2 = conn.execute(
                    "SELECT * FROM leases WHERE lease_id = ?", (lease_id,)
                ).fetchone()
                asn = conn.execute(
                    """
                    SELECT * FROM assignments
                    WHERE lease_id = ? AND status = 'active'
                    ORDER BY created_at DESC LIMIT 1
                    """,
                    (lease_id,),
                ).fetchone()
                conn.execute(
                    """
                    INSERT INTO events(work_item_id, event_type, message, created_at)
                    VALUES (?, 'lease_adopted', ?, ?)
                    """,
                    (
                        lease["work_item_id"],
                        f"cross-role handoff: {adopter_session_id} consumed "
                        f"{lease_id} from {owner} as {required_role}",
                        now_s,
                    ),
                )
                return {
                    "outcome": "adopted_cross_role_handoff",
                    "lease": dict(lease2) if lease2 else dict(lease),
                    "assignment": dict(asn) if asn else None,
                    "reasons": [
                        "cross-role handoff: independent required-role worker consumed "
                        "controller allocation without abandonment"
                    ],
                    "adopted_by_session_id": adopter_session_id,
                    "adopted_from_session_id": owner,
                    "required_role": required_role,
                    "handoff_status": "adopted",
                }
            # Non-active: create new lease + assignment (transfer)
            new_lease_id = f"lease-{uuid.uuid4().hex[:16]}"
            new_asn_id = f"asn-{uuid.uuid4().hex[:16]}"
@@ -19912,6 +19912,7 @@ def _allocator_candidates_from_gitea(
                        state="open",
                        labels=tuple(labels),
                        title=title,
                        body=body,
                        priority=20 if "status:ready" in labels else 1,
                        blocked=blocked,
                        dependency_unmet=dep_unmet,
@@ -21148,21 +21149,10 @@ def gitea_adopt_workflow_lease(
    remote: str = "dadeschools",
    host: str | None = None,
 ) -> dict:
-    """Adopt a control-plane lease through the sanctioned path (#601 / #843).
+    """Adopt a control-plane lease through the sanctioned path (#601).
-    Same-owner resume refreshes provenance. Foreign active leases are refused
+    Same-owner resume refreshes provenance. Foreign active leases are refused.
-    unless the lease is a pending controller cross-role handoff and the caller
+    Expired leases may be reclaimed; provenance records adopted_from/by.
    holds the required role (independent consume without sharing the
    controller session). Expired leases may be reclaimed; provenance records
    adopted_from/by. Terminal (abandoned/released) leases cannot be adopted.
    #843 F1: the adopter role is derived authoritatively from the active
    authenticated profile — never from caller input. A supplied ``role`` that
    does not exactly match the profile-derived role is rejected (no silent
    accept or reinterpretation), and handoff provenance ``required_profile`` /
    ``required_namespace`` restrictions are validated against the same
    authoritative caller context. Caller-supplied role/profile/namespace can
    never grant authority.
    """
    read_block = _profile_operation_gate("gitea.read")
    if read_block:
@@ -21171,64 +21161,25 @@ def gitea_adopt_workflow_lease(
            "reasons": read_block,
            "permission_report": _permission_block_report("gitea.read"),
        }
    profile = get_profile()
    profile_name = (profile.get("profile_name") or "").strip() or "session"
    active_role = (_profile_role_kind(profile) or "").strip().lower()
    if not active_role:
        return {
            "success": False,
            "outcome": "blocked",
            "mutation_performed": False,
            "reasons": [
                "active profile role could not be derived authoritatively; "
                "refusing lease adoption (fail closed, #843)"
            ],
            "lease_id": lease_id,
            "authoritative_source": "control_plane_db",
            "file_lock_only": False,
            "comment_lease_only": False,
        }
    if role is not None and str(role).strip():
        supplied_role = str(role).strip().lower()
        if supplied_role != active_role:
            return {
                "success": False,
                "outcome": "blocked",
                "mutation_performed": False,
                "profile_role_kind": active_role,
                "supplied_role": supplied_role,
                "reasons": [
                    f"caller-supplied role '{supplied_role}' does not match "
                    f"the authenticated profile-derived role '{active_role}'; "
                    "caller-supplied role/profile/namespace can never grant "
                    "authority (fail closed, #843)"
                ],
                "lease_id": lease_id,
                "authoritative_source": "control_plane_db",
                "file_lock_only": False,
                "comment_lease_only": False,
            }
    db, errs = _control_plane_db_or_error()
    if db is None:
        return {"success": False, "reasons": errs}
    profile = get_profile()
    profile_name = (profile.get("profile_name") or "").strip() or "session"
    active_role = _profile_role_kind(profile) or "author"
    sid = (session_id or "").strip() or (
        f"{profile_name}-{os.getpid()}-{uuid.uuid4().hex[:8]}"
    )
    adopter_namespace = allocator_service.DEFAULT_ROLE_NAMESPACES.get(
        active_role, f"gitea-{active_role}"
    )
    try:
        return lease_lifecycle.adopt_lease(
            db,
            lease_id=lease_id,
            adopter_session_id=sid,
-            role=active_role,
+            role=(role or active_role).strip() or "author",
            worktree_path=worktree_path,
            expected_head_sha=expected_head_sha,
            owner_pid=os.getpid(),
            operator_authorized=bool(operator_authorized),
            adopter_profile_name=profile_name,
            adopter_namespace=adopter_namespace,
        )
    except (lease_lifecycle.LeaseLifecycleError, control_plane_db.ControlPlaneError) as exc:
        return {
@@ -39,7 +39,6 @@ SAFE_RELEASE_OWNED = "release_owned"
 SAFE_STALE_PROMPT = "stale_prompt_lease"
 SAFE_UNKNOWN = "inspect_only"
 SAFE_NO_AUTHORITY = "file_or_comment_not_authoritative"
 SAFE_CONSUME_CROSS_ROLE = "consume_cross_role_handoff"
 LEASE_STATUS_ACTIVE = "active"
 LEASE_STATUS_RELEASED = "released"
@@ -251,23 +250,6 @@ def decide_safe_next_action(
                "same_owner": True,
                "also_allowed": [SAFE_ABANDON_ALLOWED, SAFE_RELEASE_OWNED],
            }
        handoff = is_pending_cross_role_handoff({"lease": lease})
        if handoff:
            return {
                "safe_next_action": SAFE_CONSUME_CROSS_ROLE,
                "reasons": [
                    f"controller allocation pending handoff (freshness={status}); "
                    "required-role worker may consume without abandon/reassign; "
                    f"required_role={handoff['required_role']}"
                ],
                "block": False,
                "same_owner": False,
                "owner_session_id": owner,
                "required_role": handoff["required_role"],
                "cross_role_handoff": True,
                "handoff_status": "pending",
                "also_allowed": [SAFE_ABANDON_ALLOWED],
            }
        return {
            "safe_next_action": SAFE_ABANDON_ALLOWED,
            "reasons": [
@@ -290,24 +272,6 @@ def decide_safe_next_action(
        }
    if not same_owner and status == "active":
        # #843: pending cross-role handoff is consumable by required role
        handoff = is_pending_cross_role_handoff({"lease": lease})
        if handoff:
            return {
                "safe_next_action": SAFE_CONSUME_CROSS_ROLE,
                "reasons": [
                    "controller cross-role allocation pending handoff; "
                    f"required_role={handoff['required_role']}; "
                    "consume via gitea_adopt_workflow_lease without "
                    "abandonment or sharing the controller session"
                ],
                "block": False,
                "same_owner": False,
                "owner_session_id": owner,
                "required_role": handoff["required_role"],
                "cross_role_handoff": True,
                "handoff_status": "pending",
            }
        return {
            "safe_next_action": SAFE_WAIT_FOREIGN,
            "reasons": [
@@ -476,84 +440,6 @@ def list_active_leases(
    }
 def parse_lease_provenance(lease_or_state: Mapping[str, Any] | None) -> dict[str, Any]:
    """Return durable lease provenance dict (empty when absent/unparseable)."""
    if not lease_or_state:
        return {}
    if "provenance" in lease_or_state and isinstance(lease_or_state.get("provenance"), dict):
        return dict(lease_or_state["provenance"])
    raw = None
    if "provenance_json" in lease_or_state:
        raw = lease_or_state.get("provenance_json")
    elif "lease" in lease_or_state and isinstance(lease_or_state.get("lease"), Mapping):
        raw = lease_or_state["lease"].get("provenance_json")
    if not raw:
        return {}
    if isinstance(raw, dict):
        return dict(raw)
    try:
        loaded = json.loads(raw)
    except (TypeError, json.JSONDecodeError):
        return {}
    return dict(loaded) if isinstance(loaded, dict) else {}
 def is_pending_cross_role_handoff(
    state: Mapping[str, Any] | None,
 ) -> dict[str, Any] | None:
    """Return handoff evidence when a controller allocation awaits consume (#843).
    A pending handoff is identified by durable provenance written at
    cross-role apply time — not by title heuristics or session-id guessing.
    """
    if not state:
        return None
    lease = state.get("lease") if isinstance(state.get("lease"), Mapping) else state
    if not isinstance(lease, Mapping):
        return None
    status = str(lease.get("status") or "").strip().lower()
    if status in (LEASE_STATUS_ABANDONED, LEASE_STATUS_RELEASED, LEASE_STATUS_EXPIRED):
        return None
    prov = parse_lease_provenance(state)
    if not prov and isinstance(lease, Mapping):
        prov = parse_lease_provenance(lease)
    if not prov.get("cross_role_handoff"):
        return None
    handoff_status = str(prov.get("handoff_status") or "pending").strip().lower()
    if handoff_status != "pending":
        return None
    adopted_by = (
        lease.get("adopted_by_session_id")
        or prov.get("adopted_by_session_id")
        or ""
    )
    if str(adopted_by).strip():
        return None
    required_role = str(
        prov.get("required_role") or lease.get("role") or ""
    ).strip().lower()
    if not required_role:
        return None
    return {
        "cross_role_handoff": True,
        "handoff_status": "pending",
        "required_role": required_role,
        "allocating_session_id": str(
            prov.get("allocating_session_id") or lease.get("session_id") or ""
        ),
        "allocating_role": str(prov.get("allocating_role") or "controller"),
        "lease_id": str(lease.get("lease_id") or ""),
        "assignment_id": (
            str(state["assignment"]["assignment_id"])
            if isinstance(state.get("assignment"), Mapping)
            and state["assignment"].get("assignment_id")
            else None
        ),
        "provenance": prov,
    }
 def adopt_lease(
    db: cpd.ControlPlaneDB,
    *,
@@ -564,17 +450,8 @@ def adopt_lease(
    expected_head_sha: str | None = None,
    owner_pid: int | None = None,
    operator_authorized: bool = False,
    adopter_profile_name: str | None = None,
    adopter_namespace: str | None = None,
 ) -> dict[str, Any]:
-    """Sanctioned adopt path with provenance; never silent foreign steal.
+    """Sanctioned adopt path with provenance; never silent foreign steal."""
    #843 F1: for a pending cross-role handoff, ``role`` must be the
    authoritative profile-derived role supplied by the MCP boundary — never
    caller-asserted authority. When the handoff provenance declares
    ``required_profile`` / ``required_namespace`` and the caller context is
    provided, both are validated exactly; a mismatch fails closed.
    """
    state = db.get_lease_workflow_state(lease_id)
    if not state:
        raise LeaseLifecycleError(
@@ -586,8 +463,11 @@ def adopt_lease(
    owner = str(lease.get("session_id") or "")
    same_owner = owner == str(adopter_session_id)
-    handoff = is_pending_cross_role_handoff(state)
+    if freshness["freshness"] == "active" and not same_owner:
-    adopter_role = (role or "").strip().lower()
+        raise LeaseLifecycleError(
            f"refusing to steal active foreign lease {lease_id} owned by "
            f"{owner} (fail closed)"
        )
    if freshness["freshness"] in ("abandoned", "released"):
        raise LeaseLifecycleError(
@@ -595,64 +475,13 @@ def adopt_lease(
            "(fail closed)"
        )
-    if handoff and not same_owner:
+    # Expired or stale: require abandon-style safety before ownership transfer
-        # Terminal statuses already rejected above. Freshness may be
+    # when not same owner; same owner may reclaim.
-        # active OR stale_dead_process (controller exited) — both are
+    if not same_owner and freshness["freshness"] in (
        # consumable without abandonment when handoff is still pending.
        if freshness["freshness"] not in (
            "active",
            "stale_dead_process",
            "stale_missing_worktree",
        ):
            raise LeaseLifecycleError(
                f"lease {lease_id} freshness={freshness['freshness']}; "
                "terminal or non-active allocation cannot be handoff-consumed "
                "(fail closed)"
            )
        required = handoff["required_role"]
        if adopter_role != required:
            raise LeaseLifecycleError(
                f"wrong role for cross-role handoff consume of {lease_id}: "
                f"required={required} adopter={adopter_role or 'none'} "
                "(fail closed)"
            )
        # #843 F1: provenance profile/namespace restrictions are validated
        # against the authoritative caller context when declared. Caller
        # input can never widen authority; a mismatch fails closed.
        handoff_prov = handoff.get("provenance") or {}
        required_profile = str(
            handoff_prov.get("required_profile") or ""
        ).strip()
        if required_profile and adopter_profile_name is not None:
            if str(adopter_profile_name).strip() != required_profile:
                raise LeaseLifecycleError(
                    f"wrong profile for cross-role handoff consume of "
                    f"{lease_id}: required_profile={required_profile} "
                    f"adopter_profile={adopter_profile_name} (fail closed)"
                )
        required_namespace = str(
            handoff_prov.get("required_namespace") or ""
        ).strip()
        if required_namespace and adopter_namespace is not None:
            if str(adopter_namespace).strip() != required_namespace:
                raise LeaseLifecycleError(
                    f"wrong namespace for cross-role handoff consume of "
                    f"{lease_id}: required_namespace={required_namespace} "
                    f"adopter_namespace={adopter_namespace} (fail closed)"
                )
        reason = "cross-role-handoff-consume"
    elif freshness["freshness"] == "active" and not same_owner:
        raise LeaseLifecycleError(
            f"refusing to steal active foreign lease {lease_id} owned by "
            f"{owner} (fail closed)"
        )
    elif not same_owner and freshness["freshness"] in (
        "expired",
        "stale_dead_process",
        "stale_missing_worktree",
    ):
        # Expired or stale (non-handoff): require abandon-style safety before
        # ownership transfer when not same owner; same owner may reclaim.
        if not operator_authorized and freshness["freshness"] == "expired":
            # Deterministic reclaim of expired foreign lease is allowed
            # without operator flag (sanctioned expire reclaim).
@@ -663,9 +492,6 @@ def adopt_lease(
                f"lease {lease_id} freshness={freshness['freshness']}; "
                "use abandon with proof before foreign adopt (fail closed)"
            )
        reason = "sanctioned-reclaim-adopt"
    else:
        reason = "owner-resume-adopt" if same_owner else "sanctioned-reclaim-adopt"
    provenance = build_adopt_provenance(
        adopted_from_session_id=owner,
@@ -678,14 +504,10 @@ def adopt_lease(
        worktree_path=worktree_path,
        expected_head_sha=expected_head_sha or lease.get("expected_head_sha"),
        prior_lease_id=lease_id,
-        reason=reason,
+        reason=(
            "owner-resume-adopt" if same_owner else "sanctioned-reclaim-adopt"
        ),
    )
    if handoff and not same_owner:
        provenance["cross_role_handoff"] = True
        provenance["handoff_status"] = "adopted"
        provenance["required_role"] = handoff["required_role"]
        provenance["allocating_session_id"] = handoff["allocating_session_id"]
        provenance["allocating_role"] = handoff["allocating_role"]
    result = db.adopt_lease(
        lease_id=lease_id,
@@ -696,7 +518,7 @@ def adopt_lease(
        owner_pid=owner_pid if owner_pid is not None else os.getpid(),
        provenance=provenance,
    )
-    out = {
+    return {
        "success": True,
        "outcome": result.get("outcome"),
        "same_owner": same_owner,
@@ -709,24 +531,6 @@ def adopt_lease(
        "comment_lease_only": False,
        "reasons": result.get("reasons") or [],
    }
    if handoff and not same_owner:
        out["cross_role_handoff"] = True
        out["handoff_status"] = "adopted"
        out["required_role"] = handoff["required_role"]
        out["adopted_by_session_id"] = adopter_session_id
        out["adopted_from_session_id"] = owner
        lease_row = result.get("lease") or {}
        if isinstance(lease_row, Mapping):
            out["read_after_write"] = {
                "lease_id": lease_row.get("lease_id"),
                "session_id": lease_row.get("session_id"),
                "role": lease_row.get("role"),
                "status": lease_row.get("status"),
                "adopted_by_session_id": lease_row.get("adopted_by_session_id"),
                "adopted_from_session_id": lease_row.get("adopted_from_session_id"),
                "phase": lease_row.get("phase"),
            }
    return out
 def release_lease(
@@ -0,0 +1,243 @@
 """Allocator epic / child-only container pre-rank exclusion (#844).
 Covers:
 * Issue #631-shaped child-only epic is excluded before ranking.
 * Implementable child issues remain eligible and can be selected.
 * Ordinary issues that merely mention "epic" in title/body are not excluded.
 * Excluded containers never receive assignments or workflow leases.
 * Structured skip reason ``epic_or_child_only_container`` is reported.
 """
 from __future__ import annotations
 import os
 import tempfile
 import unittest
 from allocator_service import (
    OUTCOME_ASSIGNED,
    OUTCOME_PREVIEW,
    SKIP_EPIC_OR_CHILD_ONLY_CONTAINER,
    WorkCandidate,
    allocate_next_work,
    classify_epic_or_child_only_container,
 )
 from control_plane_db import ControlPlaneDB
 REMOTE = "prgs"
 ORG = "Scaled-Tech-Consulting"
 REPO = "Gitea-Tools"
 # Minimal body mirroring issue #631 authoritative scope language.
 _EPIC_631_BODY = """
 ## Scope (umbrella)
 This epic owns the **product roadmap and linkage** for the Web Console.
 Implementation is delivered via child issues only.
 ## Explicit non-goals
 * Do not implement product features in this epic issue itself.
 * No product feature implementation is claimed complete solely on this epic.
 """
 _CHILD_BODY = """
 ## Problem
 Operators need a workflow-event timeline model for Phase 1.
 ## Acceptance criteria
 - [ ] Timeline model API exists
 """
 def _issue(
    number: int,
    *,
    title: str = "",
    body: str = "",
    labels: tuple[str, ...] = ("status:ready", "type:feature"),
    priority: int = 20,
 ) -> WorkCandidate:
    return WorkCandidate(
        kind="issue",
        number=number,
        state="open",
        labels=labels,
        title=title or f"issue {number}",
        body=body,
        priority=priority,
    )
 class ClassifyEpicContainerTest(unittest.TestCase):
    def test_631_shaped_body_and_title_is_container(self) -> None:
        c = _issue(
            631,
            title="Epic: MCP Control Plane Web Console",
            body=_EPIC_631_BODY,
        )
        is_c, detail = classify_epic_or_child_only_container(c)
        self.assertTrue(is_c)
        self.assertIsNotNone(detail)
        self.assertIn("body_marker", detail or "")
    def test_body_markers_without_epic_title(self) -> None:
        c = _issue(
            900,
            title="Control plane roadmap tracker",
            body="Implementation is delivered via child issues only.",
        )
        is_c, _ = classify_epic_or_child_only_container(c)
        self.assertTrue(is_c)
    def test_epic_label_alone_is_container(self) -> None:
        c = _issue(
            901,
            title="Roadmap linkage",
            body="Track children.",
            labels=("status:ready", "type:epic"),
        )
        is_c, detail = classify_epic_or_child_only_container(c)
        self.assertTrue(is_c)
        self.assertIn("type:epic", detail or "")
    def test_title_epic_prefix_alone_not_container(self) -> None:
        """Title-only 'Epic:' without body scope evidence stays eligible (#844)."""
        c = _issue(
            902,
            title="Epic: something mentioned only in title",
            body="Implement a concrete fix for the allocator skip list.",
        )
        is_c, detail = classify_epic_or_child_only_container(c)
        self.assertFalse(is_c)
        self.assertIsNone(detail)
    def test_incidental_epic_word_not_container(self) -> None:
        c = _issue(
            903,
            title="Document epic handoff conventions",
            body=(
                "Update the docs so implementable issues that mention an epic "
                "remain independently executable."
            ),
        )
        is_c, _ = classify_epic_or_child_only_container(c)
        self.assertFalse(is_c)
    def test_prs_never_classified(self) -> None:
        pr = WorkCandidate(
            kind="pr",
            number=10,
            state="open",
            title="Epic: fake",
            body="Implementation is delivered via child issues only.",
            head_sha="a" * 40,
            priority=5,
        )
        is_c, _ = classify_epic_or_child_only_container(pr)
        self.assertFalse(is_c)
 class AllocateEpicContainerExclusionTest(unittest.TestCase):
    def setUp(self) -> None:
        self._tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self._tmp.cleanup)
        self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
    def _alloc(self, candidates, **kwargs):
        defaults = dict(
            session_id="sess-844",
            role="author",
            remote=REMOTE,
            org=ORG,
            repo=REPO,
            profile_name="prgs-author",
            username="jcwalker3",
            claims={},
            apply=False,
        )
        defaults.update(kwargs)
        return allocate_next_work(self.db, candidates=candidates, **defaults)
    def test_631_shaped_epic_excluded_child_selected(self) -> None:
        epic = _issue(
            631,
            title="Epic: MCP Control Plane Web Console",
            body=_EPIC_631_BODY,
        )
        child = _issue(
            637,
            title="Web Console: Workflow-event timeline model (Phase 1)",
            body=_CHILD_BODY,
        )
        res = self._alloc([epic, child], apply=False)
        self.assertTrue(res["success"], res)
        self.assertEqual(res["outcome"], OUTCOME_PREVIEW)
        self.assertEqual(res["selected"]["number"], 637)
        skipped = {s["number"]: s for s in res["skipped"]}
        self.assertIn(631, skipped)
        self.assertEqual(
            skipped[631]["reason_code"], SKIP_EPIC_OR_CHILD_ONLY_CONTAINER
        )
        self.assertIn(SKIP_EPIC_OR_CHILD_ONLY_CONTAINER, skipped[631]["reason"])
    def test_container_cannot_receive_assignment_or_lease(self) -> None:
        epic = _issue(
            631,
            title="Epic: MCP Control Plane Web Console",
            body=_EPIC_631_BODY,
        )
        res = self._alloc([epic], apply=True)
        self.assertTrue(res["success"], res)
        # Only container present → no safe work; never assigned_work.
        self.assertNotEqual(res["outcome"], OUTCOME_ASSIGNED)
        self.assertIsNone(res.get("assignment"))
        self.assertIsNone(res.get("selected"))
        skipped = {s["number"]: s for s in res["skipped"]}
        self.assertEqual(
            skipped[631]["reason_code"], SKIP_EPIC_OR_CHILD_ONLY_CONTAINER
        )
        # No lease row for the epic.
        leases = self.db.list_active_leases(
            remote=REMOTE, org=ORG, repo=REPO
        ) if hasattr(self.db, "list_active_leases") else []
        # Prefer generic inventory if available.
        if not leases and hasattr(self.db, "list_leases"):
            leases = self.db.list_leases(remote=REMOTE, org=ORG, repo=REPO)
        for lease in leases or []:
            work_number = lease.get("work_number") if isinstance(lease, dict) else None
            self.assertNotEqual(work_number, 631)
    def test_incidental_epic_title_remains_eligible(self) -> None:
        ordinary = _issue(
            700,
            title="Document epic handoff conventions",
            body="Write runbook text about epic vs child issues.",
        )
        res = self._alloc([ordinary], apply=False)
        self.assertTrue(res["success"], res)
        self.assertEqual(res["selected"]["number"], 700)
        self.assertEqual(res["skipped"], [])
    def test_apply_selects_child_not_epic(self) -> None:
        epic = _issue(
            631,
            title="Epic: MCP Control Plane Web Console",
            body=_EPIC_631_BODY,
        )
        child = _issue(
            637,
            title="Web Console: Workflow-event timeline model (Phase 1)",
            body=_CHILD_BODY,
        )
        res = self._alloc([epic, child], apply=True)
        self.assertTrue(res["success"], res)
        self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
        self.assertEqual(res["selected"]["number"], 637)
        self.assertEqual(res["assignment"]["work_number"], 637)
 if __name__ == "__main__":
    unittest.main()
@@ -1,682 +0,0 @@
 """Cross-role allocation handoff consumable by independent workers (#843).
 Regression coverage for the controller→required-role consume path:
 * controller allocates author work; independent author adopts successfully
 * author adoption succeeds after allocating controller process exits
 * author adoption without sharing controller session identity
 * wrong-role adoption rejected
 * concurrent/second adoption rejected without state corruption
 * terminal allocation adoption rejected
 * successful adoption produces authoritative ownership evidence
 * genuine abandoned-lease recovery remains valid
 * process_work_queue / allocate results include consume identifiers
 * same-role allocation behavior remains compatible
 """
 from __future__ import annotations
 import os
 import tempfile
 import unittest
 from datetime import timedelta
 from unittest.mock import patch
 from allocator_service import (
    ALLOCATION_MODE_CROSS_ROLE,
    ALLOCATION_MODE_ROLE_SCOPED,
    OUTCOME_ASSIGNED,
    ROLE_AUTHOR,
    ROLE_CONTROLLER,
    ROLE_REVIEWER,
    WorkCandidate,
    allocate_next_work,
 )
 from control_plane_db import ControlPlaneDB, ForeignLeaseError, _ts, _utc_now
 import lease_lifecycle as ll
 class CrossRoleHandoffTest(unittest.TestCase):
    def setUp(self) -> None:
        self._tmp = tempfile.TemporaryDirectory()
        self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
        self.db = ControlPlaneDB(self.db_path)
        self.db.upsert_session(
            session_id="ctrl-session",
            role="controller",
            profile="prgs-controller",
            pid=99999999,  # dead-looking pid
        )
        self.db.upsert_session(
            session_id="author-worker",
            role="author",
            profile="prgs-author",
            pid=os.getpid(),
        )
        self.db.upsert_session(
            session_id="author-worker-2",
            role="author",
            profile="prgs-author",
            pid=os.getpid(),
        )
        self.db.upsert_session(
            session_id="reviewer-worker",
            role="reviewer",
            profile="prgs-reviewer",
            pid=os.getpid(),
        )
        self.wt = self._tmp.name
    def tearDown(self) -> None:
        self._tmp.cleanup()
    def _ready_issue(self, number: int = 843, title: str = "handoff target") -> WorkCandidate:
        return WorkCandidate(
            kind="issue",
            number=number,
            labels=("status:ready", "type:bug"),
            title=title,
            priority=20,
        )
    def _controller_allocate(self, number: int = 843, **kwargs):
        defaults = dict(
            db=self.db,
            session_id="ctrl-session",
            role=ROLE_CONTROLLER,
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            candidates=[self._ready_issue(number)],
            apply=True,
            profile_name="prgs-controller",
            username="controller-user",
            allocation_mode=ALLOCATION_MODE_CROSS_ROLE,
        )
        defaults.update(kwargs)
        return allocate_next_work(**defaults)
    def test_controller_allocates_author_independent_author_adopts(self) -> None:
        res = self._controller_allocate()
        self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
        self.assertEqual(res["required_role"], ROLE_AUTHOR)
        self.assertIn("consume_allocation", res)
        consume = res["consume_allocation"]
        self.assertEqual(consume["tool"], "gitea_adopt_workflow_lease")
        self.assertEqual(consume["required_role"], ROLE_AUTHOR)
        self.assertFalse(consume["controller_session_required"])
        lid = res["assignment"]["lease_id"]
        self.assertEqual(consume["lease_id"], lid)
        self.assertIn(lid, res["next_valid_command"])
        adopted = ll.adopt_lease(
            self.db,
            lease_id=lid,
            adopter_session_id="author-worker",
            role=ROLE_AUTHOR,
            worktree_path=self.wt,
        )
        self.assertTrue(adopted["success"])
        self.assertEqual(adopted["outcome"], "adopted_cross_role_handoff")
        self.assertEqual(adopted["adopted_by_session_id"], "author-worker")
        self.assertEqual(adopted["adopted_from_session_id"], "ctrl-session")
        raw = adopted["read_after_write"]
        self.assertEqual(raw["session_id"], "author-worker")
        self.assertEqual(raw["adopted_by_session_id"], "author-worker")
        self.assertEqual(raw["status"], "active")
        self.assertEqual(raw["phase"], "adopted")
        # Authoritative re-read
        state = self.db.get_lease_workflow_state(lid)
        self.assertEqual(state["lease"]["session_id"], "author-worker")
        self.assertEqual(state["lease"]["adopted_by_session_id"], "author-worker")
        self.assertEqual(state["assignment"]["session_id"], "author-worker")
        self.assertEqual(state["provenance"]["handoff_status"], "adopted")
    def test_author_adoption_after_controller_process_exits(self) -> None:
        res = self._controller_allocate(number=900)
        lid = res["assignment"]["lease_id"]
        # Force owner_pid dead + freshness stale_dead_process
        import sqlite3
        conn = sqlite3.connect(self.db_path)
        try:
            conn.execute(
                "UPDATE leases SET owner_pid = 99999999 WHERE lease_id = ?",
                (lid,),
            )
            conn.commit()
        finally:
            conn.close()
        state = self.db.get_lease_workflow_state(lid)
        fr = ll.classify_lease_freshness(
            state["lease"], pid_checker=lambda _p: False
        )
        self.assertEqual(fr["freshness"], "stale_dead_process")
        adopted = ll.adopt_lease(
            self.db,
            lease_id=lid,
            adopter_session_id="author-worker",
            role=ROLE_AUTHOR,
            worktree_path=self.wt,
        )
        self.assertEqual(adopted["outcome"], "adopted_cross_role_handoff")
        self.assertEqual(adopted["adopted_by_session_id"], "author-worker")
        # No abandon required
        state2 = self.db.get_lease_workflow_state(lid)
        self.assertEqual(state2["lease"]["status"], "active")
        self.assertNotEqual(state2["lease"]["status"], "abandoned")
    def test_adoption_without_sharing_controller_session_identity(self) -> None:
        res = self._controller_allocate(number=901)
        lid = res["assignment"]["lease_id"]
        adopted = ll.adopt_lease(
            self.db,
            lease_id=lid,
            adopter_session_id="author-worker",
            role=ROLE_AUTHOR,
            worktree_path=self.wt,
        )
        self.assertNotEqual(adopted["adopted_by_session_id"], "ctrl-session")
        self.assertFalse(adopted["same_owner"])
        self.assertEqual(adopted["adopted_from_session_id"], "ctrl-session")
    def test_wrong_role_adoption_rejected(self) -> None:
        res = self._controller_allocate(number=902)
        lid = res["assignment"]["lease_id"]
        with self.assertRaises(ll.LeaseLifecycleError) as ctx:
            ll.adopt_lease(
                self.db,
                lease_id=lid,
                adopter_session_id="reviewer-worker",
                role=ROLE_REVIEWER,
            )
        self.assertIn("wrong role", str(ctx.exception).lower())
        # State unchanged
        state = self.db.get_lease_workflow_state(lid)
        self.assertEqual(state["lease"]["session_id"], "ctrl-session")
        self.assertIsNone(state["lease"].get("adopted_by_session_id") or None)
        self.assertEqual(state["provenance"]["handoff_status"], "pending")
    def test_second_adoption_rejected_without_corruption(self) -> None:
        res = self._controller_allocate(number=903)
        lid = res["assignment"]["lease_id"]
        first = ll.adopt_lease(
            self.db,
            lease_id=lid,
            adopter_session_id="author-worker",
            role=ROLE_AUTHOR,
            worktree_path=self.wt,
        )
        self.assertEqual(first["outcome"], "adopted_cross_role_handoff")
        with self.assertRaises(ll.LeaseLifecycleError):
            ll.adopt_lease(
                self.db,
                lease_id=lid,
                adopter_session_id="author-worker-2",
                role=ROLE_AUTHOR,
                worktree_path=self.wt,
            )
        state = self.db.get_lease_workflow_state(lid)
        self.assertEqual(state["lease"]["session_id"], "author-worker")
        self.assertEqual(state["lease"]["adopted_by_session_id"], "author-worker")
        self.assertEqual(state["assignment"]["session_id"], "author-worker")
        self.assertEqual(state["lease"]["status"], "active")
    def test_terminal_allocation_adoption_rejected(self) -> None:
        res = self._controller_allocate(number=904)
        lid = res["assignment"]["lease_id"]
        # Abandon as terminal
        proof = ll.AbandonProof(
            dead_process=True,
            missing_worktree=True,
            no_open_pr=True,
            no_live_mutation_risk=True,
            owner_pid=99999999,
            worktree_path="/nonexistent/for-843",
        )
        # Attach dead pid / missing wt for abandon eligibility
        import sqlite3
        conn = sqlite3.connect(self.db_path)
        try:
            conn.execute(
                "UPDATE leases SET owner_pid = 99999999, worktree_path = ? WHERE lease_id = ?",
                ("/nonexistent/for-843", lid),
            )
            conn.commit()
        finally:
            conn.close()
        abandoned = ll.abandon_lease(
            self.db,
            lease_id=lid,
            requester_session_id="author-worker",
            proof=proof,
        )
        self.assertEqual(abandoned["outcome"], "abandoned")
        with self.assertRaises(ll.LeaseLifecycleError) as ctx:
            ll.adopt_lease(
                self.db,
                lease_id=lid,
                adopter_session_id="author-worker",
                role=ROLE_AUTHOR,
            )
        self.assertIn("abandoned", str(ctx.exception).lower())
    def test_successful_adoption_read_after_write_ownership(self) -> None:
        res = self._controller_allocate(number=905)
        lid = res["assignment"]["lease_id"]
        adopted = ll.adopt_lease(
            self.db,
            lease_id=lid,
            adopter_session_id="author-worker",
            role=ROLE_AUTHOR,
            worktree_path=self.wt,
        )
        raw = adopted["read_after_write"]
        self.assertEqual(raw["lease_id"], lid)
        self.assertEqual(raw["session_id"], "author-worker")
        self.assertEqual(raw["adopted_by_session_id"], "author-worker")
        self.assertEqual(raw["adopted_from_session_id"], "ctrl-session")
        # Re-fetch proves durable write
        state = self.db.get_lease_workflow_state(lid)
        self.assertEqual(state["lease"]["session_id"], raw["session_id"])
        self.assertEqual(
            state["lease"]["adopted_by_session_id"], raw["adopted_by_session_id"]
        )
    def test_genuine_abandoned_recovery_still_valid(self) -> None:
        """Same-role author lease abandoned remains reclaimable via abandon path."""
        same = allocate_next_work(
            self.db,
            session_id="author-worker",
            role=ROLE_AUTHOR,
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            candidates=[self._ready_issue(906, "same-role")],
            apply=True,
            profile_name="prgs-author",
            username="author-user",
            allocation_mode=ALLOCATION_MODE_ROLE_SCOPED,
        )
        self.assertEqual(same["outcome"], OUTCOME_ASSIGNED)
        lid = same["assignment"]["lease_id"]
        import sqlite3
        conn = sqlite3.connect(self.db_path)
        try:
            conn.execute(
                "UPDATE leases SET owner_pid = 99999999, worktree_path = ? WHERE lease_id = ?",
                ("/nonexistent/same-role", lid),
            )
            conn.commit()
        finally:
            conn.close()
        proof = ll.AbandonProof(
            dead_process=True,
            missing_worktree=True,
            no_open_pr=True,
            no_live_mutation_risk=True,
            owner_pid=99999999,
            worktree_path="/nonexistent/same-role",
        )
        abandoned = ll.abandon_lease(
            self.db,
            lease_id=lid,
            requester_session_id="author-worker-2",
            proof=proof,
        )
        self.assertEqual(abandoned["outcome"], "abandoned")
        # Foreign author cannot handoff-consume an abandoned non-handoff lease
        with self.assertRaises(ll.LeaseLifecycleError):
            ll.adopt_lease(
                self.db,
                lease_id=lid,
                adopter_session_id="author-worker-2",
                role=ROLE_AUTHOR,
            )
        # Reclaim path still works for expired/abandoned after force-expire
        reclaimed = ll.reclaim_expired_lease(
            self.db,
            lease_id=lid,
            session_id="author-worker-2",
            role=ROLE_AUTHOR,
            worktree_path=self.wt,
        )
        self.assertEqual(reclaimed["outcome"], "reclaimed")
        self.assertEqual(reclaimed["assignment"]["session_id"], "author-worker-2")
    def test_allocate_payload_includes_consume_identifiers(self) -> None:
        res = self._controller_allocate(number=907)
        self.assertIn("consume_allocation", res)
        c = res["consume_allocation"]
        for key in (
            "tool",
            "lease_id",
            "assignment_id",
            "required_role",
            "required_profile",
            "required_namespace",
            "instructions",
            "handoff_status",
        ):
            self.assertIn(key, c)
        self.assertEqual(c["required_namespace"], "gitea-author")
        self.assertEqual(c["required_profile"], "prgs-author")
        self.assertIn("gitea_adopt_workflow_lease", c["instructions"])
        self.assertTrue(res["lease_proof"]["cross_role_handoff"])
        self.assertEqual(res["lease_proof"]["handoff_status"], "pending")
    def test_same_role_allocation_remains_compatible(self) -> None:
        res = allocate_next_work(
            self.db,
            session_id="author-worker",
            role=ROLE_AUTHOR,
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            candidates=[self._ready_issue(908)],
            apply=True,
            profile_name="prgs-author",
            username="author-user",
        )
        self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
        self.assertNotIn("consume_allocation", res)
        lid = res["assignment"]["lease_id"]
        state = self.db.get_lease_workflow_state(lid)
        # No cross-role handoff provenance
        prov = state.get("provenance") or {}
        self.assertFalse(prov.get("cross_role_handoff"))
        # Owner resume still works
        resume = ll.adopt_lease(
            self.db,
            lease_id=lid,
            adopter_session_id="author-worker",
            role=ROLE_AUTHOR,
            worktree_path=self.wt,
        )
        self.assertTrue(resume["same_owner"])
        self.assertEqual(resume["outcome"], "adopted_owner_resume")
    def test_inspect_points_required_role_at_consume(self) -> None:
        res = self._controller_allocate(number=909)
        lid = res["assignment"]["lease_id"]
        decision = ll.inspect_lease(
            self.db, lid, caller_session_id="author-worker"
        )
        self.assertEqual(
            decision["safe_next_action"], ll.SAFE_CONSUME_CROSS_ROLE
        )
        self.assertFalse(decision["block"])
        self.assertEqual(decision["required_role"], ROLE_AUTHOR)
    def test_db_cas_rejects_concurrent_second_consume(self) -> None:
        res = self._controller_allocate(number=910)
        lid = res["assignment"]["lease_id"]
        # First consume via DB layer directly
        first = self.db.adopt_lease(
            lease_id=lid,
            adopter_session_id="author-worker",
            role=ROLE_AUTHOR,
            worktree_path=self.wt,
            provenance={
                "cross_role_handoff": True,
                "handoff_status": "adopted",
                "required_role": "author",
            },
        )
        self.assertEqual(first["outcome"], "adopted_cross_role_handoff")
        # Second CAS must fail
        with self.assertRaises(ForeignLeaseError):
            self.db.adopt_lease(
                lease_id=lid,
                adopter_session_id="author-worker-2",
                role=ROLE_AUTHOR,
                worktree_path=self.wt,
                provenance={
                    "cross_role_handoff": True,
                    "handoff_status": "pending",
                    "required_role": "author",
                },
            )
        state = self.db.get_lease_workflow_state(lid)
        self.assertEqual(state["lease"]["session_id"], "author-worker")
 class MCPBoundaryAdoptRoleBindingTest(unittest.TestCase):
    """#843 F1: MCP-boundary role binding for ``gitea_adopt_workflow_lease``.
    The library-level wrong-role test calls ``lease_lifecycle.adopt_lease``
    directly. These tests prove the MCP entry point derives the adopter role
    authoritatively from the active authenticated profile and rejects any
    caller-supplied role that disagrees, so a reviewer/merger profile cannot
    consume an author handoff by passing ``role="author"``.
    """
    AUTHOR_PROFILE = {
        "profile_name": "prgs-author",
        "role": "author",
        "allowed_operations": [
            "gitea.read",
            "gitea.pr.create",
            "gitea.branch.push",
        ],
        "forbidden_operations": [],
    }
    REVIEWER_PROFILE = {
        "profile_name": "prgs-reviewer",
        "role": "reviewer",
        "allowed_operations": [
            "gitea.read",
            "gitea.pr.review",
            "gitea.pr.approve",
            "gitea.pr.request_changes",
        ],
        "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
    }
    MERGER_PROFILE = {
        "profile_name": "prgs-merger",
        "role": "merger",
        "allowed_operations": ["gitea.read", "gitea.pr.merge"],
        "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
    }
    FOREIGN_AUTHOR_PROFILE = {
        "profile_name": "dadeschools-author",
        "role": "author",
        "allowed_operations": [
            "gitea.read",
            "gitea.pr.create",
            "gitea.branch.push",
        ],
        "forbidden_operations": [],
    }
    def setUp(self) -> None:
        self._tmp = tempfile.TemporaryDirectory()
        self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
        self.db = ControlPlaneDB(self.db_path)
        self.db.upsert_session(
            session_id="ctrl-session",
            role="controller",
            profile="prgs-controller",
            pid=99999999,
        )
        self.db.upsert_session(
            session_id="author-worker",
            role="author",
            profile="prgs-author",
            pid=os.getpid(),
        )
        self.wt = self._tmp.name
    def tearDown(self) -> None:
        self._tmp.cleanup()
    def _ready_issue(self, number: int) -> WorkCandidate:
        return WorkCandidate(
            kind="issue",
            number=number,
            labels=("status:ready", "type:bug"),
            title="handoff target",
            priority=20,
        )
    def _handoff_lease(self, number: int = 843) -> str:
        res = allocate_next_work(
            db=self.db,
            session_id="ctrl-session",
            role=ROLE_CONTROLLER,
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            candidates=[self._ready_issue(number)],
            apply=True,
            profile_name="prgs-controller",
            username="controller-user",
            allocation_mode=ALLOCATION_MODE_CROSS_ROLE,
        )
        self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
        self.assertEqual(res["required_role"], ROLE_AUTHOR)
        return res["assignment"]["lease_id"]
    def _call_adopt_tool(self, profile: dict, **kwargs):
        import gitea_mcp_server as mcp_server
        with (
            patch.object(mcp_server, "get_profile", return_value=profile),
            patch.object(
                mcp_server,
                "_control_plane_db_or_error",
                return_value=(self.db, []),
            ),
        ):
            return mcp_server.gitea_adopt_workflow_lease(
                remote="prgs", **kwargs
            )
    def _assert_handoff_untouched(self, lease_id: str) -> None:
        state = self.db.get_lease_workflow_state(lease_id)
        self.assertEqual(state["lease"]["session_id"], "ctrl-session")
        self.assertIsNone(state["lease"].get("adopted_by_session_id") or None)
        self.assertEqual(state["lease"]["status"], "active")
        self.assertEqual(state["provenance"]["handoff_status"], "pending")
    def test_reviewer_profile_cannot_consume_author_handoff_via_role_author(
        self,
    ) -> None:
        lid = self._handoff_lease(920)
        result = self._call_adopt_tool(
            self.REVIEWER_PROFILE,
            lease_id=lid,
            session_id="reviewer-worker",
            role="author",
            worktree_path=self.wt,
        )
        self.assertFalse(result["success"])
        self.assertEqual(result["outcome"], "blocked")
        self.assertEqual(result["profile_role_kind"], "reviewer")
        self.assertEqual(result["supplied_role"], "author")
        self.assertIn("does not match", result["reasons"][0])
        self._assert_handoff_untouched(lid)
    def test_merger_profile_cannot_consume_author_handoff_via_role_author(
        self,
    ) -> None:
        lid = self._handoff_lease(921)
        result = self._call_adopt_tool(
            self.MERGER_PROFILE,
            lease_id=lid,
            session_id="merger-worker",
            role="author",
            worktree_path=self.wt,
        )
        self.assertFalse(result["success"])
        self.assertEqual(result["outcome"], "blocked")
        self.assertEqual(result["profile_role_kind"], "merger")
        self._assert_handoff_untouched(lid)
    def test_reviewer_profile_rejected_without_role_argument(self) -> None:
        """Even without a spoofed role, the profile-derived role binds."""
        lid = self._handoff_lease(922)
        result = self._call_adopt_tool(
            self.REVIEWER_PROFILE,
            lease_id=lid,
            session_id="reviewer-worker",
            worktree_path=self.wt,
        )
        self.assertFalse(result["success"])
        self.assertEqual(result["outcome"], "blocked")
        self.assertIn("wrong role", result["reasons"][0].lower())
        self._assert_handoff_untouched(lid)
    def test_author_profile_mismatching_supplied_role_rejected(self) -> None:
        lid = self._handoff_lease(923)
        result = self._call_adopt_tool(
            self.AUTHOR_PROFILE,
            lease_id=lid,
            session_id="author-worker",
            role="reviewer",
            worktree_path=self.wt,
        )
        self.assertFalse(result["success"])
        self.assertEqual(result["outcome"], "blocked")
        self.assertEqual(result["profile_role_kind"], "author")
        self.assertEqual(result["supplied_role"], "reviewer")
        self._assert_handoff_untouched(lid)
    def test_foreign_profile_name_rejected_for_author_handoff(self) -> None:
        """Provenance required_profile binds even when the role matches."""
        lid = self._handoff_lease(924)
        result = self._call_adopt_tool(
            self.FOREIGN_AUTHOR_PROFILE,
            lease_id=lid,
            session_id="foreign-author-worker",
            worktree_path=self.wt,
        )
        self.assertFalse(result["success"])
        self.assertEqual(result["outcome"], "blocked")
        self.assertIn("wrong profile", result["reasons"][0].lower())
        self._assert_handoff_untouched(lid)
    def test_author_profile_consumes_author_handoff(self) -> None:
        lid = self._handoff_lease(925)
        result = self._call_adopt_tool(
            self.AUTHOR_PROFILE,
            lease_id=lid,
            session_id="author-worker",
            role="author",
            worktree_path=self.wt,
        )
        self.assertTrue(result["success"])
        self.assertEqual(result["outcome"], "adopted_cross_role_handoff")
        self.assertEqual(result["adopted_by_session_id"], "author-worker")
        self.assertEqual(result["adopted_from_session_id"], "ctrl-session")
        state = self.db.get_lease_workflow_state(lid)
        self.assertEqual(state["lease"]["session_id"], "author-worker")
        self.assertEqual(
            state["lease"]["adopted_by_session_id"], "author-worker"
        )
        self.assertEqual(state["assignment"]["session_id"], "author-worker")
        self.assertEqual(state["provenance"]["handoff_status"], "adopted")
    def test_author_profile_consumes_author_handoff_without_role_argument(
        self,
    ) -> None:
        lid = self._handoff_lease(926)
        result = self._call_adopt_tool(
            self.AUTHOR_PROFILE,
            lease_id=lid,
            session_id="author-worker",
            worktree_path=self.wt,
        )
        self.assertTrue(result["success"])
        self.assertEqual(result["outcome"], "adopted_cross_role_handoff")
        state = self.db.get_lease_workflow_state(lid)
        self.assertEqual(state["lease"]["session_id"], "author-worker")
        self.assertEqual(state["lease"]["role"], "author")
 if __name__ == "__main__":
    unittest.main()