fix(case): preserve attended authority at dispatch and validate session lanes

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a1468b-63fa-7710-85dc-8f2e16b3ebfa

src/warden/case_run.py

@@ -587,7 +587,18 @@ def _catalog_lane(catalog_source: bytes, descriptor: Descriptor):
         raise Refused("descriptor owner does not fit catalog lane owner_repo")
     if descriptor.effect in ("mint", "change") and entry.risk != "high":
         raise Refused("catalog lane risk does not cover the descriptor effect")
-    return entry
+    return entry, entries
+
+
+def _attended_admission(descriptor, actor, entry, entries):
+    from .routing.models import attended_lane_issue
+
+    if (actor != "human" or "WARDEN_AGENT_ID" in os.environ
+            or "parent-session" not in descriptor.revocation_scopes):
+        raise Refused("attended step requires a human parent session")
+    issue = attended_lane_issue(entry, entries.get)
+    if issue:
+        raise Refused(issue[1])
 
 
 def _step_mode(case_type: dict, case: dict, descriptor: Descriptor) -> str:
@@ -711,14 +722,10 @@ class Executor:
             actor != "human" or os.getenv("WARDEN_AGENT_ID")
         ):
             raise Refused("agent mint is forbidden")
-        entry = _catalog_lane(catalog_source, descriptor)
+        entry, entries = _catalog_lane(catalog_source, descriptor)
         mode = _step_mode(case_type, case, descriptor)
-        if mode == "attended" and (
-            actor != "human"
-            or os.getenv("WARDEN_AGENT_ID")
-            or "parent-session" not in descriptor.revocation_scopes
-        ):
-            raise Refused("attended step requires a human parent session")
+        if mode == "attended":
+            _attended_admission(descriptor, actor, entry, entries)
         if mode == "owner" and (
             entry.effective_delegation.mode != "native"
             or entry.owner_repo != descriptor.owner
@@ -796,7 +803,11 @@ class Executor:
             descriptor.allowed_created,
             descriptor.retry_safe,
         )
-        self._admitted[invocation] = (descriptor, initial, session_evidence)
+        # Keep the mode and exact authenticated catalog bytes with this admission;
+        # dispatch must not substitute a later ambient catalog or lose its mode.
+        self._admitted[invocation] = (
+            descriptor, initial, session_evidence, mode, catalog_source
+        )
         return descriptor, invocation, initial
 
     def _result(
@@ -861,7 +872,7 @@ class Executor:
             descriptor.adapter_source,
         ):
             raise Refused("adapter changed after admission")
