From 37f999c101cc0d90b2afb731dc6a47288ce2701d Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Thu, 8 Oct 2026 16:30:07 -0700
Subject: [PATCH 1/7] Break AMW gateway audit chain

Audit fix by Muse Code fleet instance brk-gw-audit (Grok Bot reviews before PR).
---
 BREAK-REPORT.md                               |  88 +++++
 gateway/app/services/audit_chain.py           | 248 ++++++++-----
 gateway/tests/test_audit_chain_tamper.py      | 338 ++++++++++++++++++
 .../framework_integrations/README.autogen.md  |  32 +-
 .../README.langgraph.md                       |  24 +-
 .../README.llamaindex.md                      |   6 +-
 6 files changed, 596 insertions(+), 140 deletions(-)
 create mode 100644 BREAK-REPORT.md
 create mode 100644 gateway/tests/test_audit_chain_tamper.py

diff --git a/BREAK-REPORT.md b/BREAK-REPORT.md
new file mode 100644
index 00000000..b4782b74
--- /dev/null
+++ b/BREAK-REPORT.md
@@ -0,0 +1,88 @@
+# Audit chain break report
+
+4 breaks found. 4 fixed and tested. 0 fixed but failing. 0 blocked.
+
+Middle-link deletion was already rejected (`audit_previous_hash_mismatch`). That case is covered by a test and is not counted above.
+
+Local SQLite tests only. No production calls.
+
+## High: a time window hid a deleted tail
+
+What broke: delete the last audit event for a wallet, then verify with any time window that still covers the survivors (a start date in the past, an end date in the future, or both). The answer was valid. The same deletion with no time window was already reported as `audit_chain_truncated`.
+
+How to reproduce: record three events, delete the last row, call `verify_audit_chain` with `created_after=2000-01-01` or `created_before=2099-01-01`.
+
+Root cause: `gateway/app/services/audit_chain.py` loaded the chain head only when both time filters were absent. A window skipped the anchor that detects a missing tail. The head is now loaded on every verify (snapshot at `audit_chain.py:508`, tail check at `audit_chain.py:667`).
+
+Fix: one snapshot loads every event and the head. Time filters choose which links get a hash walk. The tail is still compared to the head.
+
+Test: `test_time_window_does_not_hide_a_deleted_tail` (three windows). Before the fix each window returned `valid=True` with 2 events. After the fix, `audit_chain_truncated`.
+
+## High: sequence holes and duplicate sequence numbers verified as valid
+
+What broke: `seq` is not part of the signature. Renumber the tail from 3 to 4 and set the head's `last_seq` to 4. Hash order and the head hash still match, so verify returned valid for a chain whose sequences are 1, 2, 4. Collapsing the tail onto the previous sequence (1, 2, 2) and pointing the head at that sequence also returned valid.
+
+How to reproduce: record three events, update the last row's `seq` and the head's `last_seq` as above, call `verify_audit_chain`.
+
+Root cause: `gateway/app/services/audit_chain.py` compared only the last event's seq and chain hash to the head. It never required the sequences to be 1, 2, 3, and so on up to that head. The check is `_first_sequence_break` at `audit_chain.py:70`, applied at `audit_chain.py:679`.
+
+Fix: after the hash walk, if the head row exists, reject the chain unless the stored sequences are exactly 1 through `last_seq`. Reason: `audit_sequence_gap`. A time window does not skip this check.
+
+Test: `test_renumbered_sequence_gap_is_rejected` and `test_duplicate_sequence_is_rejected`. Before the fix both returned `valid=True`. After the fix, `audit_sequence_gap`. The windowed form of the gap test fails the same way.
+
+## High: a missing head looked like a complete chain, and the next append split it
+
+What broke: delete the `audit_chain_heads` row and leave the events. Verify returned valid. The following append treated the wallet as empty, signed the new event as a genesis (no previous hash, seq 1), and stored a new head. The older links then failed verification.
+
+How to reproduce: record two events, delete the head row, call `verify_audit_chain`, then `record_audit_event` again.
+
+Root cause: `gateway/app/services/audit_chain.py` ran the tail check only when a head row was present, and `append_chained_audit_event` inserted a new head whenever the select found none (`audit_chain.py:317`).
+
+Fix: a modern chain (any seq other than 0) with events and no head fails verify with `audit_chain_head_missing` (`audit_chain.py:654`). Append raises `AuditChainHeadMissingError` with that same message and does not insert a row (`audit_chain.py:328`). Chains whose rows are still the old default seq 0, and which have no head, still verify.
+
+Test: `test_missing_head_on_a_modern_chain_is_rejected` and `test_append_refuses_to_start_a_new_genesis_when_the_head_is_gone`. Before the fix, verify returned `valid=True` and the append did not raise. After the fix, verify returns `audit_chain_head_missing`, the append raises `AuditChainHeadMissingError`, and the event count stays 2.
+
+## High: a blank wallet id split the wallet-less chain
+
+What broke: an event with `wallet_id=""` and an event with `wallet_id=None` share one head key (`""`). Verify walked them as two chains. Global verify then reported `audit_chain_truncated` and counted one event.
+
+How to reproduce: `record_audit_event(..., wallet_id=None)` then `record_audit_event(..., wallet_id="")`, then `verify_audit_chain(wallet_id=None)`.
+
+Root cause: `append_chained_audit_event` uses `wallet_id or ""` as the head key, while verify filtered `wallet_id IS NULL` and `wallet_id == ""` separately. The shared filter is `_chain_wallet_filter` at `audit_chain.py:57`. Global wallet ids fold `""` into the null chain at `audit_chain.py:436`.
+
+Fix: NULL and blank wallet ids are one chain on verify, including the predecessor link and the head. Stored wallet ids are unchanged.
+
+Test: `test_blank_wallet_id_shares_the_wallet_less_chain`. Before the fix, global verify returned `valid=False`, reason `audit_chain_truncated`, `checked_events=1`. After the fix, global verify and `verify_audit_chain(wallet_id="")` both return valid with `checked_events=2`.
+
+## Tests
+
+New file: `gateway/tests/test_audit_chain_tamper.py`.
+
+Before any source change:
+
+```
+cd gateway && ~/metacode/fleet/heavy .venv/bin/pytest -q tests/test_audit_chain_tamper.py --tb=line -p no:cacheprovider
+```
+
+Result: `8 failed, 3 passed in 1.74s`. Failures were the sequence gap, the duplicate sequence, all three time windows, the missing head, the append, and the blank wallet. The three passes were middle-link deletion (already rejected), an empty window on a healthy chain, and a seq 0 chain with no head.
+
+After the fix:
+
+```
+cd gateway && ~/metacode/fleet/heavy .venv/bin/ruff check app/services/audit_chain.py tests/test_audit_chain_tamper.py
+cd gateway && ~/metacode/fleet/heavy .venv/bin/pytest -q tests/test_audit_chain_tamper.py tests/test_audit_chain.py tests/test_audit_snapshot.py tests/test_audit_chain_contention_surface.py --tb=short -p no:cacheprovider
+```
+
+Ruff: `All checks passed!`
+
+Pytest: `35 passed in 13.44s`.
+
+A later run of `tests/test_audit_chain_tamper.py` alone, after the append test was pointed at `AuditChainHeadMissingError`, was `11 passed in 0.44s`.
+
+## Open questions
+
+Rewinding the head to an earlier contiguous prefix is still invisible. Delete the tail, set `last_seq` and `last_chain_hash` to the new last event, and leave sequences 1..k untouched. The rows look like the chain stopped there. Seeing that requires an anchor outside this database. This change does not add one.
+
+A chain that is still all seq 0 and has no head still verifies. The next append on that chain now raises `AuditChainHeadMissingError` instead of signing a new genesis. If any deployment still has those rows, audit writes for that wallet stop until a head is repaired.
+
+`AuditChainHeadMissingError` is not the contention error (`audit_chain_head_contention`). A request that hits it is not retried as a busy chain.
diff --git a/gateway/app/services/audit_chain.py b/gateway/app/services/audit_chain.py
index 36156782..e1ad77c4 100644
--- a/gateway/app/services/audit_chain.py
+++ b/gateway/app/services/audit_chain.py
@@ -6,7 +6,7 @@
 from datetime import datetime
 from typing import Any, cast
 
-from sqlalchemy import asc, desc, literal, select, true, update
+from sqlalchemy import asc, desc, literal, or_, select, true, update
 from sqlalchemy.engine import CursorResult
 from sqlalchemy.exc import IntegrityError, OperationalError
 from sqlalchemy.sql.elements import ColumnElement
@@ -54,6 +54,37 @@ def audit_payload(
     }
 
 
