Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 101 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

fix(author bootstrap): resolve allocator session ownership, authority alignment, and test coverage (#943)

Browse Source
This commit is contained in:
sysadmin
2026-07-27 19:52:27 -04:00
parent f49e781102
commit 47bfae07d2
3 changed files with 956 additions and 339 deletions
Download Patch File Download Diff File Expand all files Collapse all files
author_issue_bootstrap.py
+52
View File
@@ -196,6 +196,58 @@ def _verify_assignment_and_lease_ids(
# Some lease rows may not yet have an assignment join; still require
# the lease itself to exist and bind to the claimed session/issue.
pass
# #943 review 622 B2: a lease that is no longer live confers no ownership.
# Existence alone previously satisfied this gate, so a released or expired
# lease could still authorize a bootstrap for a claim its session had given
# up. Checked before the session comparison so the reason names the real
# problem rather than reporting a mismatch.
from datetime import datetime, timezone
lease_status = str(lease.get("status") or "").strip().lower()
if lease_status and lease_status != "active":
return {
"success": False,
"reason_code": "lease_not_live",
"message": (
f"lease_id '{lid}' is '{lease_status}', not active; a lease that "
"is not live confers no ownership (fail closed)."
),
"exact_next_action": (
"Re-allocate the work item and pass the live assignment/lease pair."
),
}
expires_raw = str(lease.get("expires_at") or "").strip()
if expires_raw:
try:
expires_at = datetime.fromisoformat(expires_raw.replace("Z", "+00:00"))
except ValueError:
return {
"success": False,
"reason_code": "lease_not_live",
"message": (
f"lease_id '{lid}' records an unparseable expiry "
f"'{expires_raw}' (fail closed)."
),
"exact_next_action": (
"Re-allocate the work item and pass the live "
"assignment/lease pair."
),
}
if expires_at.tzinfo is None:
expires_at = expires_at.replace(tzinfo=timezone.utc)
if expires_at <= datetime.now(timezone.utc):
return {
"success": False,
"reason_code": "lease_not_live",
"message": (
f"lease_id '{lid}' expired at {expires_raw}; an expired lease "
"confers no ownership (fail closed)."
),
"exact_next_action": (
"Reclaim or re-allocate the lease, then retry with the live pair."
),
}
lease_session = str(lease.get("session_id") or "").strip()
if lease_session and lease_session != owner_session:
return {
gitea_mcp_server.py
+327 -51
View File
@@ -3580,62 +3580,291 @@ def _authenticated_username(host: str):
return user
# #943: process-local session identifier. The pre-existing call sites that mint a
# session id (workflow dashboard, lease adopt, lease reclaim) all build the same
# "<profile>-<pid>-<hex>" shape when the caller supplies none. Binding it once per
# process keeps lease-ownership comparisons stable for the life of the session
# instead of minting a fresh identifier — and therefore a fresh owner — on every
# call. Process-local only, never shared to a file (same rationale as the
# immutable session context in session_context_binding).
_ACTIVE_SESSION_ID: str | None = None
# #943 review 622 F3/F4: one coherent authority snapshot per mutation claim.
#
# The reviewed implementation read the identity from the pinned #714 session
# context and the profile from the live ``get_profile()``, so a sanctioned
# rebind could produce a claimant pair whose two halves came from different
# snapshots — and that pair is written durably into the issue lock. The
# canonical pairing is ``record_mutation_authority``: profile from
# ``get_profile()``, identity from ``_authenticated_username(host)``. This
# resolver reproduces that pairing, and uses the pinned session context only for
# drift detection, never as a value source.
_AUTHORITY_PROFILE_UNRESOLVED = "authority_profile_unresolved"
_AUTHORITY_IDENTITY_UNRESOLVED = "authority_identity_unresolved"
_AUTHORITY_IDENTITY_DRIFT = "authority_identity_drift"
_AUTHORITY_PROFILE_DRIFT = "authority_profile_drift"
def _active_username() -> str | None:
"""Authenticated identity bound to this session, or None when unbound.
def _active_mutation_authority(host: str | None) -> dict:
"""Resolve one coherent (identity, profile) authority pair, or a refusal.
Reads the immutable #714 session context that ``gitea_whoami`` seeds; it is
the authoritative identity pin every other mutation gate already consults.
Never re-derives or fabricates an identity: an unbound context returns None
so callers fail closed instead of acting as an unverified actor.
"""
ctx = session_ctx.get_session_context() or {}
return ((ctx.get("identity") or "").strip()) or None
Both halves come from the same live snapshot. The immutable #714 session
context is consulted only to detect drift: when it disagrees with the live
snapshot the result is a fail-closed refusal, never a silently blended pair.
def _active_profile_name() -> str | None:
"""Active runtime profile name, or None when it cannot be determined.
The live profile is authoritative (``get_profile``); the bound session
context is consulted only when the profile cannot be read, so the reported
name always describes the profile actually serving this process.
Returns a dict with ``ok`` True plus ``identity``/``profile_name``, or ``ok``
False plus ``reason_code``, ``reasons`` and ``expected``/``actual`` when a
drift or resolution failure is detected. Never raises for an unresolved
profile: the caller converts the refusal into a structured author block so
reason code, retryability and transport survival are preserved (F4).
"""
try:
profile = get_profile() or {}
except Exception:
profile = {}
name = (profile.get("profile_name") or "").strip()
if name:
return name
ctx = session_ctx.get_session_context() or {}
return ((ctx.get("profile_name") or "").strip()) or None
def _current_session_id() -> str | None:
"""Session identifier for this process, or None when the profile is unknown.
Uses the same "<profile>-<pid>-<hex>" shape as the existing lease call
sites. Bound once per process so repeated calls describe one session; fails
soft to None when the profile is undeterminable, letting callers fail closed
rather than inventing an owner.
"""
global _ACTIVE_SESSION_ID
if _ACTIVE_SESSION_ID:
return _ACTIVE_SESSION_ID
profile_name = _active_profile_name()
except (RuntimeError, ValueError, TypeError, KeyError, OSError) as exc:
# Narrow, and never a silent fallback to a previously cached name: an
# unresolvable profile is a fail-closed condition, matching
# record_mutation_authority's "active profile unresolved (fail closed)".
return {
"ok": False,
"reason_code": _AUTHORITY_PROFILE_UNRESOLVED,
"reasons": [
"active profile could not be resolved: "
f"{_redact(str(exc))} (fail closed)"
],
}
profile_name = (profile.get("profile_name") or "").strip()
if not profile_name:
return None
_ACTIVE_SESSION_ID = f"{profile_name}-{os.getpid()}-{uuid.uuid4().hex[:8]}"
return _ACTIVE_SESSION_ID
return {
"ok": False,
"reason_code": _AUTHORITY_PROFILE_UNRESOLVED,
"reasons": [
"active profile carries no profile_name (fail closed)"
],
}
identity = ""
if host:
identity = (_authenticated_username(host) or "").strip()
if not identity:
# A profile's expected_username is configuration, not proof of an
# authenticated actor, so it is never substituted here.
return {
"ok": False,
"reason_code": _AUTHORITY_IDENTITY_UNRESOLVED,
"reasons": [
"authenticated identity could not be resolved for the active "
"host (fail closed); call gitea_whoami and retry"
],
}
ctx = session_ctx.get_session_context() or {}
pinned_identity = (ctx.get("identity") or "").strip()
pinned_profile = (ctx.get("profile_name") or "").strip()
if pinned_identity and pinned_identity != identity:
return {
"ok": False,
"reason_code": _AUTHORITY_IDENTITY_DRIFT,
"reasons": [
"live authenticated identity disagrees with the bound session "
"context identity (fail closed)"
],
"expected": pinned_identity,
"actual": identity,
}
if pinned_profile and pinned_profile != profile_name:
return {
"ok": False,
"reason_code": _AUTHORITY_PROFILE_DRIFT,
"reasons": [
"live active profile disagrees with the bound session context "
"profile (fail closed)"
],
"expected": pinned_profile,
"actual": profile_name,
}
return {
"ok": True,
"identity": identity,
"profile_name": profile_name,
"session_context_bound": bool(pinned_identity or pinned_profile),
}
def _active_username(host: str | None = None) -> str | None:
"""Authenticated identity for the active mutation authority, else None.
Refuses unbound, blank and whitespace-only identities, and never substitutes
a profile's ``expected_username`` for an authenticated one.
"""
authority = _active_mutation_authority(host)
return authority.get("identity") if authority.get("ok") else None
def _active_profile_name(host: str | None = None) -> str | None:
"""Active profile name for the mutation authority, else None.
Shares the single snapshot with :func:`_active_username`, so the two can
never describe different authority states.
"""
authority = _active_mutation_authority(host)
return authority.get("profile_name") if authority.get("ok") else None
# #943 review 622 B1: the workflow session that owns the work — never a
# process-lifetime or PID-derived value.
#
# The reviewed implementation minted "<profile>-<pid>-<hex>" once per process.
# The MCP daemon outlives every task it serves, so that identifier conflates
# sequential author tasks and can never equal the control-plane session that
# owns an allocator-created lease; ``_verify_assignment_and_lease_ids`` therefore
# refused with lease_session_mismatch on the canonical allocated path.
# ``issue_lock_store.mint_task_session_id`` states the rule directly: an
# ownership key "deliberately contains no process identifier", because the PID
# "cannot identify *which* task holds a claim".
_SESSION_UNVERIFIED = "workflow_session_unverified"
_SESSION_REQUIRED = "workflow_session_required_for_allocated_work"
_SESSION_LOCK_OWNER_MISMATCH = "issue_lock_owner_mismatch"
def _resolve_owner_workflow_session(
*,
issue_number: int,
assignment_id: str | None,
lease_id: str | None,
session_id: str | None,
identity: str,
profile_name: str,
remote: str,
org: str | None,
repo: str | None,
) -> dict:
"""Resolve the authoritative workflow session that owns this work.
Precedence, each step fail-closed:
1. An explicit ``session_id`` is verified against the control-plane
``sessions`` table: it must exist, be active, and match this role and
profile. An unverifiable identifier is refused, never trusted.
2. Otherwise an existing issue lock for this issue supplies its per-task
``task_session_id``, but only when the lock's recorded claimant matches
the resolved authority pair.
3. Otherwise, when allocator identifiers are supplied, the request is
refused: ownership of an allocated assignment cannot be established
without its owning session, and the presence of the identifiers is not
itself evidence of ownership.
4. Otherwise — no allocator identifiers and no existing lock — a fresh
per-task key is minted through the canonical
``issue_lock_store.mint_task_session_id``. It carries no process
identifier and is never memoised, so sequential tasks on one daemon never
share an owner.
Whether the resolved session actually owns a supplied lease stays the
decision of ``author_issue_bootstrap._verify_assignment_and_lease_ids``;
this function never pre-empts, duplicates or bypasses that gate.
"""
declared = (session_id or "").strip()
if declared:
db, errs = _control_plane_db_or_error()
if db is None:
return {
"ok": False,
"reason_code": _SESSION_UNVERIFIED,
"reasons": errs,
}
try:
rows = db.list_sessions(statuses=("active",))
except Exception as exc: # noqa: BLE001 — surface structured
return {
"ok": False,
"reason_code": _SESSION_UNVERIFIED,
"reasons": [
"could not read control-plane sessions to verify "
f"session_id: {_redact(str(exc))} (fail closed)"
],
}
match = next(
(r for r in rows if str(r.get("session_id") or "") == declared), None
)
if match is None:
return {
"ok": False,
"reason_code": _SESSION_UNVERIFIED,
"reasons": [
f"session_id '{declared}' is not an active control-plane "
"session (fail closed)"
],
}
row_role = (match.get("role") or "").strip().lower()
row_profile = (match.get("profile") or "").strip()
if row_role and row_role != "author":
return {
"ok": False,
"reason_code": _SESSION_UNVERIFIED,
"reasons": [
f"session_id '{declared}' is recorded for role "
f"'{row_role}', not author (fail closed)"
],
"expected": "author",
"actual": row_role,
}
if row_profile and row_profile != profile_name:
return {
"ok": False,
"reason_code": _SESSION_UNVERIFIED,
"reasons": [
f"session_id '{declared}' is recorded for profile "
f"'{row_profile}', not '{profile_name}' (fail closed)"
],
"expected": row_profile,
"actual": profile_name,
}
return {"ok": True, "session_id": declared, "session_source": "declared"}
lock = None
try:
lock = issue_lock_store.load_issue_lock(
remote=remote,
org=org or "",
repo=repo or "",
issue_number=int(issue_number),
)
except Exception: # noqa: BLE001 — absent/unreadable lock is not fatal here
lock = None
lock_session = issue_lock_store.lease_task_session_id(lock) if lock else ""
if lock_session:
lease = (lock or {}).get("work_lease") or {}
claimant = lease.get("claimant") or {}
lock_identity = (claimant.get("username") or "").strip()
lock_profile = (claimant.get("profile") or "").strip()
if (lock_identity and lock_identity != identity) or (
lock_profile and lock_profile != profile_name
):
return {
"ok": False,
"reason_code": _SESSION_LOCK_OWNER_MISMATCH,
"reasons": [
f"issue #{int(issue_number)} is locked by "
f"'{lock_identity or 'unknown'}' ({lock_profile or 'unknown'}), "
f"not '{identity}' ({profile_name}) (fail closed)"
],
"expected": f"{lock_identity}/{lock_profile}",
"actual": f"{identity}/{profile_name}",
}
return {
"ok": True,
"session_id": lock_session,
"session_source": "issue_lock",
}
if (assignment_id or "").strip() or (lease_id or "").strip():
return {
"ok": False,
"reason_code": _SESSION_REQUIRED,
"reasons": [
"assignment_id/lease_id were supplied but no owning workflow "
"session could be established (fail closed); pass the "
"session_id that holds the lease, or bind the issue lock first"
],
}
return {
"ok": True,
"session_id": issue_lock_store.mint_task_session_id(
issue_lock_store.AUTHOR_ISSUE_WORK_LEASE
),
"session_source": "minted_task_key",
}
def _authenticated_actor(host: str) -> dict:
@@ -10234,6 +10463,7 @@ def gitea_bootstrap_author_issue_worktree(
branch_name: str | None = None,
worktree_path: str | None = None,
idempotency_key: str | None = None,
session_id: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
@@ -10255,6 +10485,14 @@ def gitea_bootstrap_author_issue_worktree(
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.
session_id: Optional workflow session that owns the assignment/lease.
Required when assignment_id/lease_id are supplied, because ownership
of an allocated lease is compared by session identifier and cannot
be derived from the daemon process (#943 review 622 B1). Verified
against the control-plane sessions table; an unverifiable value is
refused rather than trusted. Omit for an unallocated bootstrap: the
owning session is then taken from an existing issue lock, or a fresh
per-task key is minted.
remote: Known instance — 'dadeschools' or 'prgs'.
host: Override Gitea host.
org: Override Org.
@@ -10293,6 +10531,44 @@ def gitea_bootstrap_author_issue_worktree(
h, o, r = _resolve(remote, host, org, repo)
canonical_root = _canonical_local_git_root()
# One authority snapshot supplies both halves of the claimant pair (F3), and
# an unresolvable profile becomes a structured refusal rather than a silent
# fallback (F4).
authority = _active_mutation_authority(h)
if not authority.get("ok"):
return _author_mutation_block(
authority.get("reasons") or ["mutation authority unresolved"],
reason_code=authority.get("reason_code"),
retryable=False,
transport_survives=True,
expected=authority.get("expected"),
actual=authority.get("actual"),
issue_number=int(issue_number),
)
# The owning workflow session — never process- or PID-derived (B1).
session = _resolve_owner_workflow_session(
issue_number=issue_number,
assignment_id=assignment_id,
lease_id=lease_id,
session_id=session_id,
identity=authority["identity"],
profile_name=authority["profile_name"],
remote=remote,
org=o,
repo=r,
)
if not session.get("ok"):
return _author_mutation_block(
session.get("reasons") or ["owning workflow session unresolved"],
reason_code=session.get("reason_code"),
retryable=False,
transport_survives=True,
expected=session.get("expected"),
actual=session.get("actual"),
issue_number=int(issue_number),
)
import author_issue_bootstrap
return author_issue_bootstrap.bootstrap_author_issue_worktree(
@@ -10308,9 +10584,9 @@ def gitea_bootstrap_author_issue_worktree(
host=h,
org=o,
repo=r,
active_identity=_active_username(),
active_profile=_active_profile_name(),
owner_session=_current_session_id(),
active_identity=authority["identity"],
active_profile=authority["profile_name"],
owner_session=session["session_id"],
dry_run=dry_run,
)
tests/test_issue_943_runtime_context_helpers.py
+577 -288
View File
@@ -1,25 +1,33 @@
"""Regression: the author bootstrap wrapper's runtime-context helpers (#943).
"""Regression: author bootstrap runtime authority and session ownership (#943).
``gitea_bootstrap_author_issue_worktree`` passed three values down to
``author_issue_bootstrap.bootstrap_author_issue_worktree``::
Two rounds of defects live here.
active_identity=_active_username(),
active_profile=_active_profile_name(),
owner_session=_current_session_id(),
**Round 1 (#943 as filed).** ``gitea_bootstrap_author_issue_worktree`` passed
four values down to the bootstrap service that were never defined:
``_active_username``, ``_active_profile_name``, ``_current_session_id`` and
``_author_mutation_block``. Every call — dry-run included — raised
``NameError`` while evaluating the arguments, before the service was entered.
None of those three names was ever defined. Commit ``a942afe`` (#850) introduced
the references and no definition, so every call — dry-run included — raised
``NameError: name '_active_username' is not defined`` while evaluating the
arguments, before the bootstrap service was entered.
**Round 2 (review 622 on PR #944).** The first fix defined all four but made
``_current_session_id`` mint ``<profile>-<pid>-<hex>`` once per process. The MCP
daemon outlives every task it serves, so that value conflates sequential author
tasks and can never equal the control-plane session that owns an
allocator-created lease: ``_verify_assignment_and_lease_ids`` refused the whole
allocated path with ``lease_session_mismatch``. The reviewed round also read the
identity from the pinned session context while reading the profile from the live
profile, so a rebind could produce a mixed claimant pair, and it swallowed every
``get_profile()`` exception.
The defect was unreachable until PR #942 (#941) wired the bootstrap scope into
``workflow_scope_guard``: until then ``verify_preflight_purity`` refused first
with ``missing_issue_worktree``, so the guard fix is what exposed this.
These tests therefore drive real state, not mocks of internals: a temporary
control-plane SQLite database and a temporary issue-lock directory, both
redirected through the same environment variables production uses
(``GITEA_CONTROL_PLANE_DB``, ``GITEA_ISSUE_LOCK_DIR``). The ownership gate that
runs is the real one.
``test_every_global_referenced_by_the_wrapper_resolves`` is the test that would
have caught the original defect: it resolves every global name the wrapper's
body references. Asserting only that the three known helpers now exist would
not generalise to the next missing reference.
``test_every_global_referenced_by_the_wrapper_resolves`` remains: it is what
found ``_author_mutation_block``, and it generalises to the next missing
reference. It supplements the runtime coverage below rather than standing in for
it.
"""
from __future__ import annotations
@@ -34,16 +42,28 @@ import unittest
from unittest import mock
import author_issue_bootstrap as aib
import control_plane_db
import create_issue_bootstrap as cib
import gitea_mcp_server as gms
import issue_lock_store
import workflow_scope_guard
BOOTSTRAP_TASK = "bootstrap_author_issue_worktree"
WRAPPER_NAME = "gitea_bootstrap_author_issue_worktree"
RUNTIME_HELPERS = ("_active_username", "_active_profile_name", "_current_session_id")
RUNTIME_HELPERS = (
"_active_mutation_authority",
"_active_username",
"_active_profile_name",
"_resolve_owner_workflow_session",
"_author_mutation_block",
)
ORG = "Scaled-Tech-Consulting"
REPO = "Gitea-Tools"
IDENTITY = "jcwalker3"
PROFILE = "prgs-author"
# "<profile>-<pid>-<hex8>", the shape the pre-existing lease call sites mint.
SESSION_ID_RE = re.compile(r"^[A-Za-z0-9_.-]+-\d+-[0-9a-f]{8}$")
# A per-task ownership key must carry no process identifier (#790).
TASK_KEY_RE = re.compile(r"^author_issue_work-[0-9a-f]{16}$")
def _make_control_repo(tmp: str) -> tuple[str, str]:
@@ -71,13 +91,11 @@ def _make_control_repo(tmp: str) -> tuple[str, str]:
def _wrapper_ast() -> ast.FunctionDef:
"""Return the AST of the bootstrap wrapper as it exists on disk.
Read from source rather than ``inspect``: the tool decorator may replace the
callable, and the defect lived in the *source* argument expressions.
"""
path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"gitea_mcp_server.py")
"""Return the AST of the bootstrap wrapper as it exists on disk."""
path = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"gitea_mcp_server.py",
)
with open(path, encoding="utf-8") as fh:
tree = ast.parse(fh.read())
for node in ast.walk(tree):
@@ -86,36 +104,517 @@ def _wrapper_ast() -> ast.FunctionDef:
raise AssertionError(f"{WRAPPER_NAME} not found in gitea_mcp_server.py")
class RuntimeHelperResolutionTests(unittest.TestCase):
"""AC: every runtime helper the wrapper references is defined and callable."""
class _IsolatedControlPlane(unittest.TestCase):
"""Temp control-plane DB and temp issue-lock dir, via production env vars."""
def test_three_named_helpers_are_defined_and_callable(self):
for name in RUNTIME_HELPERS:
with self.subTest(helper=name):
self.assertTrue(
hasattr(gms, name), f"{name} is referenced but not defined"
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
self.tmp = self._tmp.name
self.db_path = os.path.join(self.tmp, "control-plane.sqlite3")
self.lock_dir = os.path.join(self.tmp, "issue-locks")
self.journals = os.path.join(self.tmp, "journals")
os.makedirs(self.lock_dir)
os.makedirs(self.journals)
env = mock.patch.dict(
os.environ,
{
control_plane_db.DB_PATH_ENV: self.db_path,
issue_lock_store.LOCK_DIR_ENV: self.lock_dir,
},
)
env.start()
self.addCleanup(env.stop)
self.db = control_plane_db.ControlPlaneDB(self.db_path)
def _allocate(self, session_id: str, *, issue: int = 943):
"""Create a real assignment + lease owned by *session_id*."""
self.db.upsert_session(
session_id=session_id, role="author", profile=PROFILE, pid=os.getpid()
)
res = self.db.assign_and_lease(
session_id=session_id, role="author", remote="prgs",
org=ORG, repo=REPO, kind="issue", number=issue,
)
self.assertEqual(res.outcome, "assigned", res)
return res.assignment_id, res.lease_id
def _authority(self):
"""A resolved authority pair, as the wrapper would compute it."""
return {"ok": True, "identity": IDENTITY, "profile_name": PROFILE}
def _resolve_session(self, **over):
kwargs = dict(
issue_number=943,
assignment_id=None,
lease_id=None,
session_id=None,
identity=IDENTITY,
profile_name=PROFILE,
remote="prgs",
org=ORG,
repo=REPO,
)
kwargs.update(over)
return gms._resolve_owner_workflow_session(**kwargs)
class OwnershipGateTests(_IsolatedControlPlane):
"""B2: the allocator-driven ownership path, against a real control plane."""
def setUp(self):
super().setUp()
self.repo, self.head = _make_control_repo(self.tmp)
def _bootstrap(self, **over):
kwargs = dict(
issue_number=943,
canonical_repo_root=self.repo,
expected_base_sha=self.head,
branch_name="fix/issue-943-runtime-context-helpers",
remote="prgs",
org=ORG,
repo=REPO,
active_identity=IDENTITY,
active_profile=PROFILE,
lock_dir=self.journals,
idempotency_key="test-943",
dry_run=True,
)
kwargs.update(over)
return aib.bootstrap_author_issue_worktree(**kwargs)
def test_true_owning_session_passes_the_ownership_gate(self):
"""The canonical owner reaches and completes the service."""
session = "prgs-author-task-a"
assignment_id, lease_id = self._allocate(session)
res = self._bootstrap(
assignment_id=assignment_id, lease_id=lease_id, owner_session=session
)
self.assertTrue(res.get("success"), res)
self.assertTrue(res.get("dry_run"))
self.assertEqual(res.get("base_sha"), self.head)
def test_different_session_is_refused(self):
session = "prgs-author-task-a"
assignment_id, lease_id = self._allocate(session)
res = self._bootstrap(
assignment_id=assignment_id,
lease_id=lease_id,
owner_session="prgs-author-task-b",
)
self.assertFalse(res.get("success"))
self.assertEqual(res.get("reason_code"), "lease_session_mismatch")
def test_process_derived_session_would_be_refused(self):
"""The reviewed round-1 value shape can never own an allocated lease."""
session = "prgs-author-task-a"
assignment_id, lease_id = self._allocate(session)
round_one_value = f"{PROFILE}-{os.getpid()}-deadbeef"
self.assertNotEqual(round_one_value, session)
res = self._bootstrap(
assignment_id=assignment_id,
lease_id=lease_id,
owner_session=round_one_value,
)
self.assertFalse(res.get("success"))
self.assertEqual(res.get("reason_code"), "lease_session_mismatch")
def test_unknown_lease_fails_closed(self):
session = "prgs-author-task-a"
assignment_id, _ = self._allocate(session)
res = self._bootstrap(
assignment_id=assignment_id,
lease_id="lease-does-not-exist",
owner_session=session,
)
self.assertFalse(res.get("success"))
self.assertEqual(res.get("reason_code"), "unknown_lease_id")
def test_released_lease_fails_closed(self):
session = "prgs-author-task-a"
assignment_id, lease_id = self._allocate(session)
self.db.release_lease(lease_id, session_id=session)
res = self._bootstrap(
assignment_id=assignment_id, lease_id=lease_id, owner_session=session
)
self.assertFalse(res.get("success"))
self.assertEqual(res.get("reason_code"), "lease_not_live")
def test_force_expired_lease_fails_closed(self):
session = "prgs-author-task-a"
assignment_id, lease_id = self._allocate(session)
self.db.force_expire_lease(lease_id, reason="test")
res = self._bootstrap(
assignment_id=assignment_id, lease_id=lease_id, owner_session=session
)
self.assertFalse(res.get("success"))
self.assertEqual(res.get("reason_code"), "lease_not_live")
def test_replacement_lease_does_not_inherit_prior_ownership(self):
"""A second task's lease is not ownable by the first task's session."""
first = "prgs-author-task-a"
assignment_a, lease_a = self._allocate(first)
self.db.release_lease(lease_a, session_id=first)
second = "prgs-author-task-b"
assignment_b, lease_b = self._allocate(second)
self.assertNotEqual(lease_a, lease_b)
res = self._bootstrap(
assignment_id=assignment_b, lease_id=lease_b, owner_session=first
)
self.assertFalse(res.get("success"))
self.assertEqual(res.get("reason_code"), "lease_session_mismatch")
def test_assignment_lease_identifier_mismatch_fails_closed(self):
session = "prgs-author-task-a"
_, lease_id = self._allocate(session)
res = self._bootstrap(
assignment_id="asn-not-the-recorded-one",
lease_id=lease_id,
owner_session=session,
)
self.assertFalse(res.get("success"))
self.assertEqual(res.get("reason_code"), "assignment_lease_mismatch")
def test_lease_id_without_assignment_id_fails_closed(self):
session = "prgs-author-task-a"
_, lease_id = self._allocate(session)
res = self._bootstrap(lease_id=lease_id, owner_session=session)
self.assertFalse(res.get("success"))
self.assertEqual(res.get("reason_code"), "incomplete_assignment_lease_ids")
def test_dry_run_with_valid_allocator_bindings_leaves_no_durable_state(self):
session = "prgs-author-task-a"
assignment_id, lease_id = self._allocate(session)
res = self._bootstrap(
assignment_id=assignment_id, lease_id=lease_id, owner_session=session
)
self.assertTrue(res.get("success"), res)
branches = subprocess.check_output(
["git", "-C", self.repo, "branch", "--list"], text=True
)
self.assertNotIn("issue-943", branches)
worktrees = subprocess.check_output(
["git", "-C", self.repo, "worktree", "list"], text=True
)
self.assertNotIn("issue-943", worktrees)
self.assertFalse(
os.path.exists(
os.path.join(self.repo, "branches",
"fix-issue-943-runtime-context-helpers")
)
)
journal = res.get("phase_journal") or {}
self.assertFalse(journal.get("completed"))
self.assertFalse(any((journal.get("artifacts_created") or {}).values()))
# The dry run must not have created an issue lock in the isolated dir.
self.assertEqual(os.listdir(self.lock_dir), [])
def test_apply_reaches_the_intended_transition_with_valid_bindings(self):
session = "prgs-author-task-a"
assignment_id, lease_id = self._allocate(session)
res = self._bootstrap(
assignment_id=assignment_id,
lease_id=lease_id,
owner_session=session,
dry_run=False,
)
self.assertTrue(res.get("success"), res)
self.assertNotEqual(res.get("dry_run"), True)
branches = subprocess.check_output(
["git", "-C", self.repo, "branch", "--list"], text=True
)
self.assertIn("issue-943", branches)
self.assertTrue(os.path.isdir(res.get("worktree_path") or ""))
class WorkflowSessionResolutionTests(_IsolatedControlPlane):
"""B1: the wrapper resolves the owning session, never a process identifier."""
def test_declared_session_is_verified_against_the_control_plane(self):
session = "prgs-author-task-a"
assignment_id, lease_id = self._allocate(session)
res = self._resolve_session(
session_id=session, assignment_id=assignment_id, lease_id=lease_id
)
self.assertTrue(res.get("ok"), res)
self.assertEqual(res.get("session_id"), session)
self.assertEqual(res.get("session_source"), "declared")
def test_unknown_declared_session_is_refused_not_trusted(self):
res = self._resolve_session(session_id="prgs-author-not-a-session")
self.assertFalse(res.get("ok"))
self.assertEqual(res.get("reason_code"), "workflow_session_unverified")
def test_declared_session_for_another_role_is_refused(self):
self.db.upsert_session(
session_id="prgs-reviewer-x", role="reviewer", profile="prgs-reviewer",
pid=os.getpid(),
)
res = self._resolve_session(session_id="prgs-reviewer-x")
self.assertFalse(res.get("ok"))
self.assertEqual(res.get("reason_code"), "workflow_session_unverified")
def test_declared_session_for_another_profile_is_refused(self):
self.db.upsert_session(
session_id="other-profile-session", role="author",
profile="prgs-controller", pid=os.getpid(),
)
res = self._resolve_session(session_id="other-profile-session")
self.assertFalse(res.get("ok"))
self.assertEqual(res.get("reason_code"), "workflow_session_unverified")
def test_allocated_work_without_a_session_is_refused(self):
"""Supplying a lease is not itself evidence of ownership."""
session = "prgs-author-task-a"
assignment_id, lease_id = self._allocate(session)
res = self._resolve_session(assignment_id=assignment_id, lease_id=lease_id)
self.assertFalse(res.get("ok"))
self.assertEqual(
res.get("reason_code"), "workflow_session_required_for_allocated_work"
)
def test_existing_issue_lock_supplies_its_per_task_session(self):
lock_session = issue_lock_store.mint_task_session_id(
issue_lock_store.AUTHOR_ISSUE_WORK_LEASE
)
path = issue_lock_store.lock_file_path(
remote="prgs", org=ORG, repo=REPO, issue_number=943,
lock_dir=self.lock_dir,
)
issue_lock_store.write_lock_file(
path,
{
"issue_number": 943,
"branch_name": "fix/issue-943-runtime-context-helpers",
"work_lease": {
"task_session_id": lock_session,
"claimant": {"username": IDENTITY, "profile": PROFILE},
},
},
) if hasattr(issue_lock_store, "write_lock_file") else _write_json(
path,
{
"issue_number": 943,
"branch_name": "fix/issue-943-runtime-context-helpers",
"work_lease": {
"task_session_id": lock_session,
"claimant": {"username": IDENTITY, "profile": PROFILE},
},
},
)
res = self._resolve_session()
self.assertTrue(res.get("ok"), res)
self.assertEqual(res.get("session_id"), lock_session)
self.assertEqual(res.get("session_source"), "issue_lock")
def test_issue_lock_owned_by_another_identity_is_refused(self):
path = issue_lock_store.lock_file_path(
remote="prgs", org=ORG, repo=REPO, issue_number=943,
lock_dir=self.lock_dir,
)
_write_json(
path,
{
"issue_number": 943,
"work_lease": {
"task_session_id": "author_issue_work-" + "0" * 16,
"claimant": {"username": "someone-else", "profile": PROFILE},
},
},
)
res = self._resolve_session()
self.assertFalse(res.get("ok"))
self.assertEqual(res.get("reason_code"), "issue_lock_owner_mismatch")
def test_unallocated_bootstrap_mints_a_per_task_key(self):
res = self._resolve_session()
self.assertTrue(res.get("ok"), res)
self.assertEqual(res.get("session_source"), "minted_task_key")
self.assertRegex(res["session_id"], TASK_KEY_RE)
def test_minted_key_contains_no_process_identifier(self):
res = self._resolve_session()
self.assertNotIn(str(os.getpid()), res["session_id"])
self.assertNotIn(PROFILE, res["session_id"])
def test_sequential_tasks_on_one_daemon_do_not_share_ownership(self):
"""The round-1 defect: one identifier per process for every task."""
first = self._resolve_session()["session_id"]
second = self._resolve_session()["session_id"]
third = self._resolve_session()["session_id"]
self.assertNotEqual(first, second)
self.assertNotEqual(second, third)
self.assertEqual(len({first, second, third}), 3)
def test_no_process_lifetime_cache_remains(self):
self.assertFalse(hasattr(gms, "_ACTIVE_SESSION_ID"))
self.assertFalse(hasattr(gms, "_current_session_id"))
class MutationAuthorityTests(unittest.TestCase):
"""F3/F4: one coherent authority pair, drift detected, no silent fallback."""
def _ctx(self, **over):
base = {"identity": IDENTITY, "profile_name": PROFILE}
base.update(over)
return base
def test_matching_live_and_pinned_authority_resolves(self):
with mock.patch.object(gms, "get_profile",
return_value={"profile_name": PROFILE}), \
mock.patch.object(gms, "_authenticated_username",
return_value=IDENTITY), \
mock.patch.object(gms.session_ctx, "get_session_context",
return_value=self._ctx()):
res = gms._active_mutation_authority("gitea.prgs.cc")
self.assertTrue(res.get("ok"), res)
self.assertEqual(res["identity"], IDENTITY)
self.assertEqual(res["profile_name"], PROFILE)
def test_identity_drift_fails_closed(self):
with mock.patch.object(gms, "get_profile",
return_value={"profile_name": PROFILE}), \
mock.patch.object(gms, "_authenticated_username",
return_value="someone-else"), \
mock.patch.object(gms.session_ctx, "get_session_context",
return_value=self._ctx()):
res = gms._active_mutation_authority("gitea.prgs.cc")
self.assertFalse(res.get("ok"))
self.assertEqual(res.get("reason_code"), "authority_identity_drift")
self.assertEqual(res.get("expected"), IDENTITY)
self.assertEqual(res.get("actual"), "someone-else")
def test_profile_drift_fails_closed(self):
with mock.patch.object(gms, "get_profile",
return_value={"profile_name": "prgs-controller"}), \
mock.patch.object(gms, "_authenticated_username",
return_value=IDENTITY), \
mock.patch.object(gms.session_ctx, "get_session_context",
return_value=self._ctx()):
res = gms._active_mutation_authority("gitea.prgs.cc")
self.assertFalse(res.get("ok"))
self.assertEqual(res.get("reason_code"), "authority_profile_drift")
def test_identity_and_profile_never_come_from_different_snapshots(self):
"""Round 2's mixed pair: pinned identity plus live profile."""
with mock.patch.object(gms, "get_profile",
return_value={"profile_name": "prgs-controller"}), \
mock.patch.object(gms, "_authenticated_username",
return_value="new-identity"), \
mock.patch.object(gms.session_ctx, "get_session_context",
return_value=self._ctx()):
res = gms._active_mutation_authority("gitea.prgs.cc")
self.assertFalse(res.get("ok"))
self.assertIsNone(gms._active_username("gitea.prgs.cc"))
self.assertIsNone(gms._active_profile_name("gitea.prgs.cc"))
def test_unresolvable_profile_is_a_structured_refusal_not_a_fallback(self):
"""F4: no bare-except fallback to a previously pinned profile name."""
with mock.patch.object(gms, "get_profile",
side_effect=RuntimeError("profile disabled")), \
mock.patch.object(gms, "_authenticated_username",
return_value=IDENTITY), \
mock.patch.object(gms.session_ctx, "get_session_context",
return_value=self._ctx()):
res = gms._active_mutation_authority("gitea.prgs.cc")
self.assertFalse(res.get("ok"))
self.assertEqual(res.get("reason_code"), "authority_profile_unresolved")
self.assertNotEqual(res.get("profile_name"), PROFILE)
def test_malformed_profile_without_name_fails_closed(self):
with mock.patch.object(gms, "get_profile", return_value={}), \
mock.patch.object(gms, "_authenticated_username",
return_value=IDENTITY), \
mock.patch.object(gms.session_ctx, "get_session_context",
return_value=None):
res = gms._active_mutation_authority("gitea.prgs.cc")
self.assertFalse(res.get("ok"))
self.assertEqual(res.get("reason_code"), "authority_profile_unresolved")
def test_unresolved_identity_fails_closed(self):
for value in (None, "", " "):
with self.subTest(identity=value):
with mock.patch.object(gms, "get_profile",
return_value={"profile_name": PROFILE}), \
mock.patch.object(gms, "_authenticated_username",
return_value=value), \
mock.patch.object(gms.session_ctx, "get_session_context",
return_value=None):
res = gms._active_mutation_authority("gitea.prgs.cc")
self.assertFalse(res.get("ok"))
self.assertEqual(
res.get("reason_code"), "authority_identity_unresolved"
)
self.assertTrue(callable(getattr(gms, name)), f"{name} not callable")
def test_helpers_accept_zero_arguments_as_called(self):
"""The wrapper calls each with no arguments; the signature must allow it."""
prior = gms._ACTIVE_SESSION_ID
self.addCleanup(setattr, gms, "_ACTIVE_SESSION_ID", prior)
def test_missing_host_cannot_yield_an_identity(self):
with mock.patch.object(gms, "get_profile",
return_value={"profile_name": PROFILE}), \
mock.patch.object(gms.session_ctx, "get_session_context",
return_value=None):
res = gms._active_mutation_authority(None)
self.assertFalse(res.get("ok"))
self.assertEqual(res.get("reason_code"), "authority_identity_unresolved")
def test_expected_username_is_never_substituted_for_authentication(self):
with mock.patch.object(
gms, "get_profile",
return_value={"profile_name": PROFILE, "username": IDENTITY},
), mock.patch.object(gms, "_authenticated_username", return_value=None), \
mock.patch.object(gms.session_ctx, "get_session_context",
return_value={"expected_username": IDENTITY}):
self.assertIsNone(gms._active_username("gitea.prgs.cc"))
def test_accessors_share_one_snapshot(self):
with mock.patch.object(gms, "get_profile",
return_value={"profile_name": PROFILE}), \
mock.patch.object(gms, "_authenticated_username",
return_value=IDENTITY), \
mock.patch.object(gms.session_ctx, "get_session_context",
return_value=self._ctx()):
self.assertEqual(gms._active_username("gitea.prgs.cc"), IDENTITY)
self.assertEqual(gms._active_profile_name("gitea.prgs.cc"), PROFILE)
class AuthorMutationBlockTests(unittest.TestCase):
"""Preserved: the structured refusal shape review 622 confirmed correct."""
def test_matches_the_sibling_refusal_shape(self):
res = gms._author_mutation_block(["stopped"])
self.assertIs(res["success"], False)
self.assertIs(res["performed"], False)
self.assertEqual(res["outcome"], "REFUSED")
self.assertEqual(res["reasons"], ["stopped"])
def test_carries_reason_code_and_transport_fields(self):
res = gms._author_mutation_block(
["nope"], reason_code="authority_identity_drift",
retryable=False, transport_survives=True,
expected="a", actual="b", issue_number=943,
)
self.assertEqual(res["reason_code"], "authority_identity_drift")
self.assertIs(res["retryable"], False)
self.assertIs(res["transport_survives"], True)
self.assertEqual((res["expected"], res["actual"]), ("a", "b"))
self.assertEqual(res["issue_number"], 943)
self.assertIs(res["success"], False)
class RuntimeHelperResolutionTests(unittest.TestCase):
"""Every runtime helper the wrapper references is defined and callable.
Supplements the runtime coverage above; it does not replace it.
"""
def test_named_helpers_are_defined_and_callable(self):
for name in RUNTIME_HELPERS:
with self.subTest(helper=name):
with mock.patch.object(gms, "get_profile", return_value={}), \
mock.patch.object(
gms.session_ctx, "get_session_context", return_value=None):
gms._ACTIVE_SESSION_ID = None
getattr(gms, name)() # must not raise TypeError
self.assertTrue(hasattr(gms, name), f"{name} is not defined")
self.assertTrue(callable(getattr(gms, name)))
def test_every_global_referenced_by_the_wrapper_resolves(self):
"""The generalised form of this defect: an unresolvable global name.
Collects every ``Name`` load in the wrapper body, subtracts locals
(arguments, assignments, comprehension targets, imports), and asserts the
remainder resolves against module globals or builtins.
"""
"""The generalised form of the round-1 defect: an unresolvable global."""
fn = _wrapper_ast()
bound: set[str] = {a.arg for a in fn.args.args}
bound |= {a.arg for a in fn.args.kwonlyargs}
@@ -124,7 +623,9 @@ class RuntimeHelperResolutionTests(unittest.TestCase):
if fn.args.kwarg:
bound.add(fn.args.kwarg.arg)
for node in ast.walk(fn):
if isinstance(node, ast.Name) and isinstance(node.ctx, (ast.Store, ast.Del)):
if isinstance(node, ast.Name) and isinstance(
node.ctx, (ast.Store, ast.Del)
):
bound.add(node.id)
elif isinstance(node, (ast.Import, ast.ImportFrom)):
for alias in node.names:
@@ -142,255 +643,34 @@ class RuntimeHelperResolutionTests(unittest.TestCase):
and not hasattr(builtins, node.id)
)
self.assertEqual(
unresolved, [], f"{WRAPPER_NAME} references undefined globals: {unresolved}"
unresolved, [],
f"{WRAPPER_NAME} references undefined globals: {unresolved}",
)
def test_wrapper_still_passes_all_three_runtime_values(self):
"""Guard the wiring itself: the fix must not be 'stop passing them'."""
def test_wrapper_wires_the_authority_and_session_resolvers(self):
fn = _wrapper_ast()
called = {
node.func.id
for node in ast.walk(fn)
if isinstance(node, ast.Call) and isinstance(node.func, ast.Name)
}
for name in RUNTIME_HELPERS:
with self.subTest(helper=name):
self.assertIn(name, called)
self.assertIn("_active_mutation_authority", called)
self.assertIn("_resolve_owner_workflow_session", called)
self.assertIn("_author_mutation_block", called)
class ActiveUsernameTests(unittest.TestCase):
"""AC: identity comes from the authoritative pin, and fails closed."""
def test_returns_identity_from_bound_session_context(self):
with mock.patch.object(
gms.session_ctx, "get_session_context",
return_value={"identity": "jcwalker3", "profile_name": "prgs-author"},
):
self.assertEqual(gms._active_username(), "jcwalker3")
def test_unbound_context_returns_none_so_callers_fail_closed(self):
with mock.patch.object(
gms.session_ctx, "get_session_context", return_value=None
):
self.assertIsNone(gms._active_username())
def test_blank_identity_is_not_treated_as_an_identity(self):
for blank in ("", " ", None):
with self.subTest(identity=blank):
with mock.patch.object(
gms.session_ctx, "get_session_context",
return_value={"identity": blank},
):
self.assertIsNone(gms._active_username())
def test_identity_is_not_fabricated_from_the_profile(self):
"""A profile's expected username must never stand in for a real identity."""
with mock.patch.object(
gms.session_ctx, "get_session_context",
return_value={"identity": None, "expected_username": "jcwalker3"},
):
self.assertIsNone(gms._active_username())
class ActiveProfileNameTests(unittest.TestCase):
"""AC: profile name comes from the live profile, context only as fallback."""
def test_prefers_the_live_profile(self):
with mock.patch.object(
gms, "get_profile", return_value={"profile_name": "prgs-author"}
), mock.patch.object(
gms.session_ctx, "get_session_context",
return_value={"profile_name": "prgs-reviewer"},
):
self.assertEqual(gms._active_profile_name(), "prgs-author")
def test_falls_back_to_session_context_when_profile_unreadable(self):
with mock.patch.object(gms, "get_profile", side_effect=RuntimeError("no cfg")), \
mock.patch.object(
gms.session_ctx, "get_session_context",
return_value={"profile_name": "prgs-author"}):
self.assertEqual(gms._active_profile_name(), "prgs-author")
def test_returns_none_when_neither_source_knows(self):
with mock.patch.object(gms, "get_profile", return_value={}), \
mock.patch.object(
gms.session_ctx, "get_session_context", return_value=None):
self.assertIsNone(gms._active_profile_name())
class CurrentSessionIdTests(unittest.TestCase):
"""AC: a real, stable session identifier — never a fresh owner per call."""
def setUp(self):
self._prior = gms._ACTIVE_SESSION_ID
gms._ACTIVE_SESSION_ID = None
self.addCleanup(setattr, gms, "_ACTIVE_SESSION_ID", self._prior)
def test_shape_matches_the_existing_lease_call_sites(self):
with mock.patch.object(
gms, "get_profile", return_value={"profile_name": "prgs-author"}
):
sid = gms._current_session_id()
self.assertRegex(sid, SESSION_ID_RE)
self.assertTrue(sid.startswith("prgs-author-"))
self.assertIn(str(os.getpid()), sid)
def test_stable_across_calls_within_one_process(self):
"""A new id per call would make lease-ownership checks unsatisfiable."""
with mock.patch.object(
gms, "get_profile", return_value={"profile_name": "prgs-author"}
):
first = gms._current_session_id()
second = gms._current_session_id()
third = gms._current_session_id()
self.assertEqual(first, second)
self.assertEqual(second, third)
def test_returns_none_when_profile_undeterminable(self):
with mock.patch.object(gms, "get_profile", return_value={}), \
mock.patch.object(
gms.session_ctx, "get_session_context", return_value=None):
self.assertIsNone(gms._current_session_id())
def test_none_result_is_not_memoised_as_a_session(self):
with mock.patch.object(gms, "get_profile", return_value={}), \
mock.patch.object(
gms.session_ctx, "get_session_context", return_value=None):
self.assertIsNone(gms._current_session_id())
with mock.patch.object(
gms, "get_profile", return_value={"profile_name": "prgs-author"}
):
self.assertIsNotNone(gms._current_session_id())
class BootstrapServiceReachedTests(unittest.TestCase):
"""AC: the values the helpers produce carry a dry-run into the service."""
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
self.tmp = self._tmp.name
self.repo, self.head = _make_control_repo(self.tmp)
self.journals = os.path.join(self.tmp, "journals")
os.makedirs(self.journals)
def _bootstrap(self, **over):
kwargs = dict(
issue_number=943,
canonical_repo_root=self.repo,
expected_base_sha=self.head,
branch_name="fix/issue-943-runtime-context-helpers",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
active_identity="jcwalker3",
active_profile="prgs-author",
owner_session="prgs-author-4242-abcdef12",
lock_dir=self.journals,
idempotency_key="test-943",
dry_run=True,
)
kwargs.update(over)
return aib.bootstrap_author_issue_worktree(**kwargs)
def test_dry_run_succeeds_with_helper_produced_bindings(self):
"""Feed the service exactly what the live helpers return."""
prior = gms._ACTIVE_SESSION_ID
self.addCleanup(setattr, gms, "_ACTIVE_SESSION_ID", prior)
with mock.patch.object(
gms, "get_profile", return_value={"profile_name": "prgs-author"}
), mock.patch.object(
gms.session_ctx, "get_session_context",
return_value={"identity": "jcwalker3", "profile_name": "prgs-author"},
):
gms._ACTIVE_SESSION_ID = None
identity = gms._active_username()
profile = gms._active_profile_name()
session = gms._current_session_id()
res = self._bootstrap(
active_identity=identity, active_profile=profile, owner_session=session
)
self.assertTrue(res.get("success"), res)
self.assertTrue(res.get("dry_run"))
self.assertEqual(res.get("issue_number"), 943)
self.assertEqual(res.get("base_sha"), self.head)
def test_dry_run_creates_no_branch_worktree_or_lease(self):
res = self._bootstrap()
self.assertTrue(res.get("success"), res)
branches = subprocess.check_output(
["git", "-C", self.repo, "branch", "--list"], text=True
)
self.assertNotIn("issue-943", branches)
worktrees = subprocess.check_output(
["git", "-C", self.repo, "worktree", "list"], text=True
)
self.assertNotIn("issue-943", worktrees)
self.assertFalse(
os.path.exists(os.path.join(self.repo, "branches",
"fix-issue-943-runtime-context-helpers"))
)
journal = res.get("phase_journal") or {}
self.assertFalse(journal.get("completed"))
self.assertFalse(any((journal.get("artifacts_created") or {}).values()))
self.assertIsNone(journal.get("lease_id"))
self.assertIsNone(journal.get("assignment_id"))
def test_missing_identity_fails_closed(self):
res = self._bootstrap(active_identity=None)
self.assertFalse(res.get("success"))
self.assertEqual(res.get("reason_code"), "missing_active_identity")
def test_missing_profile_fails_closed(self):
res = self._bootstrap(active_profile=" ")
self.assertFalse(res.get("success"))
self.assertEqual(res.get("reason_code"), "missing_active_profile")
def test_missing_session_fails_closed(self):
res = self._bootstrap(owner_session=None)
self.assertFalse(res.get("success"))
self.assertEqual(res.get("reason_code"), "missing_owner_session")
def test_expected_base_mismatch_fails_closed(self):
res = self._bootstrap(expected_base_sha="0" * 40)
self.assertFalse(res.get("success"))
self.assertEqual(res.get("reason_code"), "stale_concurrency_pin")
def test_unbound_runtime_context_cannot_reach_the_service(self):
"""With nothing bound, the helpers yield None and the service refuses."""
prior = gms._ACTIVE_SESSION_ID
self.addCleanup(setattr, gms, "_ACTIVE_SESSION_ID", prior)
with mock.patch.object(gms, "get_profile", return_value={}), \
mock.patch.object(
gms.session_ctx, "get_session_context", return_value=None):
gms._ACTIVE_SESSION_ID = None
res = self._bootstrap(
active_identity=gms._active_username(),
active_profile=gms._active_profile_name(),
owner_session=gms._current_session_id(),
)
self.assertFalse(res.get("success"))
self.assertIn(
res.get("reason_code"),
{"missing_owner_session", "missing_active_identity",
"missing_active_profile"},
)
def test_apply_reaches_the_intended_transition(self):
res = self._bootstrap(dry_run=False)
self.assertTrue(res.get("success"), res)
self.assertNotEqual(res.get("dry_run"), True)
branches = subprocess.check_output(
["git", "-C", self.repo, "branch", "--list"], text=True
)
self.assertIn("issue-943", branches)
self.assertTrue(os.path.isdir(res.get("worktree_path") or ""))
def test_wrapper_accepts_an_optional_session_id(self):
"""ABI addition stays backward compatible: optional, defaulting to None."""
fn = _wrapper_ast()
names = [a.arg for a in fn.args.args]
self.assertIn("session_id", names)
offset = len(names) - len(fn.args.defaults)
default = fn.args.defaults[names.index("session_id") - offset]
self.assertIsInstance(default, ast.Constant)
self.assertIsNone(default.value)
class Issue941ScopeGuardNotRegressedTests(unittest.TestCase):
"""AC: PR #942's bootstrap-scope wiring still holds."""
"""Preserved: PR #942's bootstrap-scope wiring still holds."""
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
@@ -454,5 +734,14 @@ class Issue941ScopeGuardNotRegressedTests(unittest.TestCase):
self.assertFalse(cib.is_create_issue_task(BOOTSTRAP_TASK))
def _write_json(path: str, payload: dict) -> None:
"""Write an issue-lock file directly, for lock-precedence tests."""
import json
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w", encoding="utf-8") as fh:
json.dump(payload, fh)
if __name__ == "__main__":
unittest.main()