-        session = self._admitted[invocation][2]
+        _, _, session, mode, catalog_source = self._admitted[invocation]
         if (
             self._verify_evidence(
                 f"caller-session:{invocation.actor}:{invocation.principal}:{invocation.lane}",
@@ -873,6 +884,9 @@ class Executor:
             raise Refused("caller session no longer verified")
         if descriptor.effect == "mint" and os.getenv("WARDEN_AGENT_ID"):
             raise Refused("agent mint is forbidden")
+        if mode == "attended":
+            entry, entries = _catalog_lane(catalog_source, descriptor)
+            _attended_admission(descriptor, invocation.actor, entry, entries)
         return adapter
 
     def _attempt(self, invocation: Invocation, state: StepState, operation: str):

src/warden/case_type.py

@@ -24,6 +24,7 @@ from typing import Iterable, List, Optional
 import yaml
 
 from warden.routing.catalog import Catalog
+from warden.routing.models import attended_lane_issue
 
 VALUE_HANDLING = ("none", "generated_in_custody")
 REQUIRED_STEP_FIELDS = ("id", "owner", "entry_point", "auth", "warden_executes", "attended", "value_handling")
@@ -138,13 +139,10 @@ def check_follow_through(doc: dict, catalog: Optional[Catalog] = None) -> Follow
                 findings.append(Finding("catalog", "LANE-MISSING", where, f"lane '{lane}' is not in the catalog"))
             elif not entry.is_active or not entry.reviewed:
                 findings.append(Finding("catalog", "LANE-UNREVIEWED", where, f"lane '{lane}' is not active and reviewed"))
-            elif attended and entry.lane == "attended-change":
-                session = catalog.get(entry.session_lane or "")
-                if session is None or not session.is_active or not session.reviewed or session.lane != "login":
-                    findings.append(Finding("catalog", "LANE-KIND", where,
-                        f"attended-change lane '{lane}' session_lane '{entry.session_lane}' does not resolve to an active, reviewed login lane"))
-            elif attended and entry.lane != "login":
-                findings.append(Finding("catalog", "LANE-KIND", where, f"attended step's lane '{lane}' is not a login or attended-change lane"))
+            elif attended:
+                issue = attended_lane_issue(entry, catalog.get)
+                if issue:
+                    findings.append(Finding("catalog", issue[0], where, issue[1]))
 
         if executes:
             warden_steps.append(sid)

src/warden/routing/models.py

@@ -8,7 +8,7 @@ pointers only (the no-double-source rule, enforced in `catalog.py`).
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import List, Optional
+from typing import Callable, List, Optional
 
 
 # Risk grade vocabulary (ADR-0007). Grades outside LOW_RISK_GRADES — including
@@ -324,3 +324,20 @@ class RouteEntry:
             elif any(t in h or h in t for h in haystack):
                 score += 1
         return score
+
+
+def attended_lane_issue(
+    entry: RouteEntry, get_entry: Callable[[str], Optional[RouteEntry]]
+) -> Optional[tuple[str, str]]:
+    """Shared planner/executor boundary, using one parsed catalog snapshot."""
+    if not entry.is_active or not entry.reviewed:
+        return ("LANE-UNREVIEWED", f"lane '{entry.id}' is not active and reviewed")
+    if entry.lane == "attended-change":
+        session = get_entry(entry.session_lane or "")
+        if (session is None or not session.is_active or not session.reviewed
+                or session.lane != "login"):
+            return ("LANE-KIND", f"attended-change lane '{entry.id}' session_lane "
+                    f"'{entry.session_lane}' does not resolve to an active, reviewed login lane")
+    elif entry.lane != "login":
+        return ("LANE-KIND", f"attended step's lane '{entry.id}' is not a login or attended-change lane")
+    return None

tests/test_access.py

@@ -4,6 +4,8 @@ from __future__ import annotations
 import json
 from pathlib import Path
 
+import pytest
+
 from typer.testing import CliRunner
 
 from warden.access import agent_read_boundary_identity, expand_handoff, policy_gate_status
@@ -13,6 +15,34 @@ from warden.routing.models import RouteEntry
 runner = CliRunner()
 
 
+@pytest.fixture(autouse=True)
+def synthetic_openbao_advisory_identity(monkeypatch):
+    """Advisory CLI tests must not inspect or use workstation credentials.
+
+    Identity-display tests override this stub explicitly. The real helper has
+    separate subprocess-mocked unit coverage in test_openbao_caller.py.
+    """
+    monkeypatch.setattr(
+        "warden.openbao_caller.describe_ambient_openbao_caller",
+        lambda *args, **kwargs: {
+            "source": "none", "entity_id": None, "display_name": None,
+            "auth_path": None, "policies": [], "oidc_role": None, "ttl": None,
+            "error": "synthetic test: no ambient OpenBao token",
+        },
+    )
+
+
+def test_advisory_does_not_invoke_ambient_lookup(monkeypatch):
+    monkeypatch.setenv("BAO_TOKEN", "fixture-only-never-a-real-token")
+    def forbidden(*args, **kwargs):
+        raise AssertionError("advisory unit tests must not launch a token lookup")
+    monkeypatch.setattr("warden.openbao_caller.subprocess.run", forbidden)
+    result = runner.invoke(app, ["access", "npm token", "--json"])
+    assert result.exit_code == 0
+    assert json.loads(result.stdout)["openbao_caller"]["source"] == "none"
+
+
+
 def _repo_catalog() -> Path:
     return Path(__file__).resolve().parents[1] / "registry" / "routing" / "catalog.yaml"
 

tests/test_case_1_3_integration.py

@@ -1,10 +1,15 @@
 """Integration: the real net-kingdom 1.3 candidate against this kernel and catalog.
 
 Reads the actual 1.3 type, its forgejo example and the committed composition
-artifact from net-kingdom, plus this branch's real routing catalog. Only two things
-are simulated, both in memory: draft lanes flipped to active (promotion needs owner
-confirmation) and a 1.3 PARAMETER_PINS entry (PIN-1 waits for the founder's
-acknowledgment). Trust is fixture-only, and nothing is dispatched.
+artifact from net-kingdom, plus this branch's real routing catalog. The descriptor
+map and metadata come from those committed files. All authority is fixture-only:
+we change the proposed type to acknowledged, change the example case to
+materialized and replace its acceptance record, promote draft lanes in memory,
+and temporarily install the 1.3 parameter pin. Stub adapters replace native
+adapters; an unconditional evidence verifier simulates every acceptance and caller
+session. WARDEN_AGENT_ID is removed to simulate a human runtime. No adapter is
+dispatched. This proves conditional source admission, not authenticated owner
+acceptance, native execution or readiness; unresolved Apps sessions stay refused.
 """
 from dataclasses import asdict
 import hashlib
@@ -99,12 +104,15 @@ def test_artifact_descriptors_are_the_type_map():
     assert {s: digest(encode(d)) for s, d in descs.items()} == type_doc()["spec"]["execution_descriptors"]
 
 
-def test_all_seven_admitted_once_promoted_and_pinned(tmp_path):
+def test_conditional_admission_keeps_unresolved_apps_sessions_refused(tmp_path):
     for step, desc in descriptors().items():
-        admitted, invocation, _ = admit(tmp_path, desc)
-        assert admitted == desc
-        assert (invocation.step, invocation.actor, invocation.target) == (step, "human", "forgejo")
-
+        if step in ("deliver-consumer-secret", "hand-over"):
+            with pytest.raises(Refused, match="does not resolve to an active, reviewed login lane"):
+                admit(tmp_path, desc)
+        else:
+            admitted, invocation, _ = admit(tmp_path, desc)
+            assert admitted == desc
+            assert (invocation.step, invocation.actor, invocation.target) == (step, "human", "forgejo")
 
 def test_draft_lane_steps_refused_until_promoted(tmp_path):
     for step, desc in descriptors().items():

tests/test_case_admission_modes.py

@@ -53,6 +53,7 @@ def test_attended_mode_admits_human_parent_session(rig, monkeypatch):
     admitted = admit(
         rig,
         step=mode_step(False, True),
+        entries=[dict(lane_entry(), lane="login")],
         desc=dict(effect="change", revocation_scopes=("parent-session",)),
     )
     assert admitted[1].actor == "human"

tests/test_case_attended_boundary.py

@@ -0,0 +1,113 @@
+"""Synthetic attended admission/dispatch regressions; no owner/native actions."""
+from dataclasses import replace
+
+import pytest
+
+from tests.test_case_admission_modes import admit, mode_step
+from tests.test_case_run import fixture_catalog, lane_entry, make_rig
+from warden.case_receipt_state import Outcome, Refused
+from warden.case_type import check_follow_through
+from warden.routing.catalog import CatalogError, _parse_entry, load_catalog
+
+
+def attended(rig, entries):
+    return admit(rig, step=mode_step(False, True), entries=entries,
+                 desc=dict(effect="change", revocation_scopes=("parent-session",)))
+
+
+@pytest.mark.parametrize("operation", ["run", "rollback", "reconcile"])
+@pytest.mark.parametrize("marker", ["agent-1", "", None])
+def test_identity_change_refuses_before_any_write_or_callback(tmp_path, monkeypatch, operation, marker):
+    monkeypatch.delenv("WARDEN_AGENT_ID", raising=False)
+    rig = make_rig(tmp_path)
+    executor = rig[0]
+    admitted = attended(rig, [dict(lane_entry(), lane="login")])
+    executor.evidence_verifier = lambda *_: True
+    if operation == "rollback":
+        executor.run(*admitted)
+    elif operation == "reconcile":
+        old = executor.adapters[admitted[0].adapter]
+        def lost(_):
+            raise RuntimeError("synthetic lost reply")
+        executor.adapters[old.reference] = replace(old, invoke=lost)
+        executor.run(*admitted)
+    calls = []
+    old = executor.adapters[admitted[0].adapter]
+    callback = getattr(old, "invoke" if operation == "run" else operation)
+    def observed(*args):
+        calls.append(args)
+        return callback(*args)
+    executor.adapters[old.reference] = replace(old, **{
+        "invoke" if operation == "run" else operation: observed,
+    })
+    def rows():
+        return tuple(executor.journal.db.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
+                     for table in ("intents", "receipts"))
+    before = rows()
+    if marker is None:
+        result = getattr(executor, operation)(*admitted)
+        assert result.last.outcome is {"run": Outcome.SUCCESS,
+                                       "rollback": Outcome.ROLLED_BACK,
+                                       "reconcile": Outcome.NO_CHANGE}[operation]
+        assert len(calls) == 1
+        assert rows() != before
+        return
+    monkeypatch.setenv("WARDEN_AGENT_ID", marker)
+    with pytest.raises(Refused, match="attended step requires a human parent session"):
+        getattr(executor, operation)(*admitted)
+    assert rows() == before
+    assert not calls
+
+
+def entries_for(case):
+    change = dict(lane_entry(), lane="attended-change", session_lane="login-lane")
+    login = dict(lane_entry("login-lane"), lane="login")
+    if case == "missing":
+        return [change]
+    if case == "draft-login":
+        login["status"] = "draft"
+    elif case == "unreviewed-login":
+        login["reviewed"] = ""
+    elif case == "wrong-kind-login":
+        login["lane"] = "secret"
+    elif case == "draft-change":
+        change["status"] = "draft"
+    elif case == "unreviewed-change":
+        change["reviewed"] = ""
+    elif case == "missing-session-field":
+        del change["session_lane"]
+    elif case in ("secret", "ceremony"):
+        change["lane"] = case
+        del change["session_lane"]
+    return [change, login]
+
+
+@pytest.mark.parametrize("case, code", [
+    ("valid", None), ("missing", "LANE-KIND"), ("draft-login", "LANE-KIND"),
+    ("unreviewed-login", "LANE-KIND"), ("wrong-kind-login", "LANE-KIND"),
+    ("draft-change", "LANE-UNREVIEWED"), ("unreviewed-change", "LANE-UNREVIEWED"),
+    ("secret", "LANE-KIND"), ("ceremony", "LANE-KIND"),
+])
+def test_attended_change_planner_and_executor_agree(tmp_path, monkeypatch, case, code):
+    monkeypatch.delenv("WARDEN_AGENT_ID", raising=False)
+    entries = entries_for(case)
+    path = tmp_path / "catalog.yaml"
+    path.write_bytes(fixture_catalog(*entries))
+    catalog = load_catalog(path)
+    step = dict(mode_step(False, True)[0], attended_justification="synthetic test",
+                entry_point="owner/test", auth="synthetic session", value_handling="none")
+    doc = {"spec": {"follow_through": [step], "rollback": [{"reverses": step["id"]}]}}
+    report = check_follow_through(doc, catalog)
+    assert [f.code for f in report.findings] == ([] if code is None else [code])
+    rig = make_rig(tmp_path)
+    if code:
+        with pytest.raises(Refused):
+            attended(rig, entries)
+        assert rig[0].journal.db.execute("SELECT COUNT(*) FROM receipts").fetchone()[0] == 0
+    else:
+        assert attended(rig, entries)[1].lane == "issuer-lane"
+
+
+def test_attended_change_missing_session_is_parser_rejection():
+    with pytest.raises(CatalogError, match="must name session_lane"):
+        _parse_entry(entries_for("missing-session-field")[0], 0)

workplans/WARDEN-WP-0044-security-case-follow-through.md

@@ -310,3 +310,47 @@ suite passes 87 checks, including 14 new malformed-map integration cases; all
 four missing/changed modules still refuse. Review:
 `docs/reviews/2026-10-16-browser-producer-pin-refresh.md`. No adapter attestation,
 native acceptance, credential use or privileged dispatch. T03 remains progress.
+
+
+## T03 source correction and safe-test follow-through — 2026-10-17
+
+The reviewed KERNEL-2/3 source now authenticates and parses the actual catalog
+snapshot, checks lane owner/risk/delegation, and resolves each type step's Warden,
+attended or owner dispatch mode. Four proposed attended-change lanes remain
+DRAFT; no lane is promoted and PARAMETER_PINS still contains only 1.1/1.2.
+
+Independent review KR1-F1/F2 found cached attended run/rollback/reconcile could
+lose the human/runtime boundary, and actual admission lacked the planner's
+establishing-login checks. The exact successor retains the mode and authenticated
+catalog bytes and repeats caller/session, agent marker and attended-lane checks
+before any journal write or callback on all three operations. Planner and executor
+share active/reviewed login/session-lane validation. Two unresolved Apps session
+lanes correctly remain refused. Integration tests now explicitly disclose all
+fixture authority substitutions; they establish no owner or native authority.
+
+The new 19-test regression matrix reproduces 13 failures on its predecessor and
+passes on the candidate. A fresh independent reviewer adds changed-authority and
+catalog-snapshot probes: 90 focused checks and six guarded integration checks
+pass; the combined predecessor comparison has 22 expected failures. Author's
+isolated full suite passes 765 with four integration-marked tests deselected.
+Evidence: Custodian warden-1.3-kernel-fixes-r1 and warden-1.3-kernel-review-r2
+under docs/changes/netkingdom-target-2026-10-17. Original unsuccessful review,
+invalid setup attempts and unknown ambient-lookup outcome are retained.
+
+A separate reviewed test-only fix stubs ambient OpenBao identity in access CLI
+unit tests and adds a synthetic sentinel regression which fails before any
+subprocess can launch on the predecessor; 26 access/helper checks pass. The real
+production helper is unchanged. The first old review's unisolated launches may
+have invoked ambient `bao token lookup`; remote reach/success remain unknown,
+with no secret displayed. This patch does not erase that adverse outcome.
+
+Explicit remaining safe-test work is retained in this live T03 scope, owned by
+ops-warden: replace or wrap the ordinary unit-test entry path with allowlisted
+environment/empty synthetic HOME and guarded subprocess/network execution,
+including nested validator children. Completion requires representative access
+and case tests to refuse token-file/environment and live provider access under
+synthetic negative controls while permitted local fixtures remain usable. Until
+then use the preserved isolated review runner, not the ambient `make test` path.
+No credential lookup, live investigation or rotation is authorized by this note.
+T03 remains progress; contained sessions, authenticated owner provenance,
+materialization, exact type acknowledgment/pin and first native case remain open.