+def _chain_wallet_filter(wallet_id: str | None) -> ColumnElement[bool]:
+    """Rows that belong to one wallet chain.
+
+    The head table stores "" for a missing wallet id. A blank string is
+    the same chain as NULL. Walking those rows apart made each half look
+    truncated or unlinked.
+    """
+    column = ControlPlaneAuditEventModel.wallet_id
+    if wallet_id:
+        return cast(ColumnElement[bool], column == wallet_id)
+    return cast(ColumnElement[bool], or_(column.is_(None), column == ""))
+
+
+def _first_sequence_break(
+    events: list[ControlPlaneAuditEventModel],
+    last_seq: int,
+) -> str | None:
+    """Event id where ``seq`` stops being 1..last_seq, in chain order.
+
+    ``seq`` is not signed. A stored tail can be renumbered so it still
+    matches the head while a hole or a duplicate sits in front of it.
+    ``events`` must already be ordered by seq, then created_at.
+    """
+    for index, event in enumerate(events, start=1):
+        if index > last_seq or event.seq != index:
+            return event.event_id
+    if len(events) != last_seq:
+        return events[-1].event_id if events else None
+    return None
+
+
 def _sign_with_previous(
     model: ControlPlaneAuditEventModel,
     previous_hash: str | None,
@@ -113,12 +144,7 @@ async def sign_audit_model(model: ControlPlaneAuditEventModel) -> None:
     async with factory() as session:
         result = await session.execute(
             select(ControlPlaneAuditEventModel)
-            .where(
-                cast(
-                    ColumnElement[bool],
-                    ControlPlaneAuditEventModel.wallet_id == model.wallet_id,
-                )
-            )
+            .where(_chain_wallet_filter(model.wallet_id))
             .order_by(
                 desc(cast(ColumnElement[Any], ControlPlaneAuditEventModel.seq)),
                 desc(cast(ColumnElement[Any], ControlPlaneAuditEventModel.created_at)),
@@ -139,6 +165,16 @@ class AuditEventConflictError(RuntimeError):
     """A caller reused an audit event identity for different signed evidence."""
 
 
+class AuditChainHeadMissingError(RuntimeError):
+    """Events are stored for this chain but the head row is gone.
+
+    Appending would sign a new genesis (empty previous hash, seq 1) and
+    split the links that are already there. Callers get this instead.
+    """
+
+    reason = "audit_chain_head_missing"
+
+
 class AuditChainContendedError(RuntimeError):
     """The per-wallet chain head stayed contended for the whole retry budget.
 
@@ -278,6 +314,18 @@ async def append_chained_audit_event(
                 ).first()
                 observed_seq = row[0] if row else 0
                 previous_hash = row[1] if row else None
+                if row is None:
+                    # No head does not mean an empty chain. A genesis signed
+                    # over rows that are already stored splits the hash links.
+                    prior = (
+                        await session.execute(
+                            select(ControlPlaneAuditEventModel.event_id)
+                            .where(_chain_wallet_filter(model.wallet_id))
+                            .limit(1)
+                        )
+                    ).first()
+                    if prior is not None:
+                        raise AuditChainHeadMissingError("audit_chain_head_missing")
                 model.seq = observed_seq + 1
                 _sign_with_previous(model, previous_hash, signing_key_id=key.key_id)
                 now = utc_now()
@@ -383,7 +431,11 @@ async def verify_audit_chain(
             wallet_ids_result = await session.execute(
                 select(cast(Any, ControlPlaneAuditEventModel.wallet_id)).distinct()
             )
-            event_wallets = {row[0] for row in wallet_ids_result.all()}
+            # Blank wallet ids share the "" head with NULL. Fold them into
+            # one entry so this walk does not count that chain twice.
+            event_wallets = {
+                (None if not row[0] else row[0]) for row in wallet_ids_result.all()
+            }
             # Also enumerate wallets from the chain-head table. Deriving the list
             # from events alone let an attacker who deleted ALL of a wallet's
             # events erase that wallet from the check entirely: with no events it
@@ -435,105 +487,79 @@ async def _verify_single_chain(
     created_after: datetime | None = None,
     created_before: datetime | None = None,
 ) -> AuditChainVerification:
-    """Verify one wallet's audit chain (``wallet_id=None`` -> the wallet-less
-    chain of ``wallet_id IS NULL`` events)."""
+    """Verify one wallet's audit chain.
+
+    ``wallet_id=None`` and ``wallet_id=""`` are the wallet-less chain
+    (NULL and blank ids together).
+    """
     created_after = to_naive_utc(created_after) if created_after else None
     created_before = to_naive_utc(created_before) if created_before else None
-    stmt = select(ControlPlaneAuditEventModel).order_by(
-        asc(cast(ColumnElement[Any], ControlPlaneAuditEventModel.seq)),
-        asc(cast(ColumnElement[Any], ControlPlaneAuditEventModel.created_at)),
-    )
-    if wallet_id is None:
-        stmt = stmt.where(cast(Any, ControlPlaneAuditEventModel.wallet_id).is_(None))
-    else:
-        stmt = stmt.where(
-            cast(
-                ColumnElement[bool], ControlPlaneAuditEventModel.wallet_id == wallet_id
-            )
-        )
-    if created_after:
-        stmt = stmt.where(
-            cast(
-                ColumnElement[bool],
-                ControlPlaneAuditEventModel.created_at >= created_after,
-            )
-        )
-    if created_before:
-        stmt = stmt.where(
-            cast(
-                ColumnElement[bool],
-                ControlPlaneAuditEventModel.created_at <= created_before,
-            )
+    stmt = (
+        select(ControlPlaneAuditEventModel)
+        .where(_chain_wallet_filter(wallet_id))
+        .order_by(
+            asc(cast(ColumnElement[Any], ControlPlaneAuditEventModel.seq)),
+            asc(cast(ColumnElement[Any], ControlPlaneAuditEventModel.created_at)),
         )
+    )
 
     factory = get_session_factory()
     async with factory() as session:
-        head = None
-        if created_after is None and created_before is None:
-            # One statement gives events and head the same database snapshot.
-            # A separate head SELECT could see a legitimate concurrent append
-            # and label the earlier event snapshot as a truncated chain.
-            # Anchor the outer joins to one row so an empty/deleted event set
-            # still loads its head and actual truncation remains detectable.
-            anchor = select(literal(1).label("anchor")).subquery()
-            event_rows = aliased(ControlPlaneAuditEventModel, stmt.subquery())
-            head_rows = aliased(
-                AuditChainHeadModel,
-                select(AuditChainHeadModel)
-                .where(
-                    cast(
-                        ColumnElement[bool],
-                        AuditChainHeadModel.wallet_key == (wallet_id or ""),
-                    )
+        # One statement gives events and head the same database snapshot.
+        # A separate head SELECT could see a legitimate concurrent append
+        # and label the earlier event snapshot as a truncated chain.
+        # Time filters used to skip this load, so a wide window hid a
+        # deleted tail. Apply those filters in memory after the snapshot.
+        # Anchor the outer joins to one row so an empty or deleted event
+        # set still loads its head and truncation remains detectable.
+        anchor = select(literal(1).label("anchor")).subquery()
+        event_rows = aliased(ControlPlaneAuditEventModel, stmt.subquery())
+        head_rows = aliased(
+            AuditChainHeadModel,
+            select(AuditChainHeadModel)
+            .where(
+                cast(
+                    ColumnElement[bool],
+                    AuditChainHeadModel.wallet_key == (wallet_id or ""),
                 )
-                .subquery(),
             )
-            snapshot = await session.execute(
-                select(event_rows, head_rows)
-                .select_from(anchor)
-                .outerjoin(event_rows, true())
-                .outerjoin(head_rows, true())
-                .order_by(
-                    cast(ColumnElement[Any], event_rows.seq),
-                    cast(ColumnElement[Any], event_rows.created_at),
-                )
+            .subquery(),
+        )
+        snapshot = await session.execute(
+            select(event_rows, head_rows)
+            .select_from(anchor)
+            .outerjoin(event_rows, true())
+            .outerjoin(head_rows, true())
+            .order_by(
+                cast(ColumnElement[Any], event_rows.seq),
+                cast(ColumnElement[Any], event_rows.created_at),
             )
-            rows = snapshot.all()
-            events = [row[0] for row in rows if row[0] is not None]
-            head = rows[0][1] if rows else None
-        else:
-            result = await session.execute(stmt)
-            events = list(result.scalars().all())
+        )
+        rows = snapshot.all()
+        all_events = [row[0] for row in rows if row[0] is not None]
+        head = rows[0][1] if rows else None
+
+    if created_after is not None or created_before is not None:
+        events = [
+            event
+            for event in all_events
+            if (created_after is None or event.created_at >= created_after)
+            and (created_before is None or event.created_at <= created_before)
+        ]
+    else:
+        events = all_events
 
     # A window that excludes the wallet's genesis event starts mid-chain: the
-    # first in-window event legitimately links to an event *outside* the
+    # first in-window event legitimately links to an event outside the
     # window. Seed the expected predecessor from the last stored event before
     # the window instead of from genesis, otherwise every non-genesis window
-    # reports a valid chain as tampered (``audit_previous_hash_mismatch``).
+    # reports a valid chain as tampered (audit_previous_hash_mismatch).
     previous_hash: str | None = None
     if events and created_after:
         first_seq = events[0].seq
-        pred_stmt = (
-            select(cast(Any, ControlPlaneAuditEventModel.chain_hash))
-            .where(
-                cast(ColumnElement[bool], ControlPlaneAuditEventModel.seq < first_seq)
-            )
-            .order_by(desc(cast(ColumnElement[Any], ControlPlaneAuditEventModel.seq)))
-            .limit(1)
-        )
-        if wallet_id is None:
-            pred_stmt = pred_stmt.where(
-                cast(Any, ControlPlaneAuditEventModel.wallet_id).is_(None)
-            )
-        else:
-            pred_stmt = pred_stmt.where(
-                cast(
-                    ColumnElement[bool],
-                    ControlPlaneAuditEventModel.wallet_id == wallet_id,
-                )
-            )
-        async with factory() as session:
-            previous_hash = (await session.execute(pred_stmt)).scalar_one_or_none()
+        below = [event for event in all_events if event.seq < first_seq]
+        if below:
+            previous_hash = below[-1].chain_hash
     first_event_id = events[0].event_id if events else None
     last_event_id = events[-1].event_id if events else None
     for event in events:
@@ -619,23 +645,47 @@ async def _verify_single_chain(
             )
         previous_hash = event.chain_hash
 
-    # Tail-truncation check: the last verified event must match the head the
-    # append path recorded. If the head says seq=N/chain_hash=H but the loaded
-    # chain ends earlier (or is empty), the tail was deleted.
-    if head is not None:
-        last_event = events[-1] if events else None
+    # The hash walk above covers the requested window only. The head and
+    # the sequence belong to the whole chain. Checking them on the window,
+    # or skipping them when the head row is gone, let a deleted tail or a
+    # renumbered seq verify as a complete chain.
+    anchor_first = all_events[0].event_id if all_events else None
+    anchor_last = all_events[-1].event_id if all_events else None
+    if head is None:
+        if any(event.seq != 0 for event in all_events):
+            broken = next(event for event in all_events if event.seq != 0)
+            return AuditChainVerification(
+                False,
+                len(all_events),
+                anchor_first,
+                anchor_last,
+                "audit_chain_head_missing",
+                broken.event_id,
+            )
+    else:
+        last_event = all_events[-1] if all_events else None
         if last_event is None or (
             last_event.seq != head.last_seq
             or last_event.chain_hash != head.last_chain_hash
         ):
             return AuditChainVerification(
                 False,
-                len(events),
-                first_event_id,
-                last_event_id,
+                len(all_events),
+                anchor_first,
+                anchor_last,
                 "audit_chain_truncated",
                 last_event.event_id if last_event else None,
             )
+        broken_id = _first_sequence_break(all_events, head.last_seq)
+        if broken_id is not None:
+            return AuditChainVerification(
+                False,
+                len(all_events),
+                anchor_first,
+                anchor_last,
+                "audit_sequence_gap",
+                broken_id,
+            )
 
     return AuditChainVerification(
         True,
diff --git a/gateway/tests/test_audit_chain_tamper.py b/gateway/tests/test_audit_chain_tamper.py
new file mode 100644
index 00000000..1d92e224
--- /dev/null
+++ b/gateway/tests/test_audit_chain_tamper.py
@@ -0,0 +1,338 @@
+"""Adversarial checks for audit hash-chain tamper and sequence gaps.
+
+These tests mutate stored rows the way a database writer can: delete a
+link, renumber seq, drop the head, or split the wallet-less chain. They
+do not call the network.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+
+import pytest
+from sqlalchemy import select
+
+from app.core.time import utc_now
+from app.db.database import get_session_factory
+from app.db.models import AuditChainHeadModel, ControlPlaneAuditEventModel
+from app.services.audit_chain import AuditChainHeadMissingError, verify_audit_chain
+from app.services.audit_log import record_audit_event
+
+
+@pytest.mark.anyio
+async def test_middle_deletion_breaks_the_hash_link(clean_database):
+    """Removing a middle link must not leave a valid chain.
+
+    The survivor's previous hash still names the deleted link, so the
+    walk cannot close. This was already rejected before the sequence
+    check; the test pins that reason.
+    """
+    wallet_id = "wlt-middle-delete"
+    for n in range(3):
+        await record_audit_event(
+            event="trust.middle", wallet_id=wallet_id, metadata={"n": n}
+        )
+
+    factory = get_session_factory()
+    async with factory() as session:
+        middle = (
+            (
+                await session.execute(
+                    select(ControlPlaneAuditEventModel)
+                    .where(ControlPlaneAuditEventModel.wallet_id == wallet_id)
+                    .order_by(ControlPlaneAuditEventModel.seq)
+                )
+            )
+            .scalars()
+            .all()
+        )[1]
+        await session.delete(middle)
+        await session.commit()
+
+    result = await verify_audit_chain(wallet_id=wallet_id)
+    assert result.valid is False
+    assert result.reason == "audit_previous_hash_mismatch"
+
+
+@pytest.mark.anyio
+async def test_renumbered_sequence_gap_is_rejected(clean_database):
+    """A hole in seq is a break even when the hash walk still lines up.
+
+    seq is not covered by the signature. Moving the tail's seq up to a
+    new head seq keeps the hash order and the head hash, so a check that
+    only compares the tail to the head reports the chain as valid.
+    """
+    wallet_id = "wlt-seq-gap"
+    for n in range(3):
+        await record_audit_event(
+            event="trust.gap", wallet_id=wallet_id, metadata={"n": n}
+        )
+
+    factory = get_session_factory()
+    async with factory() as session:
+        rows = (
+            (
+                await session.execute(
+                    select(ControlPlaneAuditEventModel)
+                    .where(ControlPlaneAuditEventModel.wallet_id == wallet_id)
+                    .order_by(ControlPlaneAuditEventModel.seq)
+                )
+            )
+            .scalars()
+            .all()
+        )
+        rows[-1].seq = 4
+        head = await session.get(AuditChainHeadModel, wallet_id)
+        assert head is not None
+        head.last_seq = 4
+        session.add(rows[-1])
+        session.add(head)
+        await session.commit()
+
+    result = await verify_audit_chain(wallet_id=wallet_id)
+    assert result.valid is False
+    assert result.reason == "audit_sequence_gap"
+    assert result.broken_event_id is not None
+
+    windowed = await verify_audit_chain(
+        wallet_id=wallet_id,
+        created_after=datetime(2000, 1, 1),
+    )
+    assert windowed.valid is False
+    assert windowed.reason == "audit_sequence_gap"
+
+
+@pytest.mark.anyio
+async def test_duplicate_sequence_is_rejected(clean_database):
+    """Two rows with one seq must not verify just because the tail matches.
+
+    Collapsing the tail onto the previous seq and pointing the head at
+    that same seq keeps hash order (created_at breaks the tie) and
+    satisfies the tail check.
+    """
+    wallet_id = "wlt-dup-seq"
+    for n in range(3):
+        await record_audit_event(
+            event="trust.dup", wallet_id=wallet_id, metadata={"n": n}
+        )
+
+    factory = get_session_factory()
+    async with factory() as session:
+        rows = (
+            (
+                await session.execute(
+                    select(ControlPlaneAuditEventModel)
+                    .where(ControlPlaneAuditEventModel.wallet_id == wallet_id)
+                    .order_by(ControlPlaneAuditEventModel.seq)
+                )
+            )
+            .scalars()
+            .all()
+        )
+        rows[-1].seq = rows[-2].seq
+        head = await session.get(AuditChainHeadModel, wallet_id)
+        assert head is not None
+        head.last_seq = rows[-1].seq
+        head.last_chain_hash = rows[-1].chain_hash
+        session.add(rows[-1])
+        session.add(head)
+        await session.commit()
+
+    result = await verify_audit_chain(wallet_id=wallet_id)
+    assert result.valid is False
+    assert result.reason == "audit_sequence_gap"
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    "window",
+    [
+        {"created_after": datetime(2000, 1, 1)},
+        {"created_before": datetime(2099, 1, 1)},
+        {
+            "created_after": datetime(2000, 1, 1),
+            "created_before": datetime(2099, 1, 1),
+        },
+    ],
+)
+async def test_time_window_does_not_hide_a_deleted_tail(clean_database, window):
+    """A wide time window must not skip the head anchor.
+
+    Deleting the tail leaves a hash-valid prefix. Verification used to
+    load the head only when both time filters were absent, so any window
+    that still covered the survivors returned valid.
+    """
+    wallet_id = "wlt-window-tail"
+    for n in range(3):
+        await record_audit_event(
+            event="trust.windowtail", wallet_id=wallet_id, metadata={"n": n}
+        )
+
+    factory = get_session_factory()
+    async with factory() as session:
+        last = (
+            (
+                await session.execute(
+                    select(ControlPlaneAuditEventModel)
+                    .where(ControlPlaneAuditEventModel.wallet_id == wallet_id)
+                    .order_by(ControlPlaneAuditEventModel.seq.desc())
+                )
+            )
+            .scalars()
+            .first()
+        )
+        await session.delete(last)
+        await session.commit()
+
+    plain = await verify_audit_chain(wallet_id=wallet_id)
+    assert plain.valid is False
+    assert plain.reason == "audit_chain_truncated"
+
+    windowed = await verify_audit_chain(wallet_id=wallet_id, **window)
+    assert windowed.valid is False
+    assert windowed.reason == "audit_chain_truncated"
+
+
+@pytest.mark.anyio
+async def test_window_that_excludes_every_event_keeps_a_valid_chain(clean_database):
+    """An empty window is not a truncated chain when the full chain is intact."""
+    wallet_id = "wlt-window-empty"
+    await record_audit_event(event="trust.keep", wallet_id=wallet_id)
+    result = await verify_audit_chain(
+        wallet_id=wallet_id,
+        created_before=datetime(2000, 1, 1),
+    )
+    assert result.valid is True
+    assert result.checked_events == 0
+
+
+@pytest.mark.anyio
+async def test_missing_head_on_a_modern_chain_is_rejected(clean_database):
+    """Dropping the head row must not make the remaining events look complete.
+
+    The head is the anchor for tail deletion. With the row gone, the old
+    verifier skipped that check and returned valid.
+    """
+    wallet_id = "wlt-head-gone"
+    for n in range(2):
+        await record_audit_event(
+            event="trust.head", wallet_id=wallet_id, metadata={"n": n}
+        )
+
+    factory = get_session_factory()
+    async with factory() as session:
+        head = await session.get(AuditChainHeadModel, wallet_id)
+        assert head is not None
+        await session.delete(head)
+        await session.commit()
+
+    result = await verify_audit_chain(wallet_id=wallet_id)
+    assert result.valid is False
+    assert result.reason == "audit_chain_head_missing"
+
+
+@pytest.mark.anyio
+async def test_append_refuses_to_start_a_new_genesis_when_the_head_is_gone(
+    clean_database,
+):
+    """The next append must not sign a fresh genesis over a live chain.
+
+    Without a head the append treated the wallet as empty, stored seq 1
+    with an empty previous hash, and left the older links unverifiable.
+    """
+    wallet_id = "wlt-head-append"
+    for n in range(2):
+        await record_audit_event(
+            event="trust.headappend", wallet_id=wallet_id, metadata={"n": n}
+        )
+
+    factory = get_session_factory()
+    async with factory() as session:
+        head = await session.get(AuditChainHeadModel, wallet_id)
+        assert head is not None
+        await session.delete(head)
+        await session.commit()
+
+    with pytest.raises(AuditChainHeadMissingError, match="audit_chain_head_missing"):
+        await record_audit_event(
+            event="trust.headappend", wallet_id=wallet_id, metadata={"n": 2}
+        )
+
+    async with factory() as session:
+        count = (
+            (
+                await session.execute(
+                    select(ControlPlaneAuditEventModel).where(
+                        ControlPlaneAuditEventModel.wallet_id == wallet_id
+                    )
+                )
+            )
+            .scalars()
+            .all()
+        )
+    assert len(count) == 2
+    result = await verify_audit_chain(wallet_id=wallet_id)
+    assert result.valid is False
+    assert result.reason == "audit_chain_head_missing"
+
+
+@pytest.mark.anyio
+async def test_blank_wallet_id_shares_the_wallet_less_chain(clean_database):
+    """NULL and "" are one chain, because they already share one head key.
+
+    Verification used to walk them separately. Each half then failed: the
+    NULL side looked truncated, and the blank side did not link to genesis.
+    """
+    await record_audit_event(event="system.a", wallet_id=None, metadata={"n": 1})
+    await record_audit_event(event="system.b", wallet_id="", metadata={"n": 2})
+
+    global_result = await verify_audit_chain(wallet_id=None)
+    assert global_result.valid is True, global_result
+    assert global_result.checked_events == 2
+
+    blank = await verify_audit_chain(wallet_id="")
+    assert blank.valid is True, blank
+    assert blank.checked_events == 2
+
+
+@pytest.mark.anyio
+async def test_legacy_seq_zero_chain_without_a_head_still_verifies(clean_database):
+    """Rows from before seq existed are all seq 0 and have no head.
+
+    Those rows must keep verifying. A modern chain (seq starting at 1)
+    with the head removed must not.
+    """
+    wallet_id = "wlt-legacy-seq0"
+    base = utc_now()
+    for n in range(2):
+        await record_audit_event(
+            event="trust.legacy",
+            wallet_id=wallet_id,
+            created_at=base + timedelta(seconds=n),
+            metadata={"n": n},
+        )
+
+    factory = get_session_factory()
+    async with factory() as session:
+        rows = (
+            (
+                await session.execute(
+                    select(ControlPlaneAuditEventModel).where(
+                        ControlPlaneAuditEventModel.wallet_id == wallet_id
+                    )
+                )
+            )
+            .scalars()
+            .all()
+        )
+        for row in rows:
+            row.seq = 0
+            session.add(row)
+        head = await session.get(AuditChainHeadModel, wallet_id)
+        assert head is not None
+        await session.delete(head)
+        await session.commit()
+
+    result = await verify_audit_chain(wallet_id=wallet_id)
+    assert result.valid is True, result
+    assert result.checked_events == 2
diff --git a/integrations/framework_integrations/README.autogen.md b/integrations/framework_integrations/README.autogen.md
index e6ba9ccd..63174f50 100644
--- a/integrations/framework_integrations/README.autogen.md
+++ b/integrations/framework_integrations/README.autogen.md
@@ -25,9 +25,7 @@ from framework_integrations import B2AClient, get_autogen_tools
 
 # Initialize client
 client = B2AClient(
-    api_url="http://localhost:8000",
-    api_key="your-api-key",
-    wallet_id="your-wallet-id"
+    api_url="http://localhost:8000", api_key="your-api-key", wallet_id="your-wallet-id"
 )
 
 # Get AutoGen-compatible function map
@@ -35,22 +33,17 @@ function_map = get_autogen_tools(client)
 
 # Create assistant agent
 assistant = autogen.AssistantAgent(
-    name="assistant",
-    llm_config=llm_config,
-    function_map=function_map
+    name="assistant", llm_config=llm_config, function_map=function_map
 )
 
 # Create user proxy
 user_proxy = autogen.UserProxyAgent(
-    name="user_proxy",
-    human_input_mode="NEVER",
-    max_consecutive_auto_reply=10
+    name="user_proxy", human_input_mode="NEVER", max_consecutive_auto_reply=10
 )
 
 # Start conversation
 user_proxy.initiate_chat(
-    assistant,
-    message="Check my balance and emit a telemetry event"
+    assistant, message="Check my balance and emit a telemetry event"
 )
 ```
 
@@ -72,28 +65,19 @@ import autogen
 
 # Create agents with B2A tools
 researcher = autogen.AssistantAgent(
-    name="researcher",
-    llm_config=llm_config,
-    function_map=get_autogen_tools(client)
+    name="researcher", llm_config=llm_config, function_map=get_autogen_tools(client)
 )
 
 writer = autogen.AssistantAgent(
-    name="writer",
-    llm_config=llm_config,
-    function_map=get_autogen_tools(client)
+    name="writer", llm_config=llm_config, function_map=get_autogen_tools(client)
 )
 
 # Group chat
-group_chat = autogen.GroupChat(
-    agents=[researcher, writer],
-    messages=[],
-    max_round=10
-)
+group_chat = autogen.GroupChat(agents=[researcher, writer], messages=[], max_round=10)
 
 manager = autogen.GroupChatManager(groupchat=group_chat)
 
 user_proxy.initiate_chat(
-    manager,
-    message="Research AI developments and write a summary"
+    manager, message="Research AI developments and write a summary"
 )
 ```
diff --git a/integrations/framework_integrations/README.langgraph.md b/integrations/framework_integrations/README.langgraph.md
index 69a1deb1..6a8a3d30 100644
--- a/integrations/framework_integrations/README.langgraph.md
+++ b/integrations/framework_integrations/README.langgraph.md
@@ -25,9 +25,7 @@ from framework_integrations import B2AClient, get_langgraph_tools
 
 # Initialize client
 client = B2AClient(
-    api_url="http://localhost:8000",
-    api_key="your-api-key",
-    wallet_id="your-wallet-id"
+    api_url="http://localhost:8000", api_key="your-api-key", wallet_id="your-wallet-id"
 )
 
 # Get LangGraph-compatible tools
@@ -66,14 +64,16 @@ tools = get_langgraph_tools(client)
 researcher = create_react_agent(
     model,
     tools=tools,
-    state_modifier="You are a research agent. Use tools to gather information."
+    state_modifier="You are a research agent. Use tools to gather information.",
 )
 
-result = await researcher.ainvoke({
-    "messages": [
-        "Research the latest AI developments and send results to researcher-002"
-    ]
-})
+result = await researcher.ainvoke(
+    {
+        "messages": [
+            "Research the latest AI developments and send results to researcher-002"
+        ]
+    }
+)
 ```
 
 ## Example: Autonomous Task Agent
@@ -84,10 +84,8 @@ tools = get_langgraph_tools(client)
 task_agent = create_react_agent(
     model,
     tools=tools,
-    prompt="You autonomously complete tasks. Monitor your budget and heal when needed."
+    prompt="You autonomously complete tasks. Monitor your budget and heal when needed.",
 )
 
-result = await task_agent.ainvoke({
-    "messages": ["Process the pending task queue"]
-})
+result = await task_agent.ainvoke({"messages": ["Process the pending task queue"]})
 ```
diff --git a/integrations/framework_integrations/README.llamaindex.md b/integrations/framework_integrations/README.llamaindex.md
index 471d8f04..45615e60 100644
--- a/integrations/framework_integrations/README.llamaindex.md
+++ b/integrations/framework_integrations/README.llamaindex.md
@@ -25,9 +25,7 @@ from framework_integrations import B2AClient, get_llamaindex_tools
 
 # Initialize client
 client = B2AClient(
-    api_url="http://localhost:8000",
-    api_key="your-api-key",
-    wallet_id="your-wallet-id"
+    api_url="http://localhost:8000", api_key="your-api-key", wallet_id="your-wallet-id"
 )
 
 # Get LlamaIndex-compatible tools
@@ -64,7 +62,7 @@ tools = get_llamaindex_tools(client)
 agent = ReActAgent.from_tools(
     tools,
     llm=llm,
-    system_prompt="You are a data processing agent. Use tools to handle data."
+    system_prompt="You are a data processing agent. Use tools to handle data.",
 )
 
 result = await agent.achat("Process the uploaded dataset")

From 5441b5f01d72ff8f551b1b565f2339fd81f88121 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Fri, 23 Oct 2026 19:43:55 -0700
Subject: [PATCH 2/7] Fix audit head race under concurrent appends

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/app/services/audit_chain.py | 13 ++++++++++---
 1 file changed, 10 insertions(+), 3 deletions(-)

diff --git a/gateway/app/services/audit_chain.py b/gateway/app/services/audit_chain.py
index e1ad77c4..3a614808 100644
--- a/gateway/app/services/audit_chain.py
+++ b/gateway/app/services/audit_chain.py
@@ -61,7 +61,7 @@ def _chain_wallet_filter(wallet_id: str | None) -> ColumnElement[bool]:
     the same chain as NULL. Walking those rows apart made each half look
     truncated or unlinked.
     """
-    column = ControlPlaneAuditEventModel.wallet_id
+    column = cast(ColumnElement[str | None], ControlPlaneAuditEventModel.wallet_id)
     if wallet_id:
         return cast(ColumnElement[bool], column == wallet_id)
     return cast(ColumnElement[bool], or_(column.is_(None), column == ""))
@@ -283,6 +283,7 @@ async def append_chained_audit_event(
     # spending the full budget on it only delays the real error.
     integrity_race_attempts = 2
     integrity_failures = 0
+    missing_head_observations = 0
     for attempt in range(attempts):
         session = factory()
         try:
@@ -319,13 +320,19 @@ async def append_chained_audit_event(
                     # over rows that are already stored splits the hash links.
                     prior = (
                         await session.execute(
-                            select(ControlPlaneAuditEventModel.event_id)
+                            select(cast(Any, ControlPlaneAuditEventModel.event_id))
                             .where(_chain_wallet_filter(model.wallet_id))
                             .limit(1)
                         )
                     ).first()
                     if prior is not None:
-                        raise AuditChainHeadMissingError("audit_chain_head_missing")
+                        # The head SELECT may precede another writer's first
+                        # commit while this event SELECT follows it. Retry in
+                        # a new transaction before calling the head missing.
+                        missing_head_observations += 1
+                        if missing_head_observations > 1:
+                            raise AuditChainHeadMissingError("audit_chain_head_missing")
+                        raise _HeadConflict()
                 model.seq = observed_seq + 1
                 _sign_with_previous(model, previous_hash, signing_key_id=key.key_id)
                 now = utc_now()

From 2a1f31a072dea0ebed61668ff37a1e690f4747fa Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Fri, 23 Oct 2026 19:44:06 -0700
Subject: [PATCH 3/7] fix: recheck missing audit head after concurrent append

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/app/services/audit_chain.py | 13 ++++++++++---
 1 file changed, 10 insertions(+), 3 deletions(-)

diff --git a/gateway/app/services/audit_chain.py b/gateway/app/services/audit_chain.py
index e1ad77c4..11603ebe 100644
--- a/gateway/app/services/audit_chain.py
+++ b/gateway/app/services/audit_chain.py
@@ -61,7 +61,7 @@ def _chain_wallet_filter(wallet_id: str | None) -> ColumnElement[bool]:
     the same chain as NULL. Walking those rows apart made each half look
     truncated or unlinked.
     """
-    column = ControlPlaneAuditEventModel.wallet_id
+    column = cast(ColumnElement[str | None], ControlPlaneAuditEventModel.wallet_id)
     if wallet_id:
         return cast(ColumnElement[bool], column == wallet_id)
     return cast(ColumnElement[bool], or_(column.is_(None), column == ""))
@@ -283,6 +283,7 @@ async def append_chained_audit_event(
     # spending the full budget on it only delays the real error.
     integrity_race_attempts = 2
     integrity_failures = 0
+    missing_head_seen = False
     for attempt in range(attempts):
         session = factory()
         try:
@@ -319,13 +320,19 @@ async def append_chained_audit_event(
                     # over rows that are already stored splits the hash links.
                     prior = (
                         await session.execute(
-                            select(ControlPlaneAuditEventModel.event_id)
+                            select(cast(Any, ControlPlaneAuditEventModel.event_id))
                             .where(_chain_wallet_filter(model.wallet_id))
                             .limit(1)
                         )
                     ).first()
                     if prior is not None:
-                        raise AuditChainHeadMissingError("audit_chain_head_missing")
+                        # The head read can precede another writer's commit,
+                        # while this event read follows it. Recheck in a fresh
+                        # transaction before treating the head as deleted.
+                        if missing_head_seen:
+                            raise AuditChainHeadMissingError("audit_chain_head_missing")
+                        missing_head_seen = True
+                        raise _HeadConflict()
                 model.seq = observed_seq + 1
                 _sign_with_previous(model, previous_hash, signing_key_id=key.key_id)
                 now = utc_now()

From 377c5605a9cd08c60e45725670809f0c8c6d7fef Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Mon, 26 Oct 2026 00:50:18 -0700
Subject: [PATCH 4/7] fix(audit): reconcile sequence check with signed seq from
 main

Main (#43) put seq in the signed payload and still accepts legacy
seq-less signatures. The contiguity check assumed seq was unsigned and
rejected legacy rows with any stored seq. Only require contiguity across
seq-signed runs, and update tamper tests: renumbering a signed row now
fails the payload hash, and legacy seq-0 fixtures are signed seq-less.
---
 gateway/app/services/audit_chain.py      |  22 +++--
 gateway/tests/test_audit_chain_tamper.py | 114 +++++++++++++++++++----
 2 files changed, 107 insertions(+), 29 deletions(-)

diff --git a/gateway/app/services/audit_chain.py b/gateway/app/services/audit_chain.py
index ff2fb465..a774d21c 100644
--- a/gateway/app/services/audit_chain.py
+++ b/gateway/app/services/audit_chain.py
@@ -117,19 +117,21 @@ def _chain_wallet_filter(wallet_id: str | None) -> ColumnElement[bool]:
 
 def _first_sequence_break(
     events: list[ControlPlaneAuditEventModel],
-    last_seq: int,
 ) -> str | None:
-    """Event id where ``seq`` stops being 1..last_seq, in chain order.
+    """Event id where a run of seq-signed rows stops being contiguous.
 
-    ``seq`` is not signed. A stored tail can be renumbered so it still
-    matches the head while a hole or a duplicate sits in front of it.
-    ``events`` must already be ordered by seq, then created_at.
+    Legacy signatures omit ``seq``, so their stored numbers cannot establish
+    contiguity. ``events`` must already be ordered by seq, then created_at.
     """
-    for index, event in enumerate(events, start=1):
-        if index > last_seq or event.seq != index:
+    expected_seq: int | None = 1
+    for event in events:
+        payload = _payload_matching_stored_hash(event)
+        if payload is not None and "seq" not in payload:
+            expected_seq = None
+            continue
+        if expected_seq is not None and event.seq != expected_seq:
             return event.event_id
-    if len(events) != last_seq:
-        return events[-1].event_id if events else None
+        expected_seq = event.seq + 1
     return None
 
 
@@ -687,7 +689,7 @@ async def _verify_single_chain(
                 "audit_chain_truncated",
                 last_event.event_id if last_event else None,
             )
-        broken_id = _first_sequence_break(all_events, head.last_seq)
+        broken_id = _first_sequence_break(all_events)
         if broken_id is not None:
             return AuditChainVerification(
                 False,
diff --git a/gateway/tests/test_audit_chain_tamper.py b/gateway/tests/test_audit_chain_tamper.py
index 1d92e224..c6ddff5d 100644
--- a/gateway/tests/test_audit_chain_tamper.py
+++ b/gateway/tests/test_audit_chain_tamper.py
@@ -15,8 +15,14 @@
 from app.core.time import utc_now
 from app.db.database import get_session_factory
 from app.db.models import AuditChainHeadModel, ControlPlaneAuditEventModel
-from app.services.audit_chain import AuditChainHeadMissingError, verify_audit_chain
+from app.services.audit_chain import (
+    AuditChainHeadMissingError,
+    _sign_with_previous,
+    verify_audit_chain,
+)
 from app.services.audit_log import record_audit_event
+from app.services.signing_keys import get_signing_key_service, sha256_hex
+from tests.test_audit_chain import _payload_without_seq
 
 
 @pytest.mark.anyio
@@ -56,12 +62,7 @@ async def test_middle_deletion_breaks_the_hash_link(clean_database):
 
 @pytest.mark.anyio
 async def test_renumbered_sequence_gap_is_rejected(clean_database):
-    """A hole in seq is a break even when the hash walk still lines up.
-
-    seq is not covered by the signature. Moving the tail's seq up to a
-    new head seq keeps the hash order and the head hash, so a check that
-    only compares the tail to the head reports the chain as valid.
-    """
+    """Renumbering breaks a signature; a signed gap breaks contiguity."""
     wallet_id = "wlt-seq-gap"
     for n in range(3):
         await record_audit_event(
@@ -81,14 +82,42 @@ async def test_renumbered_sequence_gap_is_rejected(clean_database):
             .scalars()
             .all()
         )
-        rows[-1].seq = 4
+        rows[-1].seq += 1
         head = await session.get(AuditChainHeadModel, wallet_id)
         assert head is not None
-        head.last_seq = 4
+        head.last_seq = rows[-1].seq
         session.add(rows[-1])
         session.add(head)
         await session.commit()
 
+    result = await verify_audit_chain(wallet_id=wallet_id)
+    assert result.valid is False
+    assert result.reason == "audit_payload_hash_mismatch"
+
+    async with factory() as session:
+        rows = (
+            (
+                await session.execute(
+                    select(ControlPlaneAuditEventModel)
+                    .where(ControlPlaneAuditEventModel.wallet_id == wallet_id)
+                    .order_by(ControlPlaneAuditEventModel.seq)
+                )
+            )
+            .scalars()
+            .all()
+        )
+        tail = rows[-1]
+        assert tail.signature_key_id is not None
+        # Seq-less signatures are exempt; only a valid seq-signed gap reaches
+        # the contiguity check.
+        _sign_with_previous(
+            tail, rows[-2].chain_hash, signing_key_id=tail.signature_key_id
+        )
+        head = await session.get(AuditChainHeadModel, wallet_id)
+        assert head is not None
+        head.last_chain_hash = tail.chain_hash
+        await session.commit()
+
     result = await verify_audit_chain(wallet_id=wallet_id)
     assert result.valid is False
     assert result.reason == "audit_sequence_gap"
@@ -104,16 +133,15 @@ async def test_renumbered_sequence_gap_is_rejected(clean_database):
 
 @pytest.mark.anyio
 async def test_duplicate_sequence_is_rejected(clean_database):
-    """Two rows with one seq must not verify just because the tail matches.
-
-    Collapsing the tail onto the previous seq and pointing the head at
-    that same seq keeps hash order (created_at breaks the tie) and
-    satisfies the tail check.
-    """
+    """A duplicate breaks its signature and, if re-signed, contiguity."""
     wallet_id = "wlt-dup-seq"
+    base = utc_now()
     for n in range(3):
         await record_audit_event(
-            event="trust.dup", wallet_id=wallet_id, metadata={"n": n}
+            event="trust.dup",
+            wallet_id=wallet_id,
+            created_at=base + timedelta(seconds=n),
+            metadata={"n": n},
         )
 
     factory = get_session_factory()
@@ -138,6 +166,35 @@ async def test_duplicate_sequence_is_rejected(clean_database):
         session.add(head)
         await session.commit()
 
+    result = await verify_audit_chain(wallet_id=wallet_id)
+    assert result.valid is False
+    assert result.reason == "audit_payload_hash_mismatch"
+
+    async with factory() as session:
+        rows = (
+            (
+                await session.execute(
+                    select(ControlPlaneAuditEventModel)
+                    .where(ControlPlaneAuditEventModel.wallet_id == wallet_id)
+                    .order_by(
+                        ControlPlaneAuditEventModel.seq,
+                        ControlPlaneAuditEventModel.created_at,
+                    )
+                )
+            )
+            .scalars()
+            .all()
+        )
+        tail = rows[-1]
+        assert tail.signature_key_id is not None
+        _sign_with_previous(
+            tail, rows[-2].chain_hash, signing_key_id=tail.signature_key_id
+        )
+        head = await session.get(AuditChainHeadModel, wallet_id)
+        assert head is not None
+        head.last_chain_hash = tail.chain_hash
+        await session.commit()
+
     result = await verify_audit_chain(wallet_id=wallet_id)
     assert result.valid is False
     assert result.reason == "audit_sequence_gap"
@@ -317,16 +374,35 @@ async def test_legacy_seq_zero_chain_without_a_head_still_verifies(clean_databas
         rows = (
             (
                 await session.execute(
-                    select(ControlPlaneAuditEventModel).where(
-                        ControlPlaneAuditEventModel.wallet_id == wallet_id
-                    )
+                    select(ControlPlaneAuditEventModel)
+                    .where(ControlPlaneAuditEventModel.wallet_id == wallet_id)
+                    .order_by(ControlPlaneAuditEventModel.created_at)
                 )
             )
             .scalars()
             .all()
         )
+        previous_hash = None
         for row in rows:
             row.seq = 0
+            row.previous_hash = previous_hash
+            payload = _payload_without_seq(row)
+            row.payload_hash = sha256_hex(payload)
+            payload["payload_hash"] = row.payload_hash
+            assert row.signature_key_id is not None
+            row.signature, row.signature_key_id, _ = (
+                get_signing_key_service().sign_payload_with_key_id(
+                    payload, row.signature_key_id
+                )
+            )
+            row.chain_hash = sha256_hex(
+                {
+                    "previous_hash": previous_hash,
+                    "payload_hash": row.payload_hash,
+                    "signature": row.signature,
+                }
+            )
+            previous_hash = row.chain_hash
             session.add(row)
         head = await session.get(AuditChainHeadModel, wallet_id)
         assert head is not None

From 64b26f6bb1c15c728eedd379bb15af1e8681975b Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Mon, 26 Oct 2026 01:37:02 -0700
Subject: [PATCH 5/7] audit chain: walk hash linkage over full chain under
 windowed verify; legacy-to-signed seq boundary check (Muse FAIL on #28)

---
 gateway/app/services/audit_chain.py      |  82 ++++++++---------
 gateway/tests/test_audit_chain.py        |   6 +-
 gateway/tests/test_audit_chain_tamper.py | 111 +++++++++++++++++++++++
 3 files changed, 151 insertions(+), 48 deletions(-)

diff --git a/gateway/app/services/audit_chain.py b/gateway/app/services/audit_chain.py
index a774d21c..50ba7b43 100644
--- a/gateway/app/services/audit_chain.py
+++ b/gateway/app/services/audit_chain.py
@@ -121,13 +121,15 @@ def _first_sequence_break(
     """Event id where a run of seq-signed rows stops being contiguous.
 
     Legacy signatures omit ``seq``, so their stored numbers cannot establish
-    contiguity. ``events`` must already be ordered by seq, then created_at.
+    contiguity within a legacy run. A following seq-signed row must continue
+    from the final legacy row's stored sequence. ``events`` must already be
+    ordered by seq, then created_at.
     """
     expected_seq: int | None = 1
     for event in events:
         payload = _payload_matching_stored_hash(event)
         if payload is not None and "seq" not in payload:
-            expected_seq = None
+            expected_seq = event.seq + 1
             continue
         if expected_seq is not None and event.seq != expected_seq:
             return event.event_id
@@ -578,20 +580,11 @@ async def _verify_single_chain(
     else:
         events = all_events
 
-    # A window that excludes the wallet's genesis event starts mid-chain: the
-    # first in-window event legitimately links to an event outside the
-    # window. Seed the expected predecessor from the last stored event before
-    # the window instead of from genesis, otherwise every non-genesis window
-    # reports a valid chain as tampered (audit_previous_hash_mismatch).
     previous_hash: str | None = None
-    if events and created_after:
-        first_seq = events[0].seq
-        below = [event for event in all_events if event.seq < first_seq]
-        if below:
-            previous_hash = below[-1].chain_hash
     first_event_id = events[0].event_id if events else None
     last_event_id = events[-1].event_id if events else None
-    for event in events:
+    window_event_ids = {event.event_id for event in events}
+    for event in all_events:
         payload = _payload_matching_stored_hash(event)
         if payload is None:
             return AuditChainVerification(
@@ -611,35 +604,36 @@ async def _verify_single_chain(
                 "audit_previous_hash_mismatch",
                 event.event_id,
             )
-        if not event.signature or not event.signature_key_id:
-            return AuditChainVerification(
-                False,
-                len(events),
-                first_event_id,
-                last_event_id,
-                "audit_signature_missing",
-                event.event_id,
-            )
-        signed_payload = {
-            **payload,
-            "payload_hash": event.payload_hash,
-            "alg": "Ed25519",
-            "kid": event.signature_key_id,
-        }
-        ok = await get_signing_key_service().verify_payload(
-            signed_payload,
-            signature=event.signature,
-            key_id=event.signature_key_id,
-        )
-        if not ok:
-            return AuditChainVerification(
-                False,
-                len(events),
-                first_event_id,
-                last_event_id,
-                "audit_signature_invalid",
-                event.event_id,
+        if event.event_id in window_event_ids:
+            if not event.signature or not event.signature_key_id:
+                return AuditChainVerification(
+                    False,
+                    len(events),
+                    first_event_id,
+                    last_event_id,
+                    "audit_signature_missing",
+                    event.event_id,
+                )
+            signed_payload = {
+                **payload,
+                "payload_hash": event.payload_hash,
+                "alg": "Ed25519",
+                "kid": event.signature_key_id,
+            }
+            ok = await get_signing_key_service().verify_payload(
+                signed_payload,
+                signature=event.signature,
+                key_id=event.signature_key_id,
             )
+            if not ok:
+                return AuditChainVerification(
+                    False,
+                    len(events),
+                    first_event_id,
+                    last_event_id,
+                    "audit_signature_invalid",
+                    event.event_id,
+                )
         expected_chain_hash = sha256_hex(
             {
                 "previous_hash": event.previous_hash,
@@ -658,10 +652,8 @@ async def _verify_single_chain(
             )
         previous_hash = event.chain_hash
 
-    # The hash walk above covers the requested window only. The head and
-    # the sequence belong to the whole chain. Checking them on the window,
-    # or skipping them when the head row is gone, let a deleted tail or a
-    # renumbered seq verify as a complete chain.
+    # The head and sequence also belong to the whole chain. Skipping them
+    # when the head row is gone lets a deleted tail or renumbered seq verify.
     anchor_first = all_events[0].event_id if all_events else None
     anchor_last = all_events[-1].event_id if all_events else None
     if head is None:
diff --git a/gateway/tests/test_audit_chain.py b/gateway/tests/test_audit_chain.py
index 0534dc42..2a687e59 100644
--- a/gateway/tests/test_audit_chain.py
+++ b/gateway/tests/test_audit_chain.py
@@ -694,8 +694,7 @@ async def test_windowed_verify_of_valid_chain_is_valid(client, clean_database):
     assert windowed.valid is True, windowed
     assert windowed.checked_events == 2
 
-    # The seeded predecessor is really checked: tampering the chain_hash of
-    # the last event *outside* the window breaks the first in-window link.
+    # The full chain walk checks the hash of an event outside the window.
     async with factory() as session:
         event = (
             await session.execute(
@@ -712,4 +711,5 @@ async def test_windowed_verify_of_valid_chain_is_valid(client, clean_database):
         created_after=third_created - timedelta(microseconds=1),
     )
     assert tampered.valid is False
-    assert tampered.reason == "audit_previous_hash_mismatch"
+    assert tampered.reason == "audit_chain_hash_mismatch"
+    assert tampered.broken_event_id == rows[1].event_id
diff --git a/gateway/tests/test_audit_chain_tamper.py b/gateway/tests/test_audit_chain_tamper.py
index c6ddff5d..7ce9f846 100644
--- a/gateway/tests/test_audit_chain_tamper.py
+++ b/gateway/tests/test_audit_chain_tamper.py
@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 from datetime import datetime, timedelta
+from uuid import uuid4
 
 import pytest
 from sqlalchemy import select
@@ -60,6 +61,116 @@ async def test_middle_deletion_breaks_the_hash_link(clean_database):
     assert result.reason == "audit_previous_hash_mismatch"
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize("deleted_seqs", [(4,), (3,), (3, 4), (2,)])
+async def test_windowed_verify_finds_deleted_legacy_and_modern_links(
+    clean_database, deleted_seqs
+):
+    wallet_id = f"wlt-{uuid4()}"
+    base = utc_now()
+    for seq in range(1, 7):
+        await record_audit_event(
+            event="trust.window-link",
+            wallet_id=wallet_id,
+            created_at=base + timedelta(seconds=seq),
+            metadata={"seq": seq},
+        )
+
+    factory = get_session_factory()
+    async with factory() as session:
+        rows = (
+            (
+                await session.execute(
+                    select(ControlPlaneAuditEventModel)
+                    .where(ControlPlaneAuditEventModel.wallet_id == wallet_id)
+                    .order_by(ControlPlaneAuditEventModel.seq)
+                )
+            )
+            .scalars()
+            .all()
+        )
+        assert [row.seq for row in rows] == list(range(1, 7))
+        previous_hash = None
+        for row in rows[:3]:
+            row.previous_hash = previous_hash
+            payload = _payload_without_seq(row)
+            row.payload_hash = sha256_hex(payload)
+            payload["payload_hash"] = row.payload_hash
+            assert row.signature_key_id is not None
+            row.signature, row.signature_key_id, _ = (
+                get_signing_key_service().sign_payload_with_key_id(
+                    payload, row.signature_key_id
+                )
+            )
+            row.chain_hash = sha256_hex(
+                {
+                    "previous_hash": previous_hash,
+                    "payload_hash": row.payload_hash,
+                    "signature": row.signature,
+                }
+            )
+            previous_hash = row.chain_hash
+        for row in rows[3:]:
+            assert row.signature_key_id is not None
+            _sign_with_previous(row, previous_hash, signing_key_id=row.signature_key_id)
+            previous_hash = row.chain_hash
+        head = await session.get(AuditChainHeadModel, wallet_id)
+        assert head is not None
+        head.last_chain_hash = previous_hash
+        await session.commit()
+
+    window = {"created_after": rows[-1].created_at}
+    intact = await verify_audit_chain(wallet_id=wallet_id, **window)
+    assert intact.valid is True, intact
+    assert intact.checked_events == 1
+    assert intact.first_event_id == intact.last_event_id == rows[-1].event_id
+
+    async with factory() as session:
+        for row in rows:
+            if row.seq in deleted_seqs:
+                stored = await session.get(ControlPlaneAuditEventModel, row.event_id)
+                assert stored is not None
+                await session.delete(stored)
+        await session.commit()
+
+    result = await verify_audit_chain(wallet_id=wallet_id, **window)
+    assert result.valid is False
+    assert result.reason == "audit_previous_hash_mismatch"
+    assert result.broken_event_id == rows[max(deleted_seqs)].event_id
+
+    if deleted_seqs == (3,):
+        # A writer can repair linkage and re-sign the modern suffix. The
+        # first seq-signed row must still reveal the gap after legacy rows.
+        async with factory() as session:
+            survivors = (
+                (
+                    await session.execute(
+                        select(ControlPlaneAuditEventModel)
+                        .where(ControlPlaneAuditEventModel.wallet_id == wallet_id)
+                        .order_by(ControlPlaneAuditEventModel.seq)
+                    )
+                )
+                .scalars()
+                .all()
+            )
+            previous_hash = survivors[1].chain_hash
+            for row in survivors[2:]:
+                assert row.signature_key_id is not None
+                _sign_with_previous(
+                    row, previous_hash, signing_key_id=row.signature_key_id
+                )
+                previous_hash = row.chain_hash
+            head = await session.get(AuditChainHeadModel, wallet_id)
+            assert head is not None
+            head.last_chain_hash = previous_hash
+            await session.commit()
+
+        relinked = await verify_audit_chain(wallet_id=wallet_id, **window)
+        assert relinked.valid is False
+        assert relinked.reason == "audit_sequence_gap"
+        assert relinked.broken_event_id == rows[3].event_id
+
+
 @pytest.mark.anyio
 async def test_renumbered_sequence_gap_is_rejected(clean_database):
     """Renumbering breaks a signature; a signed gap breaks contiguity."""

From 099429bee25e41f1dc96ae7f429aa2cf5916a517 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Mon, 26 Oct 2026 02:32:39 -0700
Subject: [PATCH 6/7] fix(audit): missing chain head is a non-retryable error

---
 gateway/app/services/audit_chain.py      | 13 ++++------
 gateway/tests/test_audit_chain_tamper.py | 31 ++++++++++++++++++++++++
 2 files changed, 36 insertions(+), 8 deletions(-)

diff --git a/gateway/app/services/audit_chain.py b/gateway/app/services/audit_chain.py
index 50ba7b43..dd1a2466 100644
--- a/gateway/app/services/audit_chain.py
+++ b/gateway/app/services/audit_chain.py
@@ -307,7 +307,6 @@ async def append_chained_audit_event(
     # spending the full budget on it only delays the real error.
     integrity_race_attempts = 2
     integrity_failures = 0
-    missing_head_observations = 0
     for attempt in range(attempts):
         session = factory()
         try:
@@ -350,13 +349,11 @@ async def append_chained_audit_event(
                         )
                     ).first()
                     if prior is not None:
-                        # The head SELECT may precede another writer's first
-                        # commit while this event SELECT follows it. Retry in
-                        # a new transaction before calling the head missing.
-                        missing_head_observations += 1
-                        if missing_head_observations > 1:
-                            raise AuditChainHeadMissingError("audit_chain_head_missing")
-                        raise _HeadConflict()
+                        # A first writer may have committed between the head
+                        # and event reads. Only that race warrants a retry.
+                        if await session.get(AuditChainHeadModel, wallet_key):
+                            raise _HeadConflict()
+                        raise AuditChainHeadMissingError("audit_chain_head_missing")
                 model.seq = observed_seq + 1
                 _sign_with_previous(model, previous_hash, signing_key_id=key.key_id)
                 now = utc_now()
diff --git a/gateway/tests/test_audit_chain_tamper.py b/gateway/tests/test_audit_chain_tamper.py
index 7ce9f846..742beaf3 100644
--- a/gateway/tests/test_audit_chain_tamper.py
+++ b/gateway/tests/test_audit_chain_tamper.py
@@ -17,6 +17,7 @@
 from app.db.database import get_session_factory
 from app.db.models import AuditChainHeadModel, ControlPlaneAuditEventModel
 from app.services.audit_chain import (
+    AuditChainContendedError,
     AuditChainHeadMissingError,
     _sign_with_previous,
     verify_audit_chain,
@@ -444,6 +445,36 @@ async def test_append_refuses_to_start_a_new_genesis_when_the_head_is_gone(
     assert result.reason == "audit_chain_head_missing"
 
 
+@pytest.mark.anyio
+async def test_missing_head_error_is_non_retryable(clean_database, monkeypatch):
+    wallet_id = f"wlt-{uuid4()}"
+    await record_audit_event(event="trust.headappend", wallet_id=wallet_id)
+
+    factory = get_session_factory()
+    async with factory() as session:
+        head = await session.get(AuditChainHeadModel, wallet_id)
+        assert head is not None
+        await session.delete(head)
+        await session.commit()
+
+    from app.services import audit_chain
+
+    attempts = 0
+
+    def counted_factory():
+        nonlocal attempts
+        attempts += 1
+        return factory()
+
+    monkeypatch.setattr(audit_chain, "get_session_factory", lambda: counted_factory)
+    with pytest.raises(AuditChainHeadMissingError) as excinfo:
+        await record_audit_event(event="trust.headappend", wallet_id=wallet_id)
+
+    assert excinfo.value.reason == "audit_chain_head_missing"
+    assert not isinstance(excinfo.value, AuditChainContendedError)
+    assert attempts == 1
+
+
 @pytest.mark.anyio
 async def test_blank_wallet_id_shares_the_wallet_less_chain(clean_database):
     """NULL and "" are one chain, because they already share one head key.

From ace9e19c4bed7016a123475b57e384fa3c2705c6 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Mon, 26 Oct 2026 06:34:13 -0700
Subject: [PATCH 7/7] fix(audit): continue legacy seq-0 chains that never had a
 head

Migration 017 set pre-seq rows to seq 0 and migration 019 created empty head
storage, so every append for such a wallet raised AuditChainHeadMissingError.
When no head exists and every stored row is a seq-0 legacy row, link the new
event to the terminal legacy chain hash at seq 1 and write the head. A
missing head over any seq-signed row still fails closed.

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/app/services/audit_chain.py      | 33 +++++++++++++++++++++---
 gateway/tests/test_audit_chain_tamper.py | 22 ++++++++++++++++
 2 files changed, 51 insertions(+), 4 deletions(-)

diff --git a/gateway/app/services/audit_chain.py b/gateway/app/services/audit_chain.py
index dd1a2466..bced1246 100644
--- a/gateway/app/services/audit_chain.py
+++ b/gateway/app/services/audit_chain.py
@@ -341,19 +341,44 @@ async def append_chained_audit_event(
                 if row is None:
                     # No head does not mean an empty chain. A genesis signed
                     # over rows that are already stored splits the hash links.
-                    prior = (
+                    terminal = (
                         await session.execute(
-                            select(cast(Any, ControlPlaneAuditEventModel.event_id))
+                            select(
+                                cast(Any, ControlPlaneAuditEventModel.seq),
+                                cast(Any, ControlPlaneAuditEventModel.chain_hash),
+                            )
                             .where(_chain_wallet_filter(model.wallet_id))
+                            .order_by(
+                                desc(
+                                    cast(
+                                        ColumnElement[Any],
+                                        ControlPlaneAuditEventModel.seq,
+                                    )
+                                ),
+                                desc(
+                                    cast(
+                                        ColumnElement[Any],
+                                        ControlPlaneAuditEventModel.created_at,
+                                    )
+                                ),
+                            )
                             .limit(1)
                         )
                     ).first()
-                    if prior is not None:
+                    if terminal is not None:
                         # A first writer may have committed between the head
                         # and event reads. Only that race warrants a retry.
                         if await session.get(AuditChainHeadModel, wallet_key):
                             raise _HeadConflict()
-                        raise AuditChainHeadMissingError("audit_chain_head_missing")
+                        if terminal[0] != 0:
+                            # A seq-signed row exists, so a head was written
+                            # and has since gone. Refuse rather than guess.
+                            raise AuditChainHeadMissingError("audit_chain_head_missing")
+                        # Every stored row predates seq (migration 017 set
+                        # them to 0) and no head was ever written. Continue
+                        # the legacy chain from its terminal link at seq 1,
+                        # which verification accepts after a legacy run.
+                        previous_hash = terminal[1]
                 model.seq = observed_seq + 1
                 _sign_with_previous(model, previous_hash, signing_key_id=key.key_id)
                 now = utc_now()
diff --git a/gateway/tests/test_audit_chain_tamper.py b/gateway/tests/test_audit_chain_tamper.py
index 742beaf3..2fdb9434 100644
--- a/gateway/tests/test_audit_chain_tamper.py
+++ b/gateway/tests/test_audit_chain_tamper.py
@@ -554,3 +554,25 @@ async def test_legacy_seq_zero_chain_without_a_head_still_verifies(clean_databas
     result = await verify_audit_chain(wallet_id=wallet_id)
     assert result.valid is True, result
     assert result.checked_events == 2
+
+    # The next append continues the legacy chain instead of refusing it.
+    await record_audit_event(
+        event="trust.legacy", wallet_id=wallet_id, metadata={"n": 2}
+    )
+    async with factory() as session:
+        appended = (
+            await session.execute(
+                select(ControlPlaneAuditEventModel).where(
+                    ControlPlaneAuditEventModel.wallet_id == wallet_id,
+                    ControlPlaneAuditEventModel.seq == 1,
+                )
+            )
+        ).scalar_one()
+        assert appended.previous_hash == previous_hash
+        head = await session.get(AuditChainHeadModel, wallet_id)
+        assert head is not None
+        assert head.last_seq == 1
+        assert head.last_chain_hash == appended.chain_hash
+    result = await verify_audit_chain(wallet_id=wallet_id)
+    assert result.valid is True, result
+    assert result.checked_events == 3