allocator_service.py

@@ -23,6 +23,7 @@ import json
 import os
 import uuid
 from dataclasses import dataclass, field
+from datetime import datetime, timezone
 from typing import Any, Mapping, Sequence
 
 from control_plane_db import (
@@ -738,6 +739,46 @@ def normalize_exclude_issue_numbers(
     return sorted(out)
 
 
+def _claim_expires_at(claim: Any) -> datetime | None:
+    """Parse a claim's ``expires_at``, or ``None`` when it is absent/malformed."""
+    if not isinstance(claim, Mapping):
+        return None
+    text = str(claim.get("expires_at") or "").strip()
+    if not text:
+        return None
+    if text.endswith("Z"):
+        text = text[:-1] + "+00:00"
+    try:
+        parsed = datetime.fromisoformat(text)
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        parsed = parsed.replace(tzinfo=timezone.utc)
+    return parsed.astimezone(timezone.utc)
+
+
+def _drop_expired_claims(
+    claims: Mapping[tuple[str, int], dict[str, Any]],
+    *,
+    now: datetime | None = None,
+) -> dict[tuple[str, int], dict[str, Any]]:
+    """Claims minus those whose lease has already expired (#643).
+
+    The read-only mirror of ``expire_stale_leases``: the sweep marks such rows
+    ``expired`` so they stop being returned as claims, and this reaches the same
+    view without writing. A claim with no parseable ``expires_at`` is **kept** —
+    an unreadable expiry is not evidence that work is free.
+    """
+    moment = now or datetime.now(timezone.utc)
+    kept: dict[tuple[str, int], dict[str, Any]] = {}
+    for key, claim in (claims or {}).items():
+        expires_at = _claim_expires_at(claim)
+        if expires_at is not None and expires_at <= moment:
+            continue
+        kept[key] = claim
+    return kept
+
+
 def candidate_set_fingerprint(
     candidates: Sequence[WorkCandidate],
     *,
@@ -826,12 +867,22 @@ def allocate_next_work(
     exclude_issue_numbers: Sequence[int] | None = None,
     expected_candidate_set_fingerprint: str | None = None,
     allocation_mode: str | None = None,
+    side_effect_free: bool = False,
 ) -> dict[str, Any]:
     """Select and optionally reserve the next work unit via control-plane DB.
 
     *apply=False* (default): dry-run selection only — no lease/assignment.
     *apply=True*: atomic ``assign_and_lease`` for the selected candidate.
 
+    *side_effect_free* (#643): a dry run that writes **nothing** to the
+    control-plane DB. A plain ``apply=False`` still registered a session row and
+    swept stale leases globally, so a caller advertising a read-only preview was
+    mutating on every call. Under this flag both writes are suppressed and stale
+    leases are instead filtered out of the claim map in memory, which yields the
+    same selection the sweep would have produced without persisting anything.
+    Incompatible with *apply* — the combination fails closed rather than
+    silently reserving.
+
     *allocation_mode* (#840): ``cross_role`` (default for controller) inspects
     the complete queue and returns one authoritative selection naming the
     required downstream role/profile/action. ``role_scoped`` keeps prior
@@ -885,40 +936,57 @@ def allocate_next_work(
             "allocation_mode": (allocation_mode or "").strip() or None,
         }
 
-    session_id = (session_id or "").strip() or f"alloc-{uuid.uuid4().hex[:12]}"
-    try:
-        db.upsert_session(
-            session_id=session_id,
-            role=role_norm,
-            profile=profile_name,
-            pid=os.getpid(),
-            controller_instance_id=controller_instance_id,
-        )
-    except Exception as exc:  # noqa: BLE001 — surface structured
+    # A side-effect-free run may never reserve: reserving is a write, and the
+    # flag is the caller's assertion that this call writes nothing (#643).
+    if side_effect_free and apply:
         return {
             "success": False,
             "outcome": OUTCOME_NO_SAFE,
+            "apply": True,
             "reasons": [
-                f"failed to register session in control-plane DB: {exc} "
-                "(fail closed, #613)"
+                "side_effect_free is incompatible with apply=True; an "
+                "assignment is a write (fail closed, #643)"
             ],
             "skipped": [],
             "assignment": None,
             "substrate": "control_plane_db",
         }
 
-    # Expire stale leases globally before selection.
-    try:
-        db.expire_stale_leases()
-    except Exception as exc:  # noqa: BLE001
-        return {
-            "success": False,
-            "outcome": OUTCOME_NO_SAFE,
-            "reasons": [f"lease expiry failed: {exc} (fail closed)"],
-            "skipped": [],
-            "assignment": None,
-            "substrate": "control_plane_db",
-        }
+    session_id = (session_id or "").strip() or f"alloc-{uuid.uuid4().hex[:12]}"
+    if not side_effect_free:
+        try:
+            db.upsert_session(
+                session_id=session_id,
+                role=role_norm,
+                profile=profile_name,
+                pid=os.getpid(),
+                controller_instance_id=controller_instance_id,
+            )
+        except Exception as exc:  # noqa: BLE001 — surface structured
+            return {
+                "success": False,
+                "outcome": OUTCOME_NO_SAFE,
+                "reasons": [
+                    f"failed to register session in control-plane DB: {exc} "
+                    "(fail closed, #613)"
+                ],
+                "skipped": [],
+                "assignment": None,
+                "substrate": "control_plane_db",
+            }
+
+        # Expire stale leases globally before selection.
+        try:
+            db.expire_stale_leases()
+        except Exception as exc:  # noqa: BLE001
+            return {
+                "success": False,
+                "outcome": OUTCOME_NO_SAFE,
+                "reasons": [f"lease expiry failed: {exc} (fail closed)"],
+                "skipped": [],
+                "assignment": None,
+                "substrate": "control_plane_db",
+            }
 
     terminal = None
     try:
@@ -953,6 +1021,12 @@ def allocate_next_work(
                 "assignment": None,
                 "substrate": "control_plane_db",
             }
+        if side_effect_free:
+            # ``list_active_claims`` filters on status alone, so without the
+            # global sweep an already-expired lease would still read as a live
+            # claim and the preview would report work as taken that is free.
+            # Drop those in memory: same view the sweep produces, no write.
+            claims = _drop_expired_claims(claims)
 
     try:
         exclude_nums = normalize_exclude_issue_numbers(exclude_issue_numbers)

tests/test_allocator_service.py

@@ -7,6 +7,7 @@ import tempfile
 import threading
 import unittest
 from concurrent.futures import ThreadPoolExecutor, as_completed
+from datetime import datetime, timezone
 
 from allocator_service import (
     OUTCOME_ASSIGNED,
@@ -15,6 +16,7 @@ from allocator_service import (
     OUTCOME_PREVIEW,
     OUTCOME_WAIT,
     WorkCandidate,
+    _drop_expired_claims,
     allocate_next_work,
     candidate_from_dict,
     classify_skip,
@@ -362,5 +364,161 @@ class AllocatorServiceTest(unittest.TestCase):
         self.assertIn("unavailable", res["reasons"][0].lower())
 
 
+class SideEffectFreeAllocationTest(unittest.TestCase):
+    """``side_effect_free`` dry runs write nothing to the control plane (#643).
+
+    A plain ``apply=False`` still called ``upsert_session`` and
+    ``expire_stale_leases`` before the apply branch was consulted, so a caller
+    advertising a read-only preview mutated on every call — one unreferenced
+    session row per preview, plus a global lease sweep.
+    """
+
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def _alloc(self, **kwargs):
+        defaults = dict(
+            db=self.db,
+            session_id="s-preview",
+            role="author",
+            remote="prgs",
+            org="org",
+            repo="repo",
+            candidates=[
+                WorkCandidate(kind="issue", number=643, labels=("status:ready",))
+            ],
+            apply=False,
+            profile_name="prgs-author",
+            username="jcwalker3",
+        )
+        defaults.update(kwargs)
+        return allocate_next_work(**defaults)
+
+    def _session_ids(self) -> set[str]:
+        return {str(r.get("session_id")) for r in self.db.list_sessions()}
+
+    def test_side_effect_free_preview_writes_no_session_row(self):
+        before = self._session_ids()
+        result = self._alloc(side_effect_free=True)
+        self.assertEqual(result["outcome"], OUTCOME_PREVIEW)
+        self.assertEqual(self._session_ids(), before)
+        self.assertNotIn("s-preview", self._session_ids())
+
+    def test_plain_dry_run_still_registers_a_session(self):
+        # The default is unchanged for every existing caller.
+        self._alloc()
+        self.assertIn("s-preview", self._session_ids())
+
+    def test_repeated_previews_do_not_accumulate_rows(self):
+        for index in range(5):
+            self._alloc(side_effect_free=True, session_id=f"s-{index}")
+        self.assertEqual(self._session_ids(), set())
+
+    def test_side_effect_free_does_not_sweep_stale_leases(self):
+        self.db.upsert_session(session_id="owner", role="author", pid=1)
+        assigned = self.db.assign_and_lease(
+            session_id="owner",
+            role="author",
+            remote="prgs",
+            org="org",
+            repo="repo",
+            kind="issue",
+            number=999,
+            lease_ttl_seconds=-60,  # already expired
+        )
+        self.assertEqual(assigned.outcome, "assigned")
+
+        self._alloc(side_effect_free=True)
+
+        # The expired row is still 'active' in the DB: nothing swept it.
+        statuses = {
+            r["lease_id"]: r["status"]
+            for r in self.db.list_leases(
+                remote="prgs", org="org", repo="repo",
+                statuses=("active", "expired"),
+            )
+        }
+        self.assertEqual(statuses.get(assigned.lease_id), "active")
+
+    def test_expired_claims_are_filtered_in_memory_so_work_stays_selectable(self):
+        """The read-only mirror of the sweep: expired claims must not block."""
+        self.db.upsert_session(session_id="owner", role="author", pid=1)
+        self.db.assign_and_lease(
+            session_id="owner",
+            role="author",
+            remote="prgs",
+            org="org",
+            repo="repo",
+            kind="issue",
+            number=643,
+            lease_ttl_seconds=-60,  # expired: must not withhold #643
+        )
+        result = self._alloc(side_effect_free=True)
+        self.assertEqual(result["outcome"], OUTCOME_PREVIEW)
+        self.assertEqual(result["selected"]["number"], 643)
+
+    def test_a_live_claim_still_withholds_the_work(self):
+        self.db.upsert_session(session_id="owner", role="author", pid=1)
+        self.db.assign_and_lease(
+            session_id="owner",
+            role="author",
+            remote="prgs",
+            org="org",
+            repo="repo",
+            kind="issue",
+            number=643,
+            lease_ttl_seconds=3600,
+        )
+        result = self._alloc(side_effect_free=True)
+        self.assertNotEqual(result["outcome"], OUTCOME_ASSIGNED)
+        self.assertNotEqual((result.get("selected") or {}).get("number"), 643)
+
+    def test_side_effect_free_with_apply_fails_closed(self):
+        result = self._alloc(side_effect_free=True, apply=True)
+        self.assertFalse(result["success"])
+        self.assertEqual(result["outcome"], OUTCOME_NO_SAFE)
+        self.assertIsNone(result["assignment"])
+        self.assertIn("incompatible with apply", result["reasons"][0])
+        # And it reserved nothing.
+        self.assertEqual(
+            self.db.list_leases(remote="prgs", org="org", repo="repo"), []
+        )
+
+
+class DropExpiredClaimsTest(unittest.TestCase):
+    """The in-memory expiry filter behind side-effect-free previews (#643)."""
+
+    def test_unparseable_expiry_is_kept_rather_than_assumed_free(self):
+        claims = {
+            ("issue", 1): {"lease_id": "l1", "expires_at": "not-a-date"},
+            ("issue", 2): {"lease_id": "l2"},
+            ("issue", 3): {"lease_id": "l3", "expires_at": None},
+        }
+        self.assertEqual(_drop_expired_claims(claims), claims)
+
+    def test_expired_dropped_and_future_kept(self):
+        now = datetime(2026, 7, 25, 12, 0, tzinfo=timezone.utc)
+        claims = {
+            ("issue", 1): {"expires_at": "2026-07-25T11:59:59+00:00"},
+            ("issue", 2): {"expires_at": "2026-07-25T12:00:01+00:00"},
+            ("issue", 3): {"expires_at": "2026-07-25T12:00:00+00:00"},  # boundary
+        }
+        kept = _drop_expired_claims(claims, now=now)
+        self.assertEqual(set(kept), {("issue", 2)})
+
+    def test_naive_and_zulu_timestamps_are_treated_as_utc(self):
+        now = datetime(2026, 7, 25, 12, 0, tzinfo=timezone.utc)
+        claims = {
+            ("issue", 1): {"expires_at": "2026-07-25T11:00:00"},  # naive, past
+            ("issue", 2): {"expires_at": "2026-07-25T13:00:00Z"},  # zulu, future
+        }
+        kept = _drop_expired_claims(claims, now=now)
+        self.assertEqual(set(kept), {("issue", 2)})
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_webui_request_initiation.py

@@ -16,6 +16,7 @@ allocator refuses an incomplete inventory rather than ranking a partial set.
 
 from __future__ import annotations
 
+import contextlib
 import json
 import os
 import pathlib
@@ -89,6 +90,63 @@ def _selection(
     }
 
 
+@contextlib.contextmanager
+def _patched_control_plane(
+    *,
+    release_sink: list[tuple[str, str]] | None = None,
+    release_error: Exception | None = None,
+    leases=None,
+):
+    """Stand in for ``control_plane_db`` so compensation paths are observable.
+
+    The service imports the module inside the function, so patching
+    ``sys.modules`` is what intercepts it. No real DB is opened.
+    """
+    module = mock.MagicMock()
+    db = mock.MagicMock()
+
+    def _release(lease_id, *, session_id):
+        if release_error is not None:
+            raise release_error
+        if release_sink is not None:
+            release_sink.append((lease_id, session_id))
+
+    db.release_lease.side_effect = _release
+    db.list_leases.side_effect = leases or (lambda **_kwargs: [])
+    module.ControlPlaneDB.return_value = db
+    with mock.patch.dict(sys.modules, {"control_plane_db": module}):
+        yield db
+
+
+def _drifting_allocator(*, lease_id: str | None = "lease-wrong"):
+    """An allocator that previews the requested unit but assigns another.
+
+    This is the #643 B1 race in miniature: the CAS fingerprint hashes only
+    ``{kind, number}``, so a lease taken on the requested unit inside the window
+    leaves the fingerprint identical while the selection moves on.
+    """
+    assignment: dict[str, Any] = {"assignment_id": "asn-wrong"}
+    if lease_id:
+        assignment["lease_id"] = lease_id
+
+    def _drifting(
+        *, request, apply, expected_candidate_set_fingerprint=None, session_id=None
+    ):
+        return {
+            "outcome": (
+                allocator_service.OUTCOME_ASSIGNED
+                if apply
+                else allocator_service.OUTCOME_PREVIEW
+            ),
+            "selected": _selection(number=999 if apply else 643),
+            "assignment": dict(assignment) if apply else None,
+            "candidate_set_fingerprint": "fp-test",
+            "session_id": session_id,
+        }
+
+    return _drifting
+
+
 def _fake_allocator(
     *,
     selection: dict[str, Any] | None = None,
@@ -587,30 +645,170 @@ class TestApplyOutcomes(unittest.TestCase):
         self.assertFalse(result["mutation_performed"])
 
     def test_allocator_drift_on_apply_is_not_read_as_an_assignment(self):
-        """The apply call must return *this* work unit, not a substitute."""
+        """The apply call must return *this* work unit, not a substitute.
 
-        def _drifting(*, request, apply, expected_candidate_set_fingerprint=None):
+        Drift is not merely refused: the allocator has already committed the
+        substitute assignment by the time egress rejects it, so the refusal must
+        also release it. Asserting only ``mutation_performed is False`` would
+        pass just as well against a leak.
+        """
+        released: list[tuple[str, str]] = []
+
+        with _patched_control_plane(release_sink=released):
+            result = request_service.apply_request(
+                _request(number=643),
+                principal=_principal(console_authz.OPERATOR),
+                confirm=True,
+                allocator=_drifting_allocator(),
+                claims_source=_no_claims,
+            )
+
+        self.assertFalse(result["ok"])
+        self.assertIsNone(result["assignment"])
+        # The substitute assignment was released, so nothing durable survives.
+        self.assertEqual(len(released), 1)
+        self.assertEqual(released[0][0], "lease-wrong")
+        compensation = result["compensation"]
+        self.assertTrue(compensation["released"])
+        self.assertEqual(compensation["lease_id"], "lease-wrong")
+        self.assertEqual(compensation["assignment_id"], "asn-wrong")
+        self.assertEqual(compensation["selected"]["number"], 999)
+        self.assertFalse(result["mutation_performed"])
+        self.assertNotIn("orphaned_assignment", result)
+
+    def test_drift_whose_release_fails_reports_the_orphan_and_a_reclaim(self):
+        """A release that fails must surface the leak, never swallow it."""
+        with _patched_control_plane(release_error=RuntimeError("db is read-only")):
+            result = request_service.apply_request(
+                _request(number=643),
+                principal=_principal(console_authz.OPERATOR),
+                confirm=True,
+                allocator=_drifting_allocator(),
+                claims_source=_no_claims,
+            )
+
+        self.assertFalse(result["ok"])
+        self.assertIsNone(result["assignment"])
+        # A lease really is out there; saying "nothing changed" would be a lie.
+        self.assertTrue(result["mutation_performed"])
+        orphan = result["orphaned_assignment"]
+        self.assertFalse(orphan["released"])
+        self.assertTrue(orphan["attempted"])
+        self.assertIn("db is read-only", orphan["error"])
+        reclaim = orphan["reclaim_action"]
+        self.assertEqual(reclaim["tool"], "gitea_release_workflow_lease")
+        self.assertEqual(reclaim["lease_id"], "lease-wrong")
+
+    def test_drift_without_a_lease_id_still_surfaces_a_reclaim(self):
+        """An assignment with no lease id cannot be released — say so."""
+        result = request_service.apply_request(
+            _request(number=643),
+            principal=_principal(console_authz.OPERATOR),
+            confirm=True,
+            allocator=_drifting_allocator(lease_id=None),
+            claims_source=_no_claims,
+        )
+        self.assertFalse(result["ok"])
+        self.assertTrue(result["mutation_performed"])
+        orphan = result["orphaned_assignment"]
+        self.assertFalse(orphan["attempted"])
+        self.assertEqual(orphan["assignment_id"], "asn-wrong")
+        self.assertIn("reclaim_action", orphan)
+
+    def test_exception_after_commit_releases_the_session_lease(self):
+        """A post-commit exception surfaces as no result — with a live lease.
+
+        ``allocate_next_work`` catches only three exception types, so anything
+        else raised after ``assign_and_lease`` committed reaches the caller as
+        ``None`` while the lease is durable. The stable per-flow session id is
+        what makes that lease findable.
+        """
+        released: list[tuple[str, str]] = []
+        seen: list[str | None] = []
+
+        def _explodes_after_commit(
+            *, request, apply, expected_candidate_set_fingerprint=None, session_id=None
+        ):
+            seen.append(session_id)
+            if not apply:
+                return {
+                    "outcome": allocator_service.OUTCOME_PREVIEW,
+                    "selected": _selection(number=643),
+                    "assignment": None,
+                    "candidate_set_fingerprint": "fp-test",
+                }
+            raise KeyError("selection['required_profile']")
+
+        def _leases(**_kwargs):
+            return [
+                {
+                    "lease_id": "lease-committed",
+                    "session_id": seen[-1],
+                    "work_kind": "issue",
+                    "work_number": 643,
+                }
+            ]
+
+        with _patched_control_plane(release_sink=released, leases=_leases):
+            result = request_service.apply_request(
+                _request(number=643),
+                principal=_principal(console_authz.OPERATOR),
+                confirm=True,
+                allocator=_explodes_after_commit,
+                claims_source=_no_claims,
+            )
+
+        self.assertFalse(result["ok"])
+        self.assertEqual(
+            result["reason_code"], request_service.REASON_EVIDENCE_UNAVAILABLE
+        )
+        self.assertEqual(len(released), 1)
+        self.assertEqual(released[0][0], "lease-committed")
+        self.assertEqual(
+            result["compensation"]["released"][0]["lease_id"], "lease-committed"
+        )
+        self.assertFalse(result["mutation_performed"])
+
+    def test_one_session_id_spans_the_dry_run_and_the_apply(self):
+        """Both halves of an apply share one control-plane identity."""
+        seen: list[str | None] = []
+
+        def _recording(
+            *, request, apply, expected_candidate_set_fingerprint=None, session_id=None
+        ):
+            seen.append(session_id)
             return {
                 "outcome": (
                     allocator_service.OUTCOME_ASSIGNED
                     if apply
                     else allocator_service.OUTCOME_PREVIEW
                 ),
-                "selected": _selection(number=999 if apply else 643),
-                "assignment": {"assignment_id": "asn-wrong"} if apply else None,
+                "selected": _selection(number=643),
+                "assignment": (
+                    {
+                        "assignment_id": "asn-ok",
+                        "lease_id": "lease-ok",
+                        "expected_head_sha": None,
+                    }
+                    if apply
+                    else None
+                ),
                 "candidate_set_fingerprint": "fp-test",
+                "session_id": session_id,
             }
 
         result = request_service.apply_request(
             _request(number=643),
             principal=_principal(console_authz.OPERATOR),
             confirm=True,
-            allocator=_drifting,
+            allocator=_recording,
             claims_source=_no_claims,
         )
-        self.assertFalse(result["ok"])
-        self.assertIsNone(result["assignment"])
-        self.assertFalse(result["mutation_performed"])
+        self.assertTrue(result["ok"])
+        self.assertEqual(len(seen), 2)
+        self.assertTrue(all(s for s in seen))
+        self.assertEqual(seen[0], seen[1], "dry-run and apply must share one id")
+        self.assertTrue(seen[0].startswith("webui-request-"))
 
     def test_viewer_cannot_apply(self):
         result = request_service.apply_request(
@@ -634,6 +832,119 @@ class TestApplyOutcomes(unittest.TestCase):
         self.assertFalse(result["mutation_performed"])
 
 
+class TestDefaultAllocatorPastTheEarlyReturns(unittest.TestCase):
+    """``default_allocator`` beyond its two fail-closed guards (#643).
+
+    Both prior tests returned before ``ControlPlaneDB`` was ever constructed, so
+    the session id, the ``side_effect_free`` routing and the CAS round-trip had
+    no coverage at all — which is how a preview that writes session rows shipped.
+    """
+
+    def setUp(self):
+        from webui.queue_loader import QueueSnapshot
+
+        self.snapshot = QueueSnapshot(
+            project_id="p",
+            repo_label="r",
+            prs=(),
+            issues=(),
+            pr_pagination=None,
+            issue_pagination=None,
+        )
+
+    @contextlib.contextmanager
+    def _harness(self):
+        """Run the real ``default_allocator`` against a recorded allocator call."""
+        calls: dict[str, Any] = {}
+
+        def _allocate(db, **kwargs):
+            calls.update(kwargs)
+            return {"outcome": allocator_service.OUTCOME_PREVIEW}
+
+        module = mock.MagicMock()
+        with mock.patch.dict(sys.modules, {"control_plane_db": module}), \
+                mock.patch(
+                    "webui.queue_loader.load_queue_snapshot",
+                    return_value=self.snapshot,
+                ), \
+                mock.patch(
+                    "webui.traffic_loader.candidates_from_queue_snapshot",
+                    return_value=[],
+                ), \
+                mock.patch.object(
+                    allocator_service, "allocate_next_work", side_effect=_allocate
+                ):
+            yield calls
+
+    def test_preview_runs_side_effect_free_and_never_applies(self):
+        with self._harness() as calls:
+            result = request_service.default_allocator(
+                request=_request(), apply=False
+            )
+        self.assertIsNotNone(result)
+        self.assertTrue(calls["side_effect_free"])
+        self.assertFalse(calls["apply"])
+
+    def test_apply_is_not_side_effect_free(self):
+        with self._harness() as calls:
+            request_service.default_allocator(request=_request(), apply=True)
+        self.assertFalse(calls["side_effect_free"])
+        self.assertTrue(calls["apply"])
+
+    def test_caller_session_id_is_passed_through_verbatim(self):
+        with self._harness() as calls:
+            request_service.default_allocator(
+                request=_request(), apply=True, session_id="webui-request-fixed"
+            )
+        self.assertEqual(calls["session_id"], "webui-request-fixed")
+
+    def test_absent_session_id_is_minted_with_the_expected_shape(self):
+        with self._harness() as calls:
+            request_service.default_allocator(request=_request(), apply=False)
+        self.assertTrue(str(calls["session_id"]).startswith("webui-request-"))
+
+    def test_scope_and_fingerprint_reach_the_allocator(self):
+        with self._harness() as calls:
+            request_service.default_allocator(
+                request=_request(number=664),
+                apply=False,
+                expected_candidate_set_fingerprint="fp-pinned",
+            )
+        self.assertEqual(calls["expected_candidate_set_fingerprint"], "fp-pinned")
+        self.assertEqual(calls["remote"], SCOPE["remote"])
+        self.assertEqual(calls["org"], SCOPE["org"])
+        self.assertEqual(calls["repo"], SCOPE["repo"])
+        self.assertEqual(calls["allocation_mode"], "role_scoped")
+
+
+class TestDefaultClaimsSource(unittest.TestCase):
+    """``default_claims_source`` had no test at all (#643)."""
+
+    def test_claims_are_read_for_the_request_scope(self):
+        db = mock.MagicMock()
+        db.list_active_claims.return_value = {("issue", 643): {"lease_id": "l1"}}
+        module = mock.MagicMock()
+        module.ControlPlaneDB.return_value = db
+        with mock.patch.dict(sys.modules, {"control_plane_db": module}):
+            claims = request_service.default_claims_source(_request())
+        self.assertEqual(claims, {("issue", 643): {"lease_id": "l1"}})
+        db.list_active_claims.assert_called_once_with(
+            remote=SCOPE["remote"], org=SCOPE["org"], repo=SCOPE["repo"]
+        )
+
+    def test_an_unreadable_substrate_denies_rather_than_returning_empty(self):
+        module = mock.MagicMock()
+        module.ControlPlaneDB.side_effect = RuntimeError("no db")
+        with mock.patch.dict(sys.modules, {"control_plane_db": module}):
+            # _load_claims converts the failure into None, which fails the
+            # lease check closed; an empty mapping would read as "nothing
+            # claimed" and wrongly authorize.
+            claims = request_service._load_claims(
+                _request(), request_service.default_claims_source
+            )
+        self.assertIsNone(claims)
+
+
 class TestAllocatorIntegrationFakes(unittest.TestCase):
     """The real default allocator refuses a partial inventory (#758)."""
 

webui/request_service.py

@@ -35,6 +35,7 @@ the resulting assignment by ``correlation_id``.
 
 from __future__ import annotations
 
+import inspect
 import uuid
 from dataclasses import dataclass, field
 from typing import Any, Callable, Mapping, Sequence
@@ -321,6 +322,17 @@ def _correlation_id() -> str:
     return f"req-{uuid.uuid4().hex}"
 
 
+def _new_session_id() -> str:
+    """Mint the control-plane session id for one request flow (#643).
+
+    Minted once per flow and reused by the dry-run and the apply, so a lease the
+    apply creates is owned by an id the caller still holds and can release. When
+    each call minted its own id, an apply wrote two session rows and bound the
+    lease to the second — leaving it with no owner able to reclaim it.
+    """
+    return f"webui-request-{uuid.uuid4().hex[:12]}"
+
+
 def _selection_matches(
     selection: Mapping[str, Any] | None, request: WorkRequest
 ) -> bool:
@@ -561,8 +573,15 @@ def preview_request(
     claims_source: ClaimsFn | None = None,
     correlation_id: str | None = None,
     audit: bool = True,
+    session_id: str | None = None,
 ) -> RequestPreview:
-    """Build the read-only intent preview for *request*. Never mutates."""
+    """Build the read-only intent preview for *request*. Never mutates.
+
+    "Never mutates" is now literal on the default path: the allocator runs
+    ``side_effect_free``, so no control-plane session row is written and no lease
+    sweep runs. *session_id* is threaded from an enclosing apply so both halves
+    of that flow share one control-plane identity (#643).
+    """
     who = principal or console_authz.ANONYMOUS
     corr = correlation_id or _correlation_id()
     decision = console_authz.authorize(ACTION_ID, who, for_execution=False)
@@ -573,7 +592,9 @@ def preview_request(
     if decision.allowed:
         # An unauthorized principal never reaches the allocator or the
         # control-plane DB: a denial must not double as a queue oracle.
-        allocation = _run_allocator(request, allocator, apply=False)
+        allocation = _run_allocator(
+            request, allocator, apply=False, session_id=session_id
+        )
         claims = _load_claims(request, claims_source)
         checks.append(_capability_check(request))
         checks.append(_lease_check(request, claims))
@@ -650,9 +671,20 @@ def apply_request(
     The only path to an assignment is the allocator agreeing, on a dry-run,
     that this work unit is what the requested role should take next. Every
     refusal returns before any mutation is attempted.
+
+    One exception is unavoidable and is compensated rather than prevented: the
+    allocator can commit an assignment and only then reveal, on egress, that it
+    selected a different work unit than the one requested. That assignment is
+    released before refusing, and ``mutation_performed`` reports whether any
+    durable control-plane state survives this call — ``False`` once the stray
+    assignment is gone, ``True`` with an explicit ``orphaned_assignment`` and
+    reclaim action when the release did not succeed (#643).
     """
     who = principal or console_authz.ANONYMOUS
     corr = correlation_id or _correlation_id()
+    # One identity for the whole flow, so a lease the apply creates is owned by
+    # an id this function still holds and can release.
+    session_id = _new_session_id()
 
     execution_decision = console_authz.authorize(ACTION_ID, who, for_execution=True)
     authorization = execution_decision.to_dict()
@@ -664,6 +696,7 @@ def apply_request(
         *,
         status: int,
         extra: dict[str, Any] | None = None,
+        mutation_performed: bool = False,
     ) -> dict[str, Any]:
         _audit(
             request,
@@ -689,7 +722,9 @@ def apply_request(
             "authorization": authorization,
             "assignment": None,
             "correlation_id": corr,
-            "mutation_performed": False,
+            # True only when durable control-plane state survives this refusal —
+            # an assignment the allocator created that could not be released.
+            "mutation_performed": bool(mutation_performed),
             "status_code": status,
         }
         payload.update(extra or {})
@@ -724,6 +759,7 @@ def apply_request(
         claims_source=claims_source,
         correlation_id=corr,
         audit=False,
+        session_id=session_id,
     )
     if not preview.authorized:
         outcome = (
@@ -747,21 +783,37 @@ def apply_request(
         allocator,
         apply=True,
         expected_candidate_set_fingerprint=fingerprint,
+        session_id=session_id,
     )
     if not allocation:
+        # "No result" is not proof of "no write": allocate_next_work only catches
+        # three exception types, so anything raised after assign_and_lease
+        # committed lands here with a lease already durable. Release whatever
+        # this flow's session owns before refusing.
+        sweep = _sweep_session_leases(request, session_id=session_id)
+        extra: dict[str, Any] = {}
+        detail = "the allocator returned no result; nothing was assigned."
+        if sweep.get("released"):
+            extra["compensation"] = sweep
+            detail = (
+                "the allocator returned no result after creating an assignment; "
+                "the assignment was released and nothing remains assigned."
+            )
+        elif sweep.get("error"):
+            extra["compensation"] = sweep
         return _refuse(
             OUTCOME_WAIT,
             REASON_EVIDENCE_UNAVAILABLE,
-            "the allocator returned no result; nothing was assigned.",
+            detail,
             status=503,
+            extra=extra or None,
         )
 
     assignment = allocation.get("assignment") or None
     outcome = _clean(allocation.get("outcome"))
+    committed = bool(outcome == allocator_service.OUTCOME_ASSIGNED and assignment)
     assigned = bool(
-        outcome == allocator_service.OUTCOME_ASSIGNED
-        and assignment
-        and _selection_matches(allocation.get("selected"), request)
+        committed and _selection_matches(allocation.get("selected"), request)
     )
     if not assigned:
         blocked = outcome in {
@@ -769,15 +821,45 @@ def apply_request(
             allocator_service.OUTCOME_BLOCKED_TERMINAL,
             allocator_service.OUTCOME_BLOCKED_EXCLUDED_OWN_LEASE,
         }
+        extra = {"allocator_evidence": _allocator_evidence(allocation)}
+        detail = (
+            f"the allocator returned {outcome or 'no outcome'} rather than "
+            "an assignment for this work unit; nothing was assigned."
+        )
+        if committed:
+            # The allocator assigned a *different* unit than the one requested.
+            # That assignment is real and durable; releasing it is the whole
+            # difference between a refusal and a silent leak.
+            compensation = _release_stray_assignment(
+                request, allocation, session_id=session_id
+            )
+            extra["compensation"] = compensation
+            if compensation.get("released"):
+                detail = (
+                    "the allocator assigned a different work unit than the one "
+                    "requested; that assignment was released and nothing "
+                    "remains assigned."
+                )
+            else:
+                extra["orphaned_assignment"] = compensation
+                return _refuse(
+                    OUTCOME_BLOCKED if blocked else OUTCOME_WAIT,
+                    REASON_ALLOCATOR_OUTCOME,
+                    (
+                        "the allocator assigned a different work unit than the "
+                        "one requested and it could not be released; it must be "
+                        "reclaimed explicitly."
+                    ),
+                    status=409,
+                    extra=extra,
+                    mutation_performed=True,
+                )
         return _refuse(
             OUTCOME_BLOCKED if blocked else OUTCOME_WAIT,
             REASON_ALLOCATOR_OUTCOME,
-            (
-                f"the allocator returned {outcome or 'no outcome'} rather than "
-                "an assignment for this work unit; nothing was assigned."
-            ),
+            detail,
             status=409,
-            extra={"allocator_evidence": _allocator_evidence(allocation)},
+            extra=extra,
         )
 
     _audit(
@@ -853,19 +935,175 @@ def _run_allocator(
     *,
     apply: bool,
     expected_candidate_set_fingerprint: str | None = None,
+    session_id: str | None = None,
 ) -> dict[str, Any] | None:
     fn = allocator or default_allocator
+    kwargs: dict[str, Any] = {
+        "request": request,
+        "apply": apply,
+        "expected_candidate_set_fingerprint": expected_candidate_set_fingerprint,
+    }
+    if session_id:
+        # Injected allocators in tests predate this argument; only pass it to
+        # callables that accept it so a fake signature is never broken.
+        if _accepts_session_id(fn):
+            kwargs["session_id"] = session_id
     try:
-        result = fn(
-            request=request,
-            apply=apply,
-            expected_candidate_set_fingerprint=expected_candidate_set_fingerprint,
-        )
+        result = fn(**kwargs)
     except Exception:  # noqa: BLE001 — an allocator failure denies, never proceeds
         return None
     return result if isinstance(result, dict) else None
 
 
+def _accepts_session_id(fn: AllocatorFn) -> bool:
+    try:
+        params = inspect.signature(fn).parameters
+    except (TypeError, ValueError):  # builtins / C callables
+        return False
+    if "session_id" in params:
+        return True
+    return any(p.kind is inspect.Parameter.VAR_KEYWORD for p in params.values())
+
+
+def _release_stray_assignment(
+    request: WorkRequest,
+    allocation: Mapping[str, Any],
+    *,
+    session_id: str | None,
+) -> dict[str, Any]:
+    """Release an assignment the allocator committed for the wrong work unit.
+
+    The apply path can commit an assignment and then discover, on egress, that
+    the allocator selected a *different* unit than the one requested — the CAS
+    fingerprint hashes only ``{kind, number}`` plus exclusions, so a lease taken
+    on the requested unit inside the window leaves the fingerprint identical and
+    the pin passes while the selection moves on. Without compensation that lease
+    is orphaned: owned by a session nothing heartbeats, and invisible because the
+    response says nothing was mutated.
+
+    Returns a record describing what was attempted so the caller can report it,
+    including a reclaim action when the release itself did not succeed.
+    """
+    assignment = allocation.get("assignment") or {}
+    lease_id = _clean(assignment.get("lease_id")) or None
+    assignment_id = _clean(assignment.get("assignment_id")) or None
+    owner = _clean(allocation.get("session_id")) or session_id or None
+    selected = allocation.get("selected") or {}
+    record: dict[str, Any] = {
+        "attempted": False,
+        "released": False,
+        "lease_id": lease_id,
+        "assignment_id": assignment_id,
+        "owner_session_id": owner,
+        "selected": {
+            "kind": _clean(selected.get("kind")).lower() or None,
+            "number": selected.get("number"),
+        },
+    }
+    if not lease_id:
+        # Nothing durable to release (or the allocator reported no lease id);
+        # still surface the assignment id so a leak is never silent.
+        record["detail"] = (
+            "the allocator reported an assignment with no lease id; nothing "
+            "could be released automatically."
+        )
+        if assignment_id:
+            record["reclaim_action"] = _reclaim_action(record)
+        return record
+    if not owner:
+        record["detail"] = (
+            "the owning session id is unknown; the assignment cannot be "
+            "released automatically."
+        )
+        record["reclaim_action"] = _reclaim_action(record)
+        return record
+
+    record["attempted"] = True
+    try:
+        import control_plane_db
+
+        control_plane_db.ControlPlaneDB().release_lease(lease_id, session_id=owner)
+    except Exception as exc:  # noqa: BLE001 — a failed release must be reported
+        record["error"] = str(exc)
+        record["detail"] = (
+            "the allocator created an assignment for a different work unit and "
+            "releasing it failed; it must be reclaimed explicitly."
+        )
+        record["reclaim_action"] = _reclaim_action(record)
+        return record
+
+    record["released"] = True
+    record["detail"] = (
+        "the allocator created an assignment for a different work unit; it was "
+        "released, so no assignment persists from this request."
+    )
+    return record
+
+
+def _reclaim_action(record: Mapping[str, Any]) -> dict[str, Any]:
+    """The explicit operator action for an assignment that outlived its request."""
+    lease_id = record.get("lease_id")
+    return {
+        "tool": "gitea_release_workflow_lease",
+        "lease_id": lease_id,
+        "session_id": record.get("owner_session_id"),
+        "assignment_id": record.get("assignment_id"),
+        "instructions": (
+            "A control-plane assignment was created for a work unit other than "
+            "the requested one and could not be released automatically. Call "
+            f"gitea_release_workflow_lease(lease_id={lease_id!r}, "
+            f"session_id={record.get('owner_session_id')!r}) to reclaim it, or "
+            "wait for the lease TTL to expire."
+        ),
+    }
+
+
+def _sweep_session_leases(
+    request: WorkRequest, *, session_id: str | None
+) -> dict[str, Any]:
+    """Release any lease this flow's session owns after an indeterminate apply.
+
+    ``_run_allocator`` returns ``None`` for *any* exception, and
+    ``allocate_next_work`` only catches ``InvalidWorkKindError``,
+    ``LeaseRequiredError`` and ``ControlPlaneError`` — so an unexpected error
+    raised *after* ``assign_and_lease`` committed surfaces as "no result" with a
+    lease already written. Because the session id is now stable across the flow,
+    that lease is findable: anything this session owns after a failed apply is by
+    definition unclaimed by anyone, so it is released.
+    """
+    record: dict[str, Any] = {"attempted": False, "released": [], "session_id": session_id}
+    if not session_id:
+        return record
+    record["attempted"] = True
+    try:
+        import control_plane_db
+
+        db = control_plane_db.ControlPlaneDB()
+        leases = db.list_leases(
+            remote=request.remote,
+            org=request.org,
+            repo=request.repo,
+            statuses=("active",),
+        )
+        for row in leases:
+            if _clean(row.get("session_id")) != session_id:
+                continue
+            lease_id = _clean(row.get("lease_id")) or None
+            if not lease_id:
+                continue
+            db.release_lease(lease_id, session_id=session_id)
+            record["released"].append(
+                {
+                    "lease_id": lease_id,
+                    "work_kind": row.get("work_kind"),
+                    "work_number": row.get("work_number"),
+                }
+            )
+    except Exception as exc:  # noqa: BLE001 — best-effort; never mask the refusal
+        record["error"] = str(exc)
+    return record
+
+
 def _load_claims(
     request: WorkRequest, claims_source: ClaimsFn | None
 ) -> Mapping[tuple[str, int], dict[str, Any]] | None:
@@ -894,12 +1132,19 @@ def default_allocator(
     request: WorkRequest,
     apply: bool,
     expected_candidate_set_fingerprint: str | None = None,
+    session_id: str | None = None,
 ) -> dict[str, Any] | None:
     """Run the real allocator over the live queue for *request*'s scope.
 
     An incomplete candidate inventory returns ``None`` rather than a ranking
     over a partial set (#758): selecting from a short list can pick the wrong
     work unit, so the request denies instead.
+
+    *session_id* is supplied by the caller so the dry-run and the apply share one
+    control-plane identity; the lease an apply creates is then owned by an id the
+    request flow still holds. A dry run additionally goes through the allocator's
+    ``side_effect_free`` path, so a preview writes no session row and sweeps no
+    leases (#643).
     """
     import control_plane_db
 
@@ -917,7 +1162,7 @@ def default_allocator(
     db = control_plane_db.ControlPlaneDB()
     result = allocator_service.allocate_next_work(
         db,
-        session_id=f"webui-request-{uuid.uuid4().hex[:12]}",
+        session_id=session_id or _new_session_id(),
         role=request.desired_role,
         remote=request.remote,
         org=request.org,
@@ -926,6 +1171,7 @@ def default_allocator(
         apply=bool(apply),
         allocation_mode="role_scoped",
         expected_candidate_set_fingerprint=expected_candidate_set_fingerprint,
+        side_effect_free=not apply,
     )
     if isinstance(result, dict):
         result.setdefault("candidate_count", len(candidates))