fix(todo): align claim agent-id folding and rejection exception contract

Two caller-observable compatibility gaps from the todo-migration bug
hunt (r2), both Low severity, fixed on the TypeScript-owner side to
match the migration direction of #3972/#4000/#4005.

Agent id normalization: the Python kernel's normalize_todo_claimed_by
folds every whitespace run (tabs included) into a single "-", while
normalizeTodoAgent only replaced literal spaces, so a claim typed as
"agent\ta" succeeded before promotion and failed with 'claimed_by must
be a public-safe agent id' after it. normalizeTodoAgent now folds \s+
runs the same way. Python stays untouched: its normalizer feeds 30+
call sites (agent registry, thread bindings, event-sourced state)
whose legacy behavior must not change.

Claim rejection exception contract: promoted rejections raised
LocalCoordinationAuthorityUnavailable (RuntimeError) while the legacy
kernel raised ValueError, so external Python API callers using
'except ValueError' stopped catching claim rejections after cutover.
Decision rejections (status="failed" results from the TypeScript
owner) now raise LocalCoordinationAuthorityRejection, a ValueError
that remains an unavailable subclass so already-migrated callers keep
working. Protocol failures
(invalid_local_coordination_todo_claim_request) and infrastructure
outages still raise the plain unavailable error, which is not a
ValueError.

Tests: TS conformance folds tab ids end-to-end through the promoted
claim path; Python covers the ValueError contract (mocked and real
TypeScript-runtime rejections), the protocol-failure control, and a
real promoted claim with claimed_by "Agent\tA" normalizing to
agent-a.

Signed-off-by: now-ing <now-ing@users.noreply.github.com>

Author: now-ing

loopx/control_plane/coordination/local_authority.py

@@ -43,6 +43,23 @@ def __init__(self, message: str, *, code: str, payload: Mapping[str, Any]) -> No
         self.payload = dict(payload)
 
 
+class LocalCoordinationAuthorityRejection(LocalCoordinationAuthorityUnavailable, ValueError):
+    """The TypeScript coordination owner definitively rejected a claim.
+
+    The legacy Python kernel raised ``ValueError`` for every claim rejection
+    (todo_not_open, claim_owner_mismatch, unregistered actor, ...).  After
+    promotion those rejections surface as ``status="failed"`` results from the
+    TypeScript transaction owner; re-raising them through this class keeps the
+    legacy ``except ValueError`` contract intact for Python API callers while
+    remaining catchable as an authority outage.  Infrastructure and protocol
+    failures keep raising :class:`LocalCoordinationAuthorityUnavailable`, which
+    is not a ``ValueError``.
+    """
+
+    def __init__(self, message: str, *, code: str, payload: Mapping[str, Any]) -> None:
+        super().__init__(message, code=code, payload=payload)
+
+
 def local_authority_is_promoted(*, runtime_root: Path, goal_id: str) -> bool:
     fence_path = legacy_coordination_writer_fence_path(
         runtime_root=runtime_root,
@@ -103,6 +120,19 @@ def claim_canonical_todo_if_promoted(
         )
     payload = dict(result)
     accepted = {"applied", "recovered", "replayed", "no_change", "planned"}
+    if (
+        payload.get("status") == "failed"
+        and payload.get("reason_code") != "invalid_local_coordination_todo_claim_request"
+    ):
+        # The TypeScript owner returned a definitive decision rejection. The
+        # legacy kernel raised ValueError for the same rejections, so keep that
+        # caller-observable contract; the wrapper error code is a protocol
+        # failure and stays an infrastructure outage instead.
+        raise LocalCoordinationAuthorityRejection(
+            str(payload.get("reason") or "canonical Todo claim was rejected"),
+            code=str(payload.get("reason_code") or "claim_rejected"),
+            payload=payload,
+        )
     if (
         payload.get("status") not in accepted
         or payload.get("source_authority") != "file_v0"

loopx/control_plane/coordination/todo_agents.ts

@@ -4,7 +4,10 @@ export function normalizeTodoAgent(value: unknown, label: string): string {
   if (typeof value !== "string") {
     throw new AuthorityStoreProtocolError(`${label} must be a public-safe agent id`);
   }
-  const candidate = value.trim().toLowerCase().replaceAll(" ", "-");
+  // Collapse every whitespace run to one "-" so ids typed with tabs or other
+  // whitespace fold exactly like the Python kernel's compact_todo_text path
+  // (loopx/control_plane/todos/contract.py normalize_todo_claimed_by).
+  const candidate = value.trim().toLowerCase().replace(/\s+/gu, "-");
   if (!/^[a-z][a-z0-9_.:@-]{0,79}$/u.test(candidate)) {
     throw new AuthorityStoreProtocolError(`${label} must be a public-safe agent id`);
   }

tests/control_plane/test_local_coordination_authority.py

@@ -10,6 +10,7 @@
 import pytest
 
 from loopx.control_plane.coordination.local_authority import (
+    LocalCoordinationAuthorityRejection,
     LocalCoordinationAuthorityUnavailable,
     claim_canonical_todo_if_promoted,
     read_canonical_todos_if_promoted,
@@ -250,6 +251,228 @@ def test_engaged_fence_never_falls_back_when_provider_is_missing(
     assert exc_info.value.code == "local_authority_todo_list_unavailable"
 
 
+def _claim_registry(tmp_path: Path) -> Path:
+    registry = tmp_path / "registry.json"
+    registry.write_text(
+        json.dumps(
+            {
+                "schema_version": 1,
+                "goals": [
+                    {
+                        "id": "goal-a",
+                        "coordination": {"registered_agents": ["agent-a"]},
+                    }
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+    return registry
+
+
+def _seed_promoted_store(runtime_root: Path) -> None:
+    """Promote one open agent Todo through the real TypeScript runtime."""
+
+    projection = build_todo_runtime_shadow_projection(
+        goal_id="goal-a",
+        todos=[
+            {
+                "schema_version": "todo_item_v0",
+                "index": 1,
+                "done": False,
+                "text": "Claim through the promoted provider head",
+                "todo_id": "todo_a",
+                "role": "agent",
+                "status": "open",
+                "archive_state": "active",
+                "source_section": TODO_SECTION_HEADINGS["agent"],
+            }
+        ],
+    )
+    canonical_bytes = json.dumps(
+        projection,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode("utf-8")
+    projection_sha256 = hashlib.sha256(canonical_bytes).hexdigest()
+    bootstrap = effect_runtime_result(
+        "coordination.runtime_shadow.bootstrap",
+        {
+            "schema_version": "loopx_coordination_runtime_shadow_bootstrap_v0",
+            "runtime_root": str(runtime_root),
+            "goal_id": "goal-a",
+            "operation_id": "bootstrap:goal-a:f34",
+            "source_version": "state:f34:0",
+            "projection": projection,
+        },
+    )
+    assert bootstrap["status"] == "applied"
+    mirrored = effect_runtime_result(
+        "coordination.runtime_shadow.commit",
+        {
+            "schema_version": "loopx_coordination_runtime_shadow_commit_v0",
+            "runtime_root": str(runtime_root),
+            "goal_id": "goal-a",
+            "operation_id": "todo:goal-a:f34:qualify",
+            "event_kind": "todo_update",
+            "source_version": "state:f34:1",
+            "projection": projection,
+        },
+    )
+    assert mirrored["status"] == "applied"
+    provider_revision = str(mirrored["provider_revision"])
+    fence = {
+        "schema_version": "loopx_legacy_coordination_writer_fence_v0",
+        "state": "engaged",
+        "goal_id": "goal-a",
+        "fence_id": "legacy-writer-fence:goal-a:f34",
+        "source_version": "state:f34:1",
+        "source_projection_sha256": projection_sha256,
+        "expected_shadow_provider_revision": provider_revision,
+    }
+    engaged = effect_runtime_result(
+        "coordination.local_authority.legacy_writer_fence.engage",
+        {
+            "schema_version": "loopx_legacy_coordination_writer_fence_engage_request_v0",
+            "runtime_root": str(runtime_root),
+            "goal_id": "goal-a",
+            "fence": fence,
+        },
+    )
+    assert engaged["status"] == "applied"
+    promoted = effect_runtime_result(
+        "coordination.local_authority.promote",
+        {
+            "schema_version": "loopx_local_coordination_promotion_request_v0",
+            "runtime_root": str(runtime_root),
+            "goal_id": "goal-a",
+            "operation_id": "promote:goal-a:f34",
+            "expected_shadow_provider_revision": provider_revision,
+            "expected_shadow_projection_sha256": projection_sha256,
+            "minimum_operations": 1,
+            "required_event_kinds": ["todo_update"],
+            "writer_fence": fence,
+        },
+    )
+    assert promoted["status"] == "applied"
+
+
+def test_promoted_claim_rejection_preserves_legacy_valueerror_contract(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    """Promoted claim rejections must stay catchable via ``except ValueError``.
+
+    The legacy kernel raised ValueError for decision rejections such as
+    todo_not_open; external Python API callers rely on that contract.
+    """
+    _engage_fence(tmp_path)
+    monkeypatch.setattr(
+        "loopx.control_plane.coordination.local_authority.effect_runtime_result",
+        lambda method, params: {
+            "status": "failed",
+            "reason_code": "todo_not_open",
+            "reason": "todo claim requires status=open",
+            "source_authority": "file_v0",
+            "decision_read_from_provider": True,
+            "legacy_fallback_used": False,
+        },
+    )
+    with pytest.raises(ValueError) as exc_info:
+        claim_canonical_todo_if_promoted(
+            registry_path=_claim_registry(tmp_path),
+            runtime_root=tmp_path,
+            goal_id="goal-a",
+            todo_id="todo_a",
+            role="agent",
+            claimed_by="agent-a",
+            actor_agent_id="agent-a",
+            dry_run=False,
+        )
+    rejection = exc_info.value
+    assert isinstance(rejection, LocalCoordinationAuthorityRejection)
+    # Callers that already migrated to the authority-unavailable handling of
+    # promoted claims keep working: the rejection is still its subclass.
+    assert isinstance(rejection, LocalCoordinationAuthorityUnavailable)
+    assert rejection.code == "todo_not_open"
+    assert str(rejection) == "todo claim requires status=open"
+
+
+def test_promoted_claim_protocol_failure_stays_infrastructure_outage(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    """Protocol-level request failures must not masquerade as ValueErrors."""
+    _engage_fence(tmp_path)
+    monkeypatch.setattr(
+        "loopx.control_plane.coordination.local_authority.effect_runtime_result",
+        lambda method, params: {
+            "status": "failed",
+            "reason_code": "invalid_local_coordination_todo_claim_request",
+            "reason": "registered_agents must be a JSON array",
+            "source_authority": "file_v0",
+            "decision_read_from_provider": True,
+            "legacy_fallback_used": False,
+        },
+    )
+    with pytest.raises(LocalCoordinationAuthorityUnavailable) as exc_info:
+        claim_canonical_todo_if_promoted(
+            registry_path=_claim_registry(tmp_path),
+            runtime_root=tmp_path,
+            goal_id="goal-a",
+            todo_id="todo_a",
+            role="agent",
+            claimed_by="agent-a",
+            actor_agent_id="agent-a",
+            dry_run=False,
+        )
+    assert exc_info.value.code == "invalid_local_coordination_todo_claim_request"
+    assert not isinstance(exc_info.value, ValueError)
+    assert not isinstance(exc_info.value, LocalCoordinationAuthorityRejection)
+
+
+def test_promoted_claim_folds_agent_id_whitespace_like_legacy(tmp_path: Path) -> None:
+    """Tabs in claimed_by must fold to "-" before and after promotion alike.
+
+    Legacy normalize_todo_claimed_by collapses any whitespace run (Python
+    compact_todo_text) to a single "-"; the TypeScript owner now folds the
+    same way, so the identical claim command keeps succeeding post-cutover.
+    """
+    _seed_promoted_store(tmp_path)
+    result = claim_canonical_todo_if_promoted(
+        registry_path=_claim_registry(tmp_path),
+        runtime_root=tmp_path,
+        goal_id="goal-a",
+        todo_id="todo_a",
+        role="agent",
+        claimed_by="Agent\tA",
+        actor_agent_id=None,
+        dry_run=False,
+    )
+    assert result is not None and result["ok"] is True
+    assert result["status"] == "applied"
+    assert result["claimed_by"] == "agent-a"
+
+
+def test_promoted_claim_missing_todo_rejection_is_valueerror(tmp_path: Path) -> None:
+    """End-to-end: a real TypeScript decision rejection raises ValueError."""
+    _seed_promoted_store(tmp_path)
+    with pytest.raises(ValueError) as exc_info:
+        claim_canonical_todo_if_promoted(
+            registry_path=_claim_registry(tmp_path),
+            runtime_root=tmp_path,
+            goal_id="goal-a",
+            todo_id="todo_missing",
+            role="agent",
+            claimed_by="agent-a",
+            actor_agent_id="agent-a",
+            dry_run=False,
+        )
+    assert isinstance(exc_info.value, LocalCoordinationAuthorityRejection)
+    assert exc_info.value.code == "todo_not_found"
+
+
 def test_todo_list_uses_provider_after_cutover_even_when_markdown_disagrees(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,

tests/control_plane_ts/local_authority_runtime.test.ts

@@ -512,6 +512,52 @@ test("provider-first Todo claim preserves the complete record and is replay-safe
   assert.equal(repeated.changed, false);
 });
 
+test("agent id normalization folds any whitespace run like the Python kernel", async () => {
+  // Parity with loopx/control_plane/todos/contract.py normalize_todo_claimed_by:
+  // compact_todo_text collapses every whitespace run (tabs included) into one
+  // space before mapping it to "-", so the same claim command keeps working
+  // before and after promotion.
+  const root = await mkdtemp(join(tmpdir(), "loopx-local-authority-claim-tab-"));
+  const store = new FileAuthorityStore(join(root, "authority", "file-v0"), "goal-a");
+  const seeded = await store.commitAuthority({
+    expected_provider_revision: null,
+    operation_id: "promote:claim-tab-test",
+    events: [{ schema_version: "promotion_v0" }],
+    next_projection: withTodoReadModel({
+      goal_id: "goal-a",
+      handoff_mode: "soft_claim",
+      todos: [todoRecord()],
+      leases: [],
+    }),
+    receipts: [],
+  });
+  assert.equal(seeded.status, "applied");
+
+  const applied = await claimLocalCoordinationTodo({
+    schema_version: LOCAL_COORDINATION_TODO_CLAIM_REQUEST_SCHEMA,
+    runtime_root: root,
+    goal_id: "goal-a",
+    todo_id: "todo_a",
+    role: "agent",
+    claimed_by: "Agent\tA",
+    actor_agent_id: "Agent \t A",
+    registered_agents: ["agent-a"],
+    operation_id: "todo-claim:goal-a:todo_a:tab",
+    observed_at: "2026-09-05T04:30:00Z",
+    dry_run: false,
+  });
+  assert.equal(applied.status, "applied", JSON.stringify(applied));
+
+  const read = await readLocalCoordinationTodo({
+    schema_version: LOCAL_COORDINATION_TODO_READ_REQUEST_SCHEMA,
+    runtime_root: root,
+    goal_id: "goal-a",
+    todo_id: "todo_a",
+  });
+  assert.equal(read.status, "found");
+  assert.equal((read.todo as Record<string, unknown>).claimed_by, "agent-a");
+});
+
 test("one TypeScript decision owns promoted and legacy Todo claims", () => {
   const input = {
     goal_id: "goal-a",