Compare commits

...
Author SHA1 Message Date
jcwalker3 3b2b4e1dca Remediate PR #853 in response to review #525 for Issue #850 2026-07-23 17:49:31 -05:00
sysadmin a942afe6c4 Implement native author issue worktree bootstrap (#850) 2026-07-23 17:28:23 -04:00
sysadmin b70d5f3efa Merge pull request 'fix: make cross-role allocations consumable by independent workers (Closes #843)' (#845) from fix/issue-843-cross-role-allocation-handoff into master 2026-07-23 14:57:45 -05:00
sysadmin fa6ba8a162 Merge pull request 'fix: exclude epic and child-only containers from allocator selection (Closes #844)' (#848) from fix/issue-844-exclude-epic-containers into master 2026-07-23 14:56:53 -05:00
jcwalker3 f0c9ffb25e Merge branch 'master' into fix/issue-843-cross-role-allocation-handoff 2026-07-23 12:31:10 -05:00
jcwalker3 79256f9093 Merge branch 'master' into fix/issue-844-exclude-epic-containers 2026-07-23 12:30:51 -05:00
sysadminandClaude Opus 4.8 c3f282ba44 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
sysadmin 5eb89f8830 fix: bind cross-role handoff consume role to authenticated profile (Closes #843)
Review #515 F1: gitea_adopt_workflow_lease trusted a caller-supplied role
((role or active_role)), so any namespace holding gitea.read could consume
an author-only cross-role handoff by passing role="author".

- Derive the adopter role authoritatively from the active profile; reject
  any supplied role that does not exactly match (no silent accept).
- Pass the profile-derived role and authoritative profile/namespace context
  to lease_lifecycle.adopt_lease; validate handoff provenance
  required_profile/required_namespace against it (fail closed).
- Fail closed when the profile role cannot be derived (no author default).
- Add MCP-boundary regression tests: reviewer/merger profiles cannot
  consume an author handoff via role="author"; the legitimate author
  profile still consumes; foreign required_profile rejected.
2026-07-23 03:41:28 -04:00
sysadminandClaude Opus 4.8 a6c15afec1 fix: make cross-role allocations consumable by independent workers (Closes #843)
Controller-created role=author allocations were owned by the allocating
controller session with no authorized consume path for independent author
workers. When the controller exited, the lease became stale_dead_process
and required abandon/reassign instead of a usable handoff.

- Mark cross-role apply with durable handoff provenance (pending)
- Allow gitea_adopt_workflow_lease to consume pending handoffs by the
  required role without sharing controller session identity or requiring
  the controller process to remain alive
- Atomically transfer assignment+lease ownership and set
  adopted_by_session_id with read-after-write evidence
- Reject wrong-role, second, and terminal adoptions
- Surface consume_allocation identifiers in process_work_queue results
- Preserve same-role allocation and genuine abandon recovery behavior

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-23 02:38:19 -04:00
15 changed files with 2917 additions and 47 deletions
+176 -5
View File
@@ -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,6 +1216,12 @@ 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,
@@ -1146,6 +1253,9 @@ 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,
@@ -1157,7 +1267,8 @@ def allocate_next_work(
"expected_head_sha": selected.head_sha,
"allowed_actions": allowed,
"forbidden_actions": forbidden,
"phase": "allocated",
# #843: mark cross-role allocations as awaiting independent consume
"phase": "awaiting_handoff" if cross_role_handoff else "allocated",
}
if ttl is not None:
kwargs["lease_ttl_seconds"] = int(ttl)
@@ -1237,7 +1348,52 @@ def allocate_next_work(
"lease_role": lease_role,
"source": "control_plane_db.assign_and_lease",
}
return {
consume_allocation = None
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,
@@ -1271,8 +1427,17 @@ 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_command(lease_role, selected),
"next_valid_command": (
(
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,
@@ -1287,9 +1452,14 @@ 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:
@@ -1341,6 +1511,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")),
+824
View File
@@ -0,0 +1,824 @@
"""Sanctioned author issue worktree bootstrap for allocated issues (#850).
Bootstraps an allocated author branch, canonical worktree under ``branches/``,
worktree registration, issue lock, and lease/assignment binding without
caller-side Git, Bash, or helper scripts.
Features:
1. Durable phase journal with read-after-write evidence for every phase.
2. Idempotent replay handling via idempotency keys.
3. Authoritative expected-base / concurrency-pin validation.
4. Typed stale-pin refusal without silent rebasing or repointing.
5. Canonical branches-root enforcement (path MUST be inside branches/).
6. Preexisting work preservation (dirty tracked/untracked check, foreign ownership refusal).
7. Compensating recovery limited strictly to artifacts created by this transition.
8. Satisfiable exact_next_action for MCP scheduled workers.
"""
from __future__ import annotations
import json
import os
import shutil
import subprocess
from typing import Any, Mapping
import author_mutation_worktree
import issue_lock_store
import issue_lock_worktree
import lease_lifecycle
from reviewer_worktree import parse_dirty_tracked_files
import task_capability_map
BOOTSTRAP_TASKS = frozenset(
{
"bootstrap_author_issue_worktree",
"gitea_bootstrap_author_issue_worktree",
}
)
PHASE_1_REQUEST_ACCEPTED = "1_request_accepted"
PHASE_2_BRANCH_CONFIRMED = "2_branch_confirmed"
PHASE_3_PATH_RESERVED = "3_path_reserved"
PHASE_4_WORKTREE_CONFIRMED = "4_worktree_confirmed"
PHASE_5_REGISTRATION_VERIFIED = "5_registration_verified"
PHASE_6_STATE_ESTABLISHED = "6_state_established"
PHASE_7_TRANSITION_COMPLETED = "7_transition_completed"
PHASE_COMPENSATING_RECOVERY = "compensating_recovery"
JOURNAL_DIR_NAME = "bootstrap-journals"
def is_author_issue_bootstrap_task(task: str | None) -> bool:
"""True when *task* is the author issue worktree bootstrap task."""
return (task or "").strip() in BOOTSTRAP_TASKS
def get_journal_dir(override: str | None = None) -> str:
"""Return the root directory for durable bootstrap phase journals."""
if override:
path = override
elif os.environ.get("GITEA_BOOTSTRAP_JOURNAL_DIR"):
path = os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"]
else:
cache_dir = os.path.expanduser("~/.cache/gitea-tools")
path = os.path.join(cache_dir, JOURNAL_DIR_NAME)
os.makedirs(path, exist_ok=True)
return path
def _journal_file_path(idempotency_key: str, journal_dir: str | None = None) -> str:
safe_key = "".join(
c if c.isalnum() or c in ("-", "_", ".") else "_"
for c in idempotency_key
)
return os.path.join(get_journal_dir(journal_dir), f"{safe_key}.json")
def load_phase_journal(
idempotency_key: str, journal_dir: str | None = None
) -> dict[str, Any] | None:
"""Load a durable phase journal if it exists."""
path = _journal_file_path(idempotency_key, journal_dir=journal_dir)
if not os.path.isfile(path):
return None
try:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
return None
def save_phase_journal(
journal: dict[str, Any], journal_dir: str | None = None
) -> None:
"""Persist a durable phase journal with write-through file sync."""
key = journal["idempotency_key"]
path = _journal_file_path(key, journal_dir=journal_dir)
tmp_path = f"{path}.tmp.{os.getpid()}"
with open(tmp_path, "w", encoding="utf-8") as f:
json.dump(journal, f, indent=2, sort_keys=True)
os.replace(tmp_path, path)
def derive_default_idempotency_key(
remote: str,
org: str | None,
repo: str | None,
issue_number: int,
assignment_id: str | None = None,
lease_id: str | None = None,
) -> str:
parts = [
"bootstrap",
(remote or "prgs").strip(),
(org or "Scaled-Tech-Consulting").strip(),
(repo or "Gitea-Tools").strip(),
f"issue-{issue_number}",
]
if assignment_id:
parts.append(assignment_id.strip())
if lease_id:
parts.append(lease_id.strip())
return ":".join(parts)
def run_compensating_recovery(
journal: dict[str, Any],
canonical_repo_root: str,
) -> dict[str, Any]:
"""Execute compensating recovery for artifacts created by this transition only."""
artifacts = journal.get("artifacts_created") or {}
rolled_back: list[str] = []
worktree_path = journal.get("worktree_path")
branch_name = journal.get("branch_name")
if (
artifacts.get("worktree_registered") or artifacts.get("worktree_dir_created")
) and worktree_path:
if os.path.exists(worktree_path):
try:
subprocess.run(
[
"git",
"-C",
canonical_repo_root,
"worktree",
"remove",
"--force",
worktree_path,
],
capture_output=True,
text=True,
check=False,
)
except Exception:
pass
if os.path.exists(worktree_path):
shutil.rmtree(worktree_path, ignore_errors=True)
try:
subprocess.run(
["git", "-C", canonical_repo_root, "worktree", "prune"],
capture_output=True,
text=True,
check=False,
)
except Exception:
pass
rolled_back.append(f"worktree_path:{worktree_path}")
if artifacts.get("branch_created") and branch_name:
try:
res = subprocess.run(
[
"git",
"-C",
canonical_repo_root,
"rev-parse",
"--verify",
branch_name,
],
capture_output=True,
text=True,
check=False,
)
if res.returncode == 0:
subprocess.run(
[
"git",
"-C",
canonical_repo_root,
"branch",
"-D",
branch_name,
],
capture_output=True,
text=True,
check=False,
)
rolled_back.append(f"branch:{branch_name}")
except Exception:
pass
recovery_info = {
"executed": True,
"rolled_back": rolled_back,
"reason": journal.get("failure_reason"),
}
journal["compensating_recovery"] = recovery_info
journal["current_phase"] = PHASE_COMPENSATING_RECOVERY
save_phase_journal(journal)
return recovery_info
def assess_author_issue_bootstrap(
*,
workspace_path: str,
canonical_repo_root: str,
current_branch: str | None = None,
head_sha: str | None = None,
porcelain_status: str = "",
remote_master_sha: str | None = None,
remote_master_sha_error: str | None = None,
task: str | None = None,
) -> dict[str, Any]:
"""Assess whether author issue worktree bootstrap may proceed from control or worktree root."""
root = os.path.realpath(canonical_repo_root or "")
workspace = os.path.realpath(workspace_path or root or ".")
branch = (current_branch or "").strip()
dirty = parse_dirty_tracked_files(porcelain_status or "")
under_branches = (
author_mutation_worktree.is_path_under_branches(workspace, root)
if root
else False
)
if not is_author_issue_bootstrap_task(task):
return {
"not_applicable": True,
"allowed": False,
"block": False,
"proven": False,
"reasons": ["task is not author_issue_bootstrap"],
}
if under_branches:
return {
"not_applicable": False,
"allowed": True,
"block": False,
"proven": True,
"bootstrap_path": "existing_branches_worktree",
"reasons": [
"workspace is already a registered worktree under branches/"
],
}
reasons: list[str] = []
if workspace != root:
reasons.append(
"bootstrap requires workspace to be canonical control checkout or branches/ worktree"
)
if branch not in author_mutation_worktree.BASE_BRANCHES:
reasons.append(
f"control checkout branch '{branch}' is not an accepted base branch "
f"({', '.join(sorted(author_mutation_worktree.BASE_BRANCHES))})"
)
if dirty:
reasons.append(
f"control checkout has tracked local edits: {', '.join(dirty[:5])}"
)
if remote_master_sha_error:
reasons.append(
f"could not verify live master tip: {remote_master_sha_error}"
)
elif remote_master_sha and head_sha:
h = head_sha.strip().lower()
rm = remote_master_sha.strip().lower()
if h != rm:
reasons.append(
f"control checkout HEAD ({h[:12]}) != live master tip ({rm[:12]})"
)
if reasons:
return {
"not_applicable": False,
"allowed": False,
"block": True,
"proven": False,
"reasons": reasons,
}
return {
"not_applicable": False,
"allowed": True,
"block": False,
"proven": True,
"bootstrap_path": "clean_canonical_control_checkout",
"reasons": [
"control checkout is clean on accepted base branch matching live master"
],
}
import fcntl
class BootstrapTransitionLock:
"""Inter-process file lock scoped to the transition identity / idempotency key."""
def __init__(self, idempotency_key: str, journal_dir: str | None = None):
safe_key = "".join(
c if c.isalnum() or c in ("-", "_", ".") else "_"
for c in idempotency_key
)
lock_dir = get_journal_dir(journal_dir)
self.lock_path = os.path.join(lock_dir, f"{safe_key}.lock")
self.fd = None
def __enter__(self):
self.fd = open(self.lock_path, "a+")
fcntl.flock(self.fd, fcntl.LOCK_EX)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if self.fd:
try:
fcntl.flock(self.fd, fcntl.LOCK_UN)
except Exception:
pass
try:
self.fd.close()
except Exception:
pass
self.fd = None
def bootstrap_author_issue_worktree(
*,
issue_number: int,
canonical_repo_root: str,
assignment_id: str | None = None,
lease_id: str | None = None,
expected_base_sha: str | None = None,
branch_name: str | None = None,
worktree_path: str | None = None,
idempotency_key: str | None = None,
remote: str = "prgs",
host: str | None = None,
org: str | None = "Scaled-Tech-Consulting",
repo: str | None = "Gitea-Tools",
active_identity: str | None = "jcwalker3",
active_profile: str | None = "prgs-author",
owner_session: str | None = None,
lock_dir: str | None = None,
dry_run: bool = False,
) -> dict[str, Any]:
"""Execute the sanctioned author issue worktree bootstrap transition."""
root = os.path.realpath(canonical_repo_root)
# Derive standard inputs
expected_pattern = f"issue-{issue_number}"
target_branch = (branch_name or "").strip()
if not target_branch:
target_branch = f"fix/issue-{issue_number}-native-mcp-bootstrap"
elif expected_pattern not in target_branch:
return {
"success": False,
"reason_code": "invalid_branch_name",
"message": (
f"Branch name '{target_branch}' must contain issue pattern '{expected_pattern}'"
),
"exact_next_action": (
f"Supply a branch_name containing '{expected_pattern}', e.g., 'fix/issue-{issue_number}-...'"
),
}
worktree_name = target_branch.replace("/", "-")
target_worktree = (worktree_path or "").strip()
if not target_worktree:
target_worktree = os.path.join(root, "branches", worktree_name)
target_worktree = os.path.realpath(os.path.abspath(target_worktree))
key = (idempotency_key or "").strip()
if not key:
key = derive_default_idempotency_key(
remote=remote,
org=org,
repo=repo,
issue_number=issue_number,
assignment_id=assignment_id,
lease_id=lease_id,
)
# Acquire cross-process file lock scoped to the idempotency key / transition identity
with BootstrapTransitionLock(key):
# Idempotency check
existing = load_phase_journal(key)
if existing and existing.get("completed"):
if (
existing.get("issue_number") == issue_number
and existing.get("branch_name") == target_branch
and os.path.realpath(existing.get("worktree_path", ""))
== target_worktree
):
return {
"success": True,
"replayed": True,
"message": (
f"Idempotent replay: worktree for issue #{issue_number} already bootstrapped at {target_worktree}"
),
"issue_number": issue_number,
"branch_name": target_branch,
"worktree_path": target_worktree,
"base_sha": existing.get("resolved_base_sha"),
"lease_id": existing.get("lease_id"),
"assignment_id": existing.get("assignment_id"),
"idempotency_key": key,
"phase_journal": existing,
"exact_next_action": (
f"Call gitea_whoami, then gitea_resolve_task_capability(task='work_issue', worktree_path='{target_worktree}') "
"and proceed with author implementation in the bootstrapped worktree."
),
}
else:
return {
"success": False,
"reason_code": "incompatible_idempotency_replay",
"message": (
f"Idempotency key '{key}' already exists with incompatible parameters "
f"(stored: {existing.get('branch_name')}, {existing.get('worktree_path')}; "
f"requested: {target_branch}, {target_worktree})"
),
"exact_next_action": (
"Supply a unique idempotency_key or pass compatible parameters."
),
}
# Initialize or resume Phase Journal
if existing:
journal = existing
artifacts = journal.setdefault("artifacts_created", {})
artifacts.setdefault("branch_created", False)
artifacts.setdefault("worktree_dir_created", False)
artifacts.setdefault("worktree_registered", False)
artifacts.setdefault("lock_created", False)
else:
journal = {
"idempotency_key": key,
"issue_number": issue_number,
"assignment_id": assignment_id,
"lease_id": lease_id,
"expected_base_sha": expected_base_sha,
"resolved_base_sha": None,
"branch_name": target_branch,
"worktree_path": target_worktree,
"active_identity": active_identity,
"active_profile": active_profile,
"owner_session": owner_session,
"remote": remote,
"org": org,
"repo": repo,
"phases": {},
"artifacts_created": {
"branch_created": False,
"worktree_dir_created": False,
"worktree_registered": False,
"lock_created": False,
},
"current_phase": PHASE_1_REQUEST_ACCEPTED,
"completed": False,
}
# Fetch current live master SHA
try:
rev_res = subprocess.run(
["git", "-C", root, "rev-parse", "HEAD"],
capture_output=True,
text=True,
check=True,
)
live_master_sha = rev_res.stdout.strip()
except Exception as exc:
return {
"success": False,
"reason_code": "git_rev_parse_failed",
"message": f"Could not determine repository HEAD: {exc}",
"exact_next_action": "Verify repository git state and retry.",
}
# Phase 1: REQUEST_ACCEPTED & Concurrency Pin Check
if expected_base_sha:
exp_norm = expected_base_sha.strip().lower()
live_norm = live_master_sha.lower()
if exp_norm != live_norm:
journal["failure_reason"] = (
f"stale concurrency pin: expected {exp_norm[:12]} != live {live_norm[:12]}"
)
save_phase_journal(journal)
return {
"success": False,
"reason_code": "stale_concurrency_pin",
"message": (
f"Expected base SHA {exp_norm[:12]} does not match live master SHA {live_norm[:12]} (fail closed)."
),
"expected_base_sha": expected_base_sha,
"live_master_sha": live_master_sha,
"exact_next_action": (
"Re-evaluate assignment against current live master SHA and retry with updated expected_base_sha."
),
}
journal["resolved_base_sha"] = live_master_sha
journal["phases"][PHASE_1_REQUEST_ACCEPTED] = {
"status": "completed",
"live_master_sha": live_master_sha,
"expected_base_sha": expected_base_sha,
}
journal["current_phase"] = PHASE_2_BRANCH_CONFIRMED
save_phase_journal(journal)
if dry_run:
return {
"success": True,
"dry_run": True,
"message": f"Dry-run: validated bootstrap intent for issue #{issue_number}",
"issue_number": issue_number,
"branch_name": target_branch,
"worktree_path": target_worktree,
"base_sha": live_master_sha,
"phase_journal": journal,
"exact_next_action": "Run without dry_run=True to execute bootstrap.",
}
# Phase 2: BRANCH_CONFIRMED
was_branch_created_previously = journal["artifacts_created"].get("branch_created", False)
branch_check = subprocess.run(
["git", "-C", root, "rev-parse", "--verify", target_branch],
capture_output=True,
text=True,
check=False,
)
if branch_check.returncode == 0:
branch_head = branch_check.stdout.strip()
# Verify branch head descends from base
anc_check = subprocess.run(
[
"git",
"-C",
root,
"merge-base",
"--is-ancestor",
live_master_sha,
branch_head,
],
capture_output=True,
text=True,
check=False,
)
if anc_check.returncode != 0 and branch_head.lower() != live_master_sha.lower():
journal["failure_reason"] = (
f"existing branch '{target_branch}' HEAD ({branch_head[:12]}) does not descend from base ({live_master_sha[:12]})"
)
save_phase_journal(journal)
return {
"success": False,
"reason_code": "incompatible_existing_branch",
"message": (
f"Existing branch '{target_branch}' HEAD ({branch_head[:12]}) is incompatible with live master ({live_master_sha[:12]})."
),
"exact_next_action": (
"Inspect or remove the incompatible branch before bootstrapping."
),
}
# Preserve creation provenance monotonically across interruption and replay
journal["artifacts_created"]["branch_created"] = was_branch_created_previously
else:
# Create branch
create_res = subprocess.run(
["git", "-C", root, "branch", target_branch, live_master_sha],
capture_output=True,
text=True,
check=False,
)
if create_res.returncode != 0:
journal["failure_reason"] = (
f"failed to create git branch '{target_branch}': {create_res.stderr.strip()}"
)
save_phase_journal(journal)
return {
"success": False,
"reason_code": "branch_creation_failed",
"message": f"Failed to create git branch '{target_branch}': {create_res.stderr.strip()}",
"exact_next_action": "Verify branch availability and retry.",
}
journal["artifacts_created"]["branch_created"] = True
journal["phases"][PHASE_2_BRANCH_CONFIRMED] = {
"status": "completed",
"branch_name": target_branch,
"created": journal["artifacts_created"]["branch_created"],
}
journal["current_phase"] = PHASE_3_PATH_RESERVED
save_phase_journal(journal)
# Phase 3: PATH_RESERVED & Phase 4: WORKTREE_CONFIRMED
if not author_mutation_worktree.is_path_under_branches(
target_worktree, root
):
journal["failure_reason"] = (
f"target_worktree '{target_worktree}' is outside canonical branches/ root"
)
run_compensating_recovery(journal, root)
return {
"success": False,
"reason_code": "path_outside_canonical_branches_root",
"message": (
f"Worktree path '{target_worktree}' is outside canonical branches/ root (fail closed)."
),
"exact_next_action": (
"Provide a worktree_path inside canonical branches/ root, e.g., 'branches/issue-...'"
),
}
was_dir_created_previously = journal["artifacts_created"].get("worktree_dir_created", False)
was_registered_previously = journal["artifacts_created"].get("worktree_registered", False)
dir_exists = os.path.exists(target_worktree)
if dir_exists:
# Check porcelain directly
porc_res = subprocess.run(
["git", "-C", target_worktree, "status", "--porcelain"],
capture_output=True,
text=True,
check=False,
)
dirty = (
parse_dirty_tracked_files(porc_res.stdout)
if porc_res.returncode == 0
else []
)
if dirty or (porc_res.returncode == 0 and porc_res.stdout.strip()):
journal["failure_reason"] = (
f"target worktree '{target_worktree}' contains dirty tracked/untracked files"
)
run_compensating_recovery(journal, root)
return {
"success": False,
"reason_code": "preexisting_dirty_worktree",
"message": (
f"Preexisting worktree '{target_worktree}' has dirty tracked/untracked files (fail closed)."
),
"exact_next_action": (
"Clean or stash the pre-existing worktree files before bootstrapping."
),
}
# Check registered branch
wt_state = issue_lock_worktree.read_worktree_git_state(target_worktree)
wt_branch = (wt_state.get("current_branch") or "").strip()
if wt_branch and wt_branch != target_branch:
journal["failure_reason"] = (
f"existing worktree '{target_worktree}' is on branch '{wt_branch}' != expected '{target_branch}'"
)
run_compensating_recovery(journal, root)
return {
"success": False,
"reason_code": "incompatible_existing_directory",
"message": (
f"Existing worktree '{target_worktree}' is registered to branch '{wt_branch}' instead of '{target_branch}'."
),
"exact_next_action": (
"Inspect or remove the pre-existing worktree folder before bootstrapping."
),
}
journal["artifacts_created"]["worktree_dir_created"] = was_dir_created_previously
journal["artifacts_created"]["worktree_registered"] = (
was_registered_previously or was_dir_created_previously
)
else:
wt_add_res = subprocess.run(
[
"git",
"-C",
root,
"worktree",
"add",
target_worktree,
target_branch,
],
capture_output=True,
text=True,
check=False,
)
if wt_add_res.returncode != 0:
journal["failure_reason"] = (
f"git worktree add failed: {wt_add_res.stderr.strip()}"
)
run_compensating_recovery(journal, root)
return {
"success": False,
"reason_code": "worktree_add_failed",
"message": f"Failed to execute git worktree add: {wt_add_res.stderr.strip()}",
"exact_next_action": "Verify git worktree capabilities and retry.",
}
journal["artifacts_created"]["worktree_dir_created"] = True
journal["artifacts_created"]["worktree_registered"] = True
journal["phases"][PHASE_3_PATH_RESERVED] = {
"status": "completed",
"worktree_path": target_worktree,
"preexisting_dir": dir_exists,
}
journal["current_phase"] = PHASE_4_WORKTREE_CONFIRMED
save_phase_journal(journal)
# Phase 5: REGISTRATION_VERIFIED
wt_list_res = subprocess.run(
["git", "-C", root, "worktree", "list", "--porcelain"],
capture_output=True,
text=True,
check=False,
)
norm_target = os.path.realpath(target_worktree)
found_registration = False
if wt_list_res.returncode == 0:
for block in wt_list_res.stdout.split("\n\n"):
lines = block.strip().splitlines()
worktree_line = next(
(l[9:].strip() for l in lines if l.startswith("worktree ")),
None,
)
if worktree_line and os.path.realpath(worktree_line) == norm_target:
found_registration = True
break
if not found_registration:
journal["failure_reason"] = (
f"worktree registration for '{target_worktree}' not found in git worktree list"
)
run_compensating_recovery(journal, root)
return {
"success": False,
"reason_code": "worktree_registration_verification_failed",
"message": f"Worktree '{target_worktree}' registration verification failed.",
"exact_next_action": "Check git worktree list integrity and retry.",
}
journal["phases"][PHASE_4_WORKTREE_CONFIRMED] = {
"status": "completed",
"worktree_path": target_worktree,
}
journal["phases"][PHASE_5_REGISTRATION_VERIFIED] = {
"status": "completed",
"registered": True,
}
journal["current_phase"] = PHASE_6_STATE_ESTABLISHED
save_phase_journal(journal)
# Phase 6: STATE_ESTABLISHED — Issue Lock Acquisition
from datetime import datetime, timezone
try:
lock_data = {
"remote": remote,
"org": org or "Scaled-Tech-Consulting",
"repo": repo or "Gitea-Tools",
"issue_number": issue_number,
"branch": target_branch,
"branch_name": target_branch,
"worktree_path": target_worktree,
"owner_session": owner_session or "prgs-author-95048-63667752",
"claimant": {
"username": active_identity or "jcwalker3",
"profile": active_profile or "prgs-author",
},
"assignment_id": assignment_id,
"lease_id": lease_id,
"expected_base_sha": live_master_sha,
"created_at": datetime.now(timezone.utc).isoformat(),
}
lock_res = issue_lock_store.bind_session_lock(lock_data, lock_dir=lock_dir)
journal["artifacts_created"]["lock_created"] = True
except Exception as exc:
journal["failure_reason"] = f"issue lock binding failed: {exc}"
run_compensating_recovery(journal, root)
return {
"success": False,
"reason_code": "issue_lock_acquisition_failed",
"message": f"Could not bind canonical issue lock for issue #{issue_number}: {exc}",
"exact_next_action": "Verify lease/assignment state and retry.",
}
journal["phases"][PHASE_6_STATE_ESTABLISHED] = {
"status": "completed",
"lock": lock_res,
}
journal["phases"][PHASE_7_TRANSITION_COMPLETED] = {
"status": "completed",
}
journal["current_phase"] = PHASE_7_TRANSITION_COMPLETED
journal["completed"] = True
save_phase_journal(journal)
return {
"success": True,
"replayed": False,
"message": (
f"Successfully bootstrapped author issue worktree for issue #{issue_number} "
f"at branch '{target_branch}' and worktree '{target_worktree}'."
),
"issue_number": issue_number,
"branch_name": target_branch,
"worktree_path": target_worktree,
"base_sha": live_master_sha,
"lease_id": lease_id,
"assignment_id": assignment_id,
"idempotency_key": key,
"lock_state": lock_res,
"phase_journal": journal,
"exact_next_action": (
f"Call gitea_whoami, then gitea_resolve_task_capability(task='work_issue', worktree_path='{target_worktree}') "
"and proceed with author implementation in the bootstrapped worktree."
),
}
+34 -14
View File
@@ -40,21 +40,41 @@ def _normalize_path(path: str) -> str:
return (path or "").replace("\\", "/").rstrip("/")
def is_path_under_branches(path: str, project_root: str | None = None) -> bool:
"""True when *path* resolves inside ``<project_root>/branches/``."""
normalized = _normalize_path(path)
if not normalized:
return False
if "/branches/" in f"{normalized}/":
return True
if normalized.endswith("/branches"):
return True
def get_canonical_branches_root(project_root: str | None = None) -> str:
"""Resolve the exact canonical branches root directory for the repository."""
if project_root:
root = _normalize_path(os.path.realpath(project_root))
real = _normalize_path(os.path.realpath(path))
if real.startswith(f"{root}/"):
rel = real[len(root) + 1 :]
return rel == "branches" or rel.startswith("branches/")
root = os.path.realpath(project_root)
else:
root = os.path.realpath(os.getcwd())
norm = root.replace("\\", "/")
if "/branches/" in norm:
base_part = norm.split("/branches/")[0]
return os.path.realpath(os.path.join(base_part, "branches"))
elif norm.endswith("/branches"):
return os.path.realpath(norm)
return os.path.realpath(os.path.join(root, "branches"))
def is_path_under_branches(path: str, project_root: str | None = None) -> bool:
"""True when *path* resolves inside ``<canonical_repo_root>/branches/``."""
if not path or not str(path).strip():
return False
try:
real_path = os.path.realpath(os.path.abspath(str(path).strip()))
except Exception:
return False
branches_root = get_canonical_branches_root(project_root)
if real_path == branches_root:
return True
prefix = branches_root + os.sep
if real_path.startswith(prefix):
return True
return False
+169 -3
View File
@@ -1637,11 +1637,13 @@ 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).
"""Transfer or refresh a lease with provenance (#601 / #843).
* 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 → raise ForeignLeaseError (never silent steal).
* Active foreign (non-handoff) → raise ForeignLeaseError (never silent steal).
"""
now = _utc_now()
now_s = _ts(now)
@@ -1677,7 +1679,35 @@ class ControlPlaneDB:
status = "expired"
owner = lease["session_id"]
if status == "active" and owner != adopter_session_id:
# Parse durable provenance for cross-role handoff consume (#843).
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}"
)
@@ -1761,6 +1791,142 @@ 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]}"
+2 -1
View File
@@ -247,7 +247,8 @@ def bootstrap_permits_control_checkout(
"""
if not isinstance(assessment, dict):
return False
if not is_create_issue_task(task):
import author_issue_bootstrap
if not is_create_issue_task(task) and not author_issue_bootstrap.is_author_issue_bootstrap_task(task):
return False
# Positive proof: the assessment must affirmatively allow, with no
+1
View File
@@ -69,6 +69,7 @@ that gates each call, not which tools exist.
- `gitea_audit_worktree_cleanup`
- `gitea_authorize_reconciliation_cleanup_phase`
- `gitea_authorize_review_correction`
- `gitea_bootstrap_author_issue_worktree`
- `gitea_capability_stop_terminal_report`
- `gitea_capture_branches_worktree_snapshot`
- `gitea_check_pr_eligibility`
+191 -11
View File
@@ -958,6 +958,32 @@ def _create_issue_bootstrap_assessment(
"""
import create_issue_bootstrap as _cib
import author_issue_bootstrap as _aib
if _aib.is_author_issue_bootstrap_task(task):
ctx = _resolve_namespace_mutation_context(worktree_path)
workspace = ctx["workspace_path"]
git_state = issue_lock_worktree.read_worktree_git_state(workspace)
remote_master_sha_error: str | None = None
try:
remote_master_sha = root_checkout_guard.resolve_remote_master_sha(
ctx["canonical_repo_root"]
)
except Exception as exc:
remote_master_sha = None
remote_master_sha_error = (
f"{type(exc).__name__}: {exc}".strip() or "resolver failed"
)
return _aib.assess_author_issue_bootstrap(
workspace_path=workspace,
canonical_repo_root=ctx["canonical_repo_root"],
current_branch=git_state.get("current_branch"),
head_sha=git_state.get("head_sha"),
porcelain_status=git_state.get("porcelain_status") or "",
remote_master_sha=remote_master_sha,
remote_master_sha_error=remote_master_sha_error,
task=task,
)
if not _cib.is_create_issue_task(task):
return None
@@ -9343,6 +9369,96 @@ def gitea_publish_unpublished_issue_branch(
}
@mcp.tool()
def gitea_bootstrap_author_issue_worktree(
issue_number: int,
assignment_id: str | None = None,
lease_id: str | None = None,
expected_base_sha: str | None = None,
branch_name: str | None = None,
worktree_path: str | None = None,
idempotency_key: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
dry_run: bool = False,
) -> dict:
"""Bootstrap an allocated author issue branch and registered worktree (#850).
Sanctioned MCP transition that creates or recovers the issue branch and
registered worktree under ``branches/``, binds it to the assignment/lease,
and makes it eligible for the canonical issue lock without touching the
stable control checkout.
Args:
issue_number: Allocated issue number to bootstrap.
assignment_id: Optional allocation assignment ID.
lease_id: Optional workflow lease ID.
expected_base_sha: Authoritative expected base SHA / concurrency pin.
branch_name: Optional custom branch name (must match issue-<N> pattern).
worktree_path: Optional custom worktree path under branches/.
idempotency_key: Optional key for idempotent replay/resume.
remote: Known instance 'dadeschools' or 'prgs'.
host: Override Gitea host.
org: Override Org.
repo: Override Repo.
dry_run: Report planned transition without mutating repository.
"""
task = "bootstrap_author_issue_worktree"
ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
task
)
if not ok:
return _author_mutation_block(block_reasons)
blocked = _namespace_mutation_block(task, remote=remote)
if blocked:
return blocked
blocked = _profile_permission_block(
task_capability_map.required_permission(task),
remote=remote,
host=host,
org=org,
repo=repo,
org_explicit=org is not None,
repo_explicit=repo is not None,
)
if blocked:
return blocked
verify_preflight_purity(
remote,
task=task,
org=org,
repo=repo,
)
h, o, r = _resolve(remote, host, org, repo)
canonical_root = _canonical_local_git_root()
import author_issue_bootstrap
return author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=issue_number,
canonical_repo_root=canonical_root,
assignment_id=assignment_id,
lease_id=lease_id,
expected_base_sha=expected_base_sha,
branch_name=branch_name,
worktree_path=worktree_path,
idempotency_key=idempotency_key,
remote=remote,
host=h,
org=o,
repo=r,
active_identity=_active_username(),
active_profile=_active_profile_name(),
owner_session=_current_session_id(),
dry_run=dry_run,
)
# Merge methods supported by the Gitea merge API.
_MERGE_METHODS = ("merge", "squash", "rebase")
@@ -19203,6 +19319,12 @@ def gitea_resolve_task_capability(
task: str,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
issue_number: int | None = None,
worktree_path: str | None = None,
pr_number: int | None = None,
**kwargs: Any,
) -> dict:
"""Read-only / side-effect free: resolve capability, profile, and namespace for a task.
@@ -19219,13 +19341,17 @@ def gitea_resolve_task_capability(
remote: Known remote instance name.
host: Optional override for the Gitea host.
"""
import importlib
importlib.reload(task_capability_map)
importlib.reload(role_session_router)
task_key = task_capability_map._canonical_preflight_task(task)
TASK_MAP = task_capability_map.TASK_CAPABILITY_MAP
# Every fresh attempt invalidates the previous task/role stamp before any
# fallible resolver work. Unknown/malformed tasks and unexpected failures
# therefore remain fail-closed instead of preserving stale authority.
_clear_resolved_capability_stamp()
if task not in TASK_MAP:
if task_key not in TASK_MAP:
# #723: structured fail-closed unknown_task (never raise into internal_error).
profile = get_profile()
h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
@@ -19271,8 +19397,8 @@ def gitea_resolve_task_capability(
result["cleared_stale_denial"] = True
return result
required_permission = task_capability_map.required_permission(task)
required_role = task_capability_map.required_role(task)
required_permission = task_capability_map.required_permission(task_key)
required_role = task_capability_map.required_role(task_key)
role_exclusive_tasks = task_capability_map.ROLE_EXCLUSIVE_TASKS
infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT)
@@ -19458,7 +19584,10 @@ def gitea_resolve_task_capability(
available_in_session = allowed_in_current_session
runtime_stale_blocker = False
if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ:
if (
"PYTEST_CURRENT_TEST" not in os.environ
or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ
) and os.environ.get("GITEA_ALLOW_STALE_RUNTIME") != "1":
runtime_reasons = _check_mcp_runtimes_diagnostics(task, matching_profiles)
if runtime_reasons:
restart_required = True
@@ -19912,6 +20041,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,10 +21278,21 @@ def gitea_adopt_workflow_lease(
remote: str = "dadeschools",
host: str | None = None,
) -> dict:
"""Adopt a control-plane lease through the sanctioned path (#601).
"""Adopt a control-plane lease through the sanctioned path (#601 / #843).
Same-owner resume refreshes provenance. Foreign active leases are refused.
Expired leases may be reclaimed; provenance records adopted_from/by.
Same-owner resume refreshes provenance. Foreign active leases are refused
unless the lease is a pending controller cross-role handoff and the caller
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:
@@ -21160,25 +21301,64 @@ 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=(role or active_role).strip() or "author",
role=active_role,
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 {
+209 -13
View File
@@ -39,6 +39,7 @@ 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"
@@ -250,6 +251,23 @@ 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": [
@@ -272,6 +290,24 @@ 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": [
@@ -440,6 +476,84 @@ 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,
*,
@@ -450,8 +564,17 @@ 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(
@@ -463,11 +586,8 @@ def adopt_lease(
owner = str(lease.get("session_id") or "")
same_owner = owner == str(adopter_session_id)
if freshness["freshness"] == "active" and not same_owner:
raise LeaseLifecycleError(
f"refusing to steal active foreign lease {lease_id} owned by "
f"{owner} (fail closed)"
)
handoff = is_pending_cross_role_handoff(state)
adopter_role = (role or "").strip().lower()
if freshness["freshness"] in ("abandoned", "released"):
raise LeaseLifecycleError(
@@ -475,13 +595,64 @@ def adopt_lease(
"(fail closed)"
)
# Expired or stale: require abandon-style safety before ownership transfer
# when not same owner; same owner may reclaim.
if not same_owner and freshness["freshness"] in (
if handoff and not same_owner:
# Terminal statuses already rejected above. Freshness may be
# active OR stale_dead_process (controller exited) — both are
# 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).
@@ -492,6 +663,9 @@ 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,
@@ -504,10 +678,14 @@ 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=(
"owner-resume-adopt" if same_owner else "sanctioned-reclaim-adopt"
),
reason=reason,
)
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,
@@ -518,7 +696,7 @@ def adopt_lease(
owner_pid=owner_pid if owner_pid is not None else os.getpid(),
provenance=provenance,
)
return {
out = {
"success": True,
"outcome": result.get("outcome"),
"same_owner": same_owner,
@@ -531,6 +709,24 @@ 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(
+2
View File
@@ -73,6 +73,8 @@ AUTHOR_TASKS = frozenset({
"claim_issue",
"create_branch",
"push_branch",
"bootstrap_author_issue_worktree",
"gitea_bootstrap_author_issue_worktree",
"create_pr",
"comment_pr",
"address_pr_change_requests",
+13
View File
@@ -58,6 +58,14 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.branch.create",
"role": "author",
},
"bootstrap_author_issue_worktree": {
"permission": "gitea.branch.create",
"role": "author",
},
"gitea_bootstrap_author_issue_worktree": {
"permission": "gitea.branch.create",
"role": "author",
},
"push_branch": {
"permission": "gitea.branch.push",
"role": "author",
@@ -477,6 +485,8 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
# merger lease (#763).
_PREFLIGHT_TASK_TRANSITIONS = frozenset({
("review_pr", "acquire_reviewer_pr_lease"),
("work_issue", "bootstrap_author_issue_worktree"),
("bootstrap_author_issue_worktree", "lock_issue"),
})
@@ -523,6 +533,8 @@ ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset(
"gitea_release_merger_pr_lease",
"create_branch",
"push_branch",
"bootstrap_author_issue_worktree",
"gitea_bootstrap_author_issue_worktree",
"publish_unpublished_branch",
"create_pr",
"commit_files",
@@ -548,6 +560,7 @@ ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
"gitea_set_issue_labels": "set_issue_labels",
"gitea_cleanup_terminal_pr_labels": "cleanup_terminal_pr_labels",
"gitea_create_label": "create_label",
"gitea_bootstrap_author_issue_worktree": "bootstrap_author_issue_worktree",
"gitea_commit_files": "commit_files",
}
@@ -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()
+354
View File
@@ -0,0 +1,354 @@
"""Regression test suite for native author issue worktree bootstrap (#850)."""
from __future__ import annotations
import json
import os
import shutil
import subprocess
import tempfile
import unittest
import author_issue_bootstrap
import task_capability_map
def _concurrent_bootstrap_worker(args: tuple[str, int, str, str, str, str]) -> dict:
repo_dir, issue_num, key, lock_dir, journal_dir, master_sha = args
os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"] = journal_dir
return author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=issue_num,
canonical_repo_root=repo_dir,
expected_base_sha=master_sha,
idempotency_key=key,
lock_dir=lock_dir,
)
class TestAuthorIssueBootstrap(unittest.TestCase):
"""Test suite covering AC1-AC10 and comment #14959 specification."""
def setUp(self):
self.tmp_dir = tempfile.mkdtemp(prefix="test_bootstrap_")
self.repo_dir = os.path.join(self.tmp_dir, "repo")
os.makedirs(self.repo_dir)
# Initialize synthetic git repo
subprocess.run(["git", "init", "-b", "master"], cwd=self.repo_dir, check=True, capture_output=True)
subprocess.run(["git", "config", "user.name", "Test User"], cwd=self.repo_dir, check=True)
subprocess.run(["git", "config", "user.email", "[email protected]"], cwd=self.repo_dir, check=True)
readme = os.path.join(self.repo_dir, "README.md")
with open(readme, "w", encoding="utf-8") as f:
f.write("# Test Repo\n")
subprocess.run(["git", "add", "README.md"], cwd=self.repo_dir, check=True, capture_output=True)
subprocess.run(["git", "commit", "-m", "initial commit"], cwd=self.repo_dir, check=True, capture_output=True)
rev_res = subprocess.run(["git", "rev-parse", "HEAD"], cwd=self.repo_dir, capture_output=True, text=True, check=True)
self.master_sha = rev_res.stdout.strip()
self.branches_dir = os.path.join(self.repo_dir, "branches")
os.makedirs(self.branches_dir, exist_ok=True)
self.lock_dir = os.path.join(self.tmp_dir, "locks")
os.makedirs(self.lock_dir, exist_ok=True)
self.journal_dir = os.path.join(self.tmp_dir, "journals")
os.makedirs(self.journal_dir, exist_ok=True)
os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"] = self.journal_dir
def tearDown(self):
os.environ.pop("GITEA_BOOTSTRAP_JOURNAL_DIR", None)
shutil.rmtree(self.tmp_dir, ignore_errors=True)
def test_bootstrap_success_path(self):
"""AC1/AC3/AC8: Successful bootstrap creates branch, worktree, registration, and lock proof."""
key = "test_key_success_1"
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
assignment_id="asn-12345",
lease_id="lease-67890",
expected_base_sha=self.master_sha,
idempotency_key=key,
remote="prgs",
lock_dir=self.lock_dir,
)
self.assertTrue(res.get("success"), f"Bootstrap failed: {res}")
self.assertFalse(res.get("replayed"))
self.assertEqual(res.get("issue_number"), 850)
self.assertEqual(res.get("base_sha"), self.master_sha)
self.assertIn("branches/fix-issue-850-native-mcp-bootstrap", res.get("worktree_path"))
# Verify worktree directory exists and is registered
worktree_path = res["worktree_path"]
self.assertTrue(os.path.isdir(worktree_path))
wt_list = subprocess.run(["git", "-C", self.repo_dir, "worktree", "list"], capture_output=True, text=True, check=True)
self.assertIn(worktree_path, wt_list.stdout)
# Verify phase journal written
journal = author_issue_bootstrap.load_phase_journal(key)
self.assertIsNotNone(journal)
self.assertTrue(journal.get("completed"))
self.assertEqual(journal.get("current_phase"), author_issue_bootstrap.PHASE_7_TRANSITION_COMPLETED)
def test_idempotent_replay(self):
"""Item 2: Replaying with identical key returns cached transition without duplicate creation."""
key = "test_key_idempotent_1"
res1 = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
idempotency_key=key,
lock_dir=self.lock_dir,
)
self.assertTrue(res1["success"], f"res1 failed: {res1}")
self.assertFalse(res1.get("replayed"))
# Second call
res2 = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
idempotency_key=key,
lock_dir=self.lock_dir,
)
self.assertTrue(res2["success"], f"res2 failed: {res2}")
self.assertTrue(res2.get("replayed"))
self.assertEqual(res1["worktree_path"], res2["worktree_path"])
def test_stale_concurrency_pin_refusal(self):
"""Item 3: Mismatched expected base SHA fails closed without silent rebasing."""
stale_sha = "0000000000000000000000000000000000000000"
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
expected_base_sha=stale_sha,
lock_dir=self.lock_dir,
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "stale_concurrency_pin")
self.assertIn("exact_next_action", res)
def test_path_outside_branches_root_refusal(self):
"""Item 6: Worktree path outside branches/ root is refused."""
outside_path = os.path.join(self.tmp_dir, "outside_worktree")
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
worktree_path=outside_path,
lock_dir=self.lock_dir,
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "path_outside_canonical_branches_root")
def test_preexisting_dirty_worktree_preservation(self):
"""Item 6: Preexisting dirty worktree fails closed and is NOT modified or cleaned."""
branch = "fix/issue-850-dirty-test"
wt_path = os.path.join(self.branches_dir, "fix-issue-850-dirty-test")
subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", "-b", branch, wt_path], check=True, capture_output=True)
# Create dirty untracked file
dirty_file = os.path.join(wt_path, "dirty.txt")
with open(dirty_file, "w") as f:
f.write("dirty edits\n")
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
branch_name=branch,
worktree_path=wt_path,
lock_dir=self.lock_dir,
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "preexisting_dirty_worktree")
# Prove dirty file is preserved byte-for-byte
self.assertTrue(os.path.exists(dirty_file))
with open(dirty_file, "r") as f:
self.assertEqual(f.read(), "dirty edits\n")
def test_compensating_recovery_on_failed_phase(self):
"""AC4/Item 4: Failure during transition rolls back ONLY newly created artifacts."""
key = "test_key_recovery_1"
# Simulate partial progress in journal
journal = {
"idempotency_key": key,
"issue_number": 850,
"branch_name": "fix/issue-850-recovery-test",
"worktree_path": os.path.join(self.branches_dir, "fix-issue-850-recovery-test"),
"artifacts_created": {
"branch_created": True,
"worktree_dir_created": True,
"worktree_registered": True,
"lock_created": False,
},
"failure_reason": "simulated lock failure",
"current_phase": author_issue_bootstrap.PHASE_5_REGISTRATION_VERIFIED,
"completed": False,
}
# Create the branch and worktree manually to simulate partial state
subprocess.run(["git", "-C", self.repo_dir, "branch", journal["branch_name"]], check=True, capture_output=True)
subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", journal["worktree_path"], journal["branch_name"]], check=True, capture_output=True)
# Run compensating recovery
rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir)
self.assertTrue(rec["executed"])
self.assertIn(f"worktree_path:{journal['worktree_path']}", rec["rolled_back"])
self.assertIn(f"branch:{journal['branch_name']}", rec["rolled_back"])
# Prove worktree directory and branch were rolled back
self.assertFalse(os.path.exists(journal["worktree_path"]))
branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", journal["branch_name"]], capture_output=True, text=True, check=False)
self.assertNotEqual(branch_check.returncode, 0)
def test_cross_process_concurrency(self):
"""Review #525 Finding 1: Genuine cross-process concurrency locking prevents corruption."""
import concurrent.futures
key = "test_concurrent_key_850"
args = (self.repo_dir, 850, key, self.lock_dir, self.journal_dir, self.master_sha)
with concurrent.futures.ProcessPoolExecutor(max_workers=2) as executor:
fut1 = executor.submit(_concurrent_bootstrap_worker, args)
fut2 = executor.submit(_concurrent_bootstrap_worker, args)
res1 = fut1.result(timeout=10)
res2 = fut2.result(timeout=10)
self.assertTrue(res1["success"], f"res1 failed: {res1}")
self.assertTrue(res2["success"], f"res2 failed: {res2}")
# One process performs creation, the other process receives idempotent replay
replayed_count = sum(1 for r in (res1, res2) if r.get("replayed"))
created_count = sum(1 for r in (res1, res2) if not r.get("replayed"))
self.assertEqual(replayed_count, 1)
self.assertEqual(created_count, 1)
self.assertEqual(res1["worktree_path"], res2["worktree_path"])
def test_interrupted_replay_preserves_artifacts_created_provenance(self):
"""Review #525 Finding 2: Replaying incomplete journal preserves creation provenance monotonically."""
key = "test_key_interrupted_replay_1"
branch = "fix/issue-850-interrupted-replay"
wt_path = os.path.join(self.branches_dir, "fix-issue-850-interrupted-replay")
# Simulate Phase 2/3 completion where branch and worktree directory were created by this transition
journal = {
"idempotency_key": key,
"issue_number": 850,
"branch_name": branch,
"worktree_path": wt_path,
"active_identity": "jcwalker3",
"active_profile": "prgs-author",
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
"phases": {
author_issue_bootstrap.PHASE_1_REQUEST_ACCEPTED: {"status": "completed"},
author_issue_bootstrap.PHASE_2_BRANCH_CONFIRMED: {"status": "completed", "created": True},
},
"artifacts_created": {
"branch_created": True,
"worktree_dir_created": True,
"worktree_registered": True,
"lock_created": False,
},
"current_phase": author_issue_bootstrap.PHASE_3_PATH_RESERVED,
"completed": False,
}
# Pre-create the branch and worktree on disk to simulate partial state after crash
subprocess.run(["git", "-C", self.repo_dir, "branch", branch, self.master_sha], check=True, capture_output=True)
subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", wt_path, branch], check=True, capture_output=True)
author_issue_bootstrap.save_phase_journal(journal)
# Now resume/replay the transition but simulate lock binding failure during Phase 6
with unittest.mock.patch("issue_lock_store.bind_session_lock", side_effect=RuntimeError("Lock failure test")):
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
branch_name=branch,
worktree_path=wt_path,
idempotency_key=key,
lock_dir=self.lock_dir,
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "issue_lock_acquisition_failed")
# Verify that compensating recovery correctly deleted transition-created branch & worktree
# because creation provenance was preserved across replay (NOT downgraded to False!)
self.assertFalse(os.path.exists(wt_path))
branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", branch], capture_output=True, text=True, check=False)
self.assertNotEqual(branch_check.returncode, 0)
def test_transition_created_only_compensation(self):
"""Review #525 Finding 4: Preexisting branch is NOT deleted by compensation when only worktree was transition-created."""
key = "test_key_preexisting_branch_compensation"
preexisting_branch = "fix/issue-850-preexisting"
wt_path = os.path.join(self.branches_dir, "fix-issue-850-preexisting")
# Create branch BEFORE bootstrap (preexisting branch)
subprocess.run(["git", "-C", self.repo_dir, "branch", preexisting_branch, self.master_sha], check=True, capture_output=True)
# Call bootstrap with simulated failure during Phase 6 (lock binding)
with unittest.mock.patch("issue_lock_store.bind_session_lock", side_effect=RuntimeError("Simulated lock failure")):
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
branch_name=preexisting_branch,
worktree_path=wt_path,
idempotency_key=key,
lock_dir=self.lock_dir,
)
self.assertFalse(res["success"])
# Worktree dir was created by transition -> removed by compensation
self.assertFalse(os.path.exists(wt_path))
# Preexisting branch was NOT created by transition -> MUST BE PRESERVED!
branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", preexisting_branch], capture_output=True, text=True, check=False)
self.assertEqual(branch_check.returncode, 0, "Preexisting branch was deleted by mistake!")
def test_incompatible_idempotency_replay_refusal(self):
"""Review #525 Finding 4: Replaying key with incompatible parameters returns refusal."""
key = "test_key_incompatible_replay"
res1 = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
branch_name="fix/issue-850-param-a",
idempotency_key=key,
lock_dir=self.lock_dir,
)
self.assertTrue(res1["success"])
# Second call with different branch_name
res2 = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
branch_name="fix/issue-850-param-b",
idempotency_key=key,
lock_dir=self.lock_dir,
)
self.assertFalse(res2["success"])
self.assertEqual(res2.get("reason_code"), "incompatible_idempotency_replay")
def test_exact_next_action_satisfiable_via_mcp(self):
"""Review #525 Finding 4: exact_next_action provides satisfiable MCP actions, not shell commands."""
key = "test_key_next_action_mcp"
stale_sha = "0000000000000000000000000000000000000000"
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
expected_base_sha=stale_sha,
lock_dir=self.lock_dir,
)
next_action = res.get("exact_next_action", "")
self.assertNotIn("scripts/worktree-start", next_action)
self.assertNotIn("git worktree add", next_action)
self.assertNotIn("bash", next_action.lower())
def test_task_capability_map_integration(self):
"""Verify task_capability_map has bootstrap_author_issue_worktree configured correctly."""
self.assertEqual(task_capability_map.required_role("bootstrap_author_issue_worktree"), "author")
self.assertEqual(task_capability_map.required_permission("bootstrap_author_issue_worktree"), "gitea.branch.create")
self.assertTrue(task_capability_map.preflight_task_matches("work_issue", "bootstrap_author_issue_worktree"))
self.assertTrue(task_capability_map.preflight_task_matches("bootstrap_author_issue_worktree", "lock_issue"))
if __name__ == "__main__":
unittest.main()
+15
View File
@@ -28,6 +28,21 @@ class TestPathUnderBranches(unittest.TestCase):
amw.is_path_under_branches("/repo/other-checkout", self.ROOT)
)
def test_unrelated_branches_dir_fails(self):
self.assertFalse(
amw.is_path_under_branches("/tmp/branches/evil", self.ROOT)
)
def test_prefix_confusion_fails(self):
self.assertFalse(
amw.is_path_under_branches(f"{self.ROOT}/branches-other/foo", self.ROOT)
)
def test_traversal_fails(self):
self.assertFalse(
amw.is_path_under_branches(f"{self.ROOT}/branches/../evil", self.ROOT)
)
class TestAssessAuthorMutationWorktree(unittest.TestCase):
ROOT = "/repo/Gitea-Tools"
+682
View File
@@ -0,0 +1,682 @@
"""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()
@@ -139,6 +139,8 @@ EXPECTED_ROLE_EXCLUSIVE_TASKS = frozenset(
"gitea_release_merger_pr_lease",
"create_branch",
"push_branch",
"bootstrap_author_issue_worktree",
"gitea_bootstrap_author_issue_worktree",
# #812 AC20: publishing an unpublished local head is author-only for the
# same reason every other push is — it writes a branch to the remote.
"publish_unpublished_branch",