21 changed files with 105 additions and 1771 deletions
--- a/.gitignore
+++ b/.gitignore
@ -3,5 +3,3 @@ __pycache__/
 *.db
 *.key
 *.pem
 *.tip
 *.tip.log
--- a/DESIGN.md
+++ b/DESIGN.md
@ -121,13 +121,9 @@ means a future route cannot pass a string where a proof belongs.
 ### Suspension is not death
-`revoke <agent>` with no end time is permanent: the enrolled public key is
+`revoke <agent>` with no end time is permanent: the identity is dead until
-cleared and the identity stays dead until someone bootstraps a fresh secret and
+someone re-enrols it. `revoke <agent> --until <unix>` is a pause, and it lapses
-calls `enrol_rekey` (then `reinstate` if still banned). `enrol` will not work —
+on its own.
 the agent row still exists — and would only burn the seed. `reinstate` alone
 lifts the ban flags but does **not** restore the compromised key — that was the
 footgun. `revoke <agent> --until <unix>` is a pause that keeps the key, and it
 lapses on its own.
 The distinction matters because the two failure modes are different. A key
 compromise wants death — a lapsing revocation would hand the attacker back in
@ -135,18 +131,8 @@ when the timer ran out, which is a revocation that silently fails at exactly the
 wrong moment. A noisy neighbour or a holiday wants a pause, and re-enrolling for
 that would destroy the audit history that says it was deliberate.
-Revoking kills live sessions, drops outstanding challenges, and invalidates
+Revoking kills live sessions and drops outstanding challenges, so a revocation
-unused view codes. Rekey invalidates codes the same way: rotating a key must
+that races an in-flight authentication cannot be overtaken by it.
 close every door the old key could open.
 A narrow check-then-insert race can still insert a session or code after the
 revoke/rekey statement has run. The choke point is `sessions.resolve`: every
 use re-checks that the agent is not revoked, still has a live key, and that
 the session fingerprint (when present — always for agent sessions, and for
 human_via_agent sessions minted with a fingerprint) matches the current key
 epoch. View codes store the issuing fingerprint and refuse redeem after rekey
 or key wipe. That is what makes revocation overtake in-flight
 verify/issue/redeem, not the first-pass invalidation alone.
 ### Existing controls stay
@ -204,23 +190,15 @@ history, forges replacements, and rebuilds the chain — `verify_chain` passes.
 So the head digest also gets **signed checkpoints**, by an enrolled agent whose
 private key the database does not contain. `verify-chain` runs both checks, and
 the rewrite is caught by the signature the attacker cannot produce. A forged
-checkpoint fails signature verification, and a checkpoint from a stranger is
+checkpoint fails signature verification, a checkpoint from a stranger is refused
-refused at signing time.
+at signing time, and a rekey invalidates every earlier checkpoint — which is
-
+correct, and says so rather than failing quietly.
 Public keys are kept in an append-only `agent_keys` history. Revoke and rekey
 *retire* a key rather than destroy it, so honest pre-compromise checkpoints and
 off-box anchors stay verifiable against the fingerprint they name. Checkpoints
 timestamped after a key's `retired_at` are flagged (not failed). Prefer a
 **dedicated checkpoint-signer identity** that is never a working agent, so a
 routine agent compromise-revoke does not touch the tip-signing key at all.
 What this still does not do is survive an attacker who can also delete the
-checkpoints, because they live in the same database. Set `anchor_path` /
+checkpoints, because they live in the same database. Real deployment anchors the
-`SIGIL_ANCHOR` and every checkpoint also publishes the signed tip to a file
+head digest somewhere the database cannot reach. This narrows the attack from
-outside the SQLite path; `verify_external_anchors` then catches a rewrite even
+"rewrite anything" to "erase the record that you did", and leaves that second
-after the in-DB checkpoints table is wiped. Without that path the tip stays
+half open.
 in-DB only, and deleting the checkpoints remains an option for whoever holds
 the file.
 Secrets are not in the log and not in the database in the clear. Codes and
 session tokens are stored as digests, enrolment secrets are hashed and spent
@ -231,7 +209,7 @@ because the review looked for it.
 ## Status
-158 passing tests, and two hardening passes that closed real defects. The worst were
+106 passing tests, and two hardening passes that closed real defects. The worst were
 concurrency: one-time codes were redeemable repeatedly (753 sessions from 200
 codes under load), and a bootstrap secret could enrol more than one key. Those
 are now atomic compare-and-set operations, each covered by a test that races
@ -244,20 +222,16 @@ sessions mintable from a bare agent name, no revocation operation at all, an
 unkeyed chain that a database-level attacker could rebuild, and SQLite without
 a busy timeout.
-Leora's contracted prod pass (Quarry 18a5d762) closed three remaining review blockers on this branch: rekey no longer un-revokes, dual-bootstrap enrol is exactly-one under race, and redeem rate limits are per-agent (anonymous fails isolated from success). Off-box checkpoint tip publish/verify ships when `anchor_path` / `SIGIL_ANCHOR` is set. SealedStore / `SIGIL_SEAL_PASS` seals the DB file with AES-256-GCM (incremental path short of SQLCipher). Independent review returned GO-WITH-FIXES; R1–R8/R11 fixes on this branch. Platform glue remains off.
+Leora's contracted prod pass (Quarry 18a5d762) closed three remaining review blockers on this branch: rekey no longer un-revokes, dual-bootstrap enrol is exactly-one under race, and redeem rate limits are per-agent (anonymous fails isolated from success). Independent review, off-box checkpoint anchor, sealed store, and platform glue remain.
 **Still not production ready**, and the reason is not a to-do list:
 - **No independent review.** Everything above is self-review, twice over.
  Self-review finds the bugs you already suspected. This needs someone else,
  ideally a cryptographer, and it is the one blocker code cannot close.
- **Off-box tip is opt-in.** With `anchor_path` set, every checkpoint publishes a
+- **Checkpoints are not anchored off-box.** Signed, yes; deleteable by the same
-  signed tip outside the SQLite file and `verify_external_anchors` catches a DB
+  attacker along with everything else, yes. See *Audit trail*.
-  rewrite that in-DB checkpoints alone would miss after deletion. Leave it unset
+- **The service is unwritten.** TLS, CSRF, transport, secret storage, routes.
  and the tip stays in-DB only.
 - **The service is unwritten.** TLS, CSRF, transport, routes. DB file seal is
  available (`SealedStore`, plaintext working copy `0600`); SQLCipher
  page-level encryption is deferred (not yet available).
 - **Rate limits are per-process**, so N processes means N times the quota.
 - **No key recovery.** Rotate and old sessions die. Deliberate.
 - **Humans remain unattributable** to themselves, which is accepted and is a
@ -267,27 +241,11 @@ The tests show the logic is correct. They do not make it safe.
 ## Next
-1. ~~Independent security review~~ — independent review returned GO-WITH-FIXES
+1. Independent security review before this guards anything real. Two hardening
-   (tip c0027bb); fix PRs for R1–R8/R11 landed on `sigil-initial`. Platform glue
+   passes made it reviewable; neither substituted for one.
-   remains off until integration review. See `docs/PROD_RUNBOOK.md`.
+2. Anchor the checkpoint head digest outside the database, so deleting the
-2. ~~Anchor the checkpoint head digest outside the database~~ — done via
+   checkpoints stops being an option for an attacker who already has the file.
   `Store(..., anchor_path=)` / `SIGIL_ANCHOR` (signed tip file + `.log`).
   **Prod must set it** (see runbook).
 3. Secret broker on top — scoped, short-lived leases bound to a key fingerprint.
   Its issuance and redemption logs *are* the work-attestation substrate, so it
   is the natural second piece.
-4. Deprecate passcode login across the platform, then rotate live credentials.
+4. Deprecate passcode login across the platform, then rotate live credentials.
 ### Info / follow-ups (no code change required yet)
 - **R9 — `issue_agent_session`:** library helper mints `agent:act` from a name
  after the caller claims to have verified. Keep internal; integration must
  never mint from headers/names. Consider renaming to `_issue_agent_session`.
 - **R10 — challenge domain separation:** signatures currently cover
  `nonce.encode()` with no context prefix. Adding
  `b"sigil-challenge-v1\0" + nonce` is not backward-compatible with in-flight
  challenges; schedule with a protocol version bump (follow-up).
 - **R12 — multi-process rate limits:** quotas are per-`Store` / per-process.
  Single writer until a shared limiter exists (documented in README + runbook).
 - **R13 — opt-in anchor:** without `SIGIL_ANCHOR`, deleting in-DB checkpoints
  allows silent history rewrite. Runbook requires the anchor on separate storage.
--- a/README.md
+++ b/README.md
@ -28,7 +28,6 @@ is **not independently reviewed** — see *Status* below.
 ```sh
 export SIGIL_DB=/tmp/demo.db
 # optional: export SIGIL_SEAL_PASS='...'  # AES-GCM seal the DB file at rest
 python3 sigil.py seed bigpickle                       # -> one-time secret
 python3 sigil.py keygen /tmp/demo.key                 # -> public_b64, fingerprint
@ -42,9 +41,8 @@ python3 sigil.py code bigpickle /tmp/demo.key         # hand a human a code
 python3 sigil.py redeem X3ZB-NTLG                     # -> human_via_agent token
 python3 sigil.py whoami "$TOKEN"                      # kind, scopes, may_act_as_agent
 python3 sigil.py events bigpickle                     # audit trail
-export SIGIL_ANCHOR=/tmp/demo.chain.tip               # off-box tip path
+python3 sigil.py checkpoint /tmp/demo.key bigpickle   # sign the audit head
-python3 sigil.py checkpoint /tmp/demo.key bigpickle   # sign head + publish tip
+python3 sigil.py verify-chain                         # chain + checkpoints
 python3 sigil.py verify-chain                         # chain + checkpoints + tip
 python3 sigil.py revoke bigpickle --reason "incident" # or --until UNIX
 python3 sigil.py sweep                                # prune stale rows
 ```
@ -55,11 +53,10 @@ python3 sigil.py sweep                                # prune stale rows
 python3 -m unittest discover -s tests -v
 ```
-158 tests. They cover the refusals, not just the happy paths: replayed
+111 tests. They cover the refusals, not just the happy paths: replayed
 signatures, wrong keys, expired nonces, single-use codes, scope escalation, and
-the invariant below. `test_hardening.py`, `test_production.py`, `test_anchor.py` and
+the invariant below. `test_hardening.py` and `test_production.py` have one test
-`test_sealed.py` have one test per defect found in review, so any of them
+per defect found in review, so any of them regressing brings that defect back.
 regressing brings that defect back.
 ## The invariant
@ -125,8 +122,6 @@ A hardening pass closed the defects below. Each is now covered by a test.
 | Rekey cleared `revoked_at` (INSERT OR REPLACE) | Rekey refuses while revoked; `update_agent_key` preserves ban columns |
 | Dual-bootstrap enrol race (last writer won) | `put_agent` is INSERT OR IGNORE; exactly one enrol wins |
 | Redeem fail limit was one global bucket | Per-agent `code_redeem` / `code_redeem_fail`; anonymous fails isolated |
 | Checkpoints lived only inside the same DB | Optional off-box signed tip (`SIGIL_ANCHOR` / `anchor_path`) |
 | DB file held roster/audit in cleartext | Optional `SealedStore` / `SIGIL_SEAL_PASS` (AES-256-GCM file seal) |
 Still open, and these are the reasons not to ship it yet:
@ -134,17 +129,14 @@ Still open, and these are the reasons not to ship it yet:
  review finds the bugs you thought of; it does not find the ones you did not.
  This needs someone else, ideally a cryptographer, before it guards anything
  real. This is the blocking item, and the only one that code cannot close.
- **The service around this is unwritten.** TLS, CSRF, session transport and the
+- **The service around this is unwritten.** TLS, CSRF, session transport, secret
-  HTTP routes. DB encryption at rest is available via `SealedStore` /
+  storage and the HTTP routes. The library refuses to mint authority from a
-  `SIGIL_SEAL_PASS` (AES-256-GCM file seal; not SQLCipher page-level). The
+  name, so a route that forgets to authenticate fails closed, but that is only
-  plaintext working copy is mode `0600` in a private temp dir while open;
+  as good as the integration.
-  SQLCipher remains deferred for hostile co-resident hosts. The library
+- **Checkpoints need somewhere to live.** A signed checkpoint inside the same
-  refuses to mint authority from a name, so a route that forgets to
+  database is still only as good as the database. Real deployment wants the head
-  authenticate fails closed, but that is only as good as the integration.
+  digest anchored off-box — published, or countersigned elsewhere — and this
- **Off-box anchor is opt-in.** Set `SIGIL_ANCHOR` (or `Store(..., anchor_path=)`)
+  does not do that for you.
  and every checkpoint also writes a signed tip outside the SQLite file.
  Rewriting the DB alone can no longer erase the tip. Without that path
  configured, checkpoints remain in-DB only — fine for tests, not for prod.
 - **No key recovery.** Rotate and the old sessions die, which is the point. There
  is no escrow. Deliberate, not an oversight.
 - **Rate limits are in-process.** They are per-`Store`, so several processes
@ -158,9 +150,4 @@ The tests assert the logic is correct. They do not make it safe.
 ## See also
 `DESIGN.md` — the design page: rationale, flows, session stamps, the invariant
-and its reasoning, and what is deferred.
+and its reasoning, and what is deferred.
 `docs/PROD_RUNBOOK.md` — production checklist: require `SIGIL_SEAL_PASS` and
 `SIGIL_ANCHOR` on separate storage, verify-chain alerts, integration sharp edges.
 `requirements.txt` — `cryptography>=44.0.1,<45`; see `requirements.lock.txt` / `constraints.txt`.
--- a/constraints.txt
+++ b/constraints.txt
@ -1,3 +0,0 @@
 # Exact pin matching requirements.lock.txt floor. Use with:
 #   pip install -r requirements.txt -c constraints.txt
 cryptography==44.0.1
--- a/docs/PROD_RUNBOOK.md
+++ b/docs/PROD_RUNBOOK.md
@ -1,46 +0,0 @@
 # Sigil production runbook
 Short checklist before Sigil guards anything real. Library-only today — do **not**
 enable platform glue until an integration review covers session minting.
 ## Required environment
 | Variable | Purpose | Notes |
 |----------|---------|--------|
 | `SIGIL_DB` | Path to the store file | Prefer a dedicated volume; backups are sensitive (roster + audit). |
 | `SIGIL_SEAL_PASS` | AES-256-GCM seal passphrase | **Required in prod.** ≥12 chars. Without it the DB file is cleartext on disk. Seal-on still uses a plaintext `0600` working copy while the process is open — not SQLCipher. |
 | `SIGIL_ANCHOR` | Off-box signed tip path | **Required in prod.** Must be on storage **separate** from `SIGIL_DB` (different volume/host). Every checkpoint publishes here; `verify_external_anchors` / `verify-chain` must alert on failure. |
 Optional: agent private keys stay **off** the platform host whenever possible; checkpoint signer key preferably offline.
 ## Dedicated checkpoint-signer identity
 Enrol a separate agent id used **only** for `checkpoint` / `publish_anchor`
 (for example `sigil-checkpoint`). Do not use a working agent as the tip signer:
 compromising and permanently revoking that agent would otherwise retire its key
 (history still verifies, but ops alerting on "key retired" gets noisy). Never
 store the checkpoint signer's private key in the Sigil DB — public history only.
 ## Ops loop
 1. Open with seal + anchor configured (CLI respects the env vars above).
 2. Periodically run chain verify **and** external-anchor verify; page on failure.
 3. After permanent revoke (compromise): bootstrap + `enrol_rekey`, then `reinstate`. Reinstate alone does not restore a wiped key.
 4. Single writer process until a shared rate-limit store exists (in-process quotas multiply across processes).
 ## Integration sharp edges (do not ignore)
 - Never mint `issue_agent_session` from a client-supplied name or header. Only after a verified challenge-response. Prefer keeping that helper internal to verified paths.
 - Human view codes are ~40 bits; keep TTL short; never log raw codes (already hashed in audit).
 - Challenge signatures today sign the raw nonce (no domain separation). Changing the signed payload needs a coordinated version bump — tracked as a follow-up, not a silent break.
 ## Deferred
 - SQLCipher / page-level encryption for multi-tenant co-resident hosts.
 - Domain-separated challenge payloads (`sigil-challenge-v1` prefix) — backward-incompatible without a protocol version.
 - Shared cross-process rate limiter.
 ## Dependencies
 Install with `pip install -r requirements.txt -c constraints.txt`
 (or `--require-hashes -r requirements.lock.txt`). Floor is `cryptography>=44.0.1`.
--- a/requirements.lock.txt
+++ b/requirements.lock.txt
@ -1,13 +0,0 @@
 # Hash-pinned lock for Sigil runtime crypto deps.
 # Install: pip install --require-hashes -r requirements.lock.txt
 # Floor cryptography==44.0.1 addresses CVE-2024-12797 in older wheels.
 cryptography==44.0.1 \
    --hash=sha256:53f23339864b617a3dfc2b0ac8d5c432625c80014c25caac9082314e9de56f41
 cffi==2.1.1 \
    --hash=sha256:a931079504ecc49efed7744c476a5c343a92fabf66dec2db95edb1b2fdc770e2
 pycparser==3.0 \
    --hash=sha256:b727414169a36b7d524c1c3e31839a521725078d7b2ff038656844266160a992
--- a/requirements.txt
+++ b/requirements.txt
@ -1,5 +0,0 @@
 # Runtime dependency for Ed25519 + AES-GCM (SealedStore / optional key files).
 # Lower bound 44.0.1: older 43.x/44.0.0 wheels ship OpenSSL affected by
 # CVE-2024-12797 (GHSA-79v4-65xg-pq4g). Upper bound <45 keeps upgrades deliberate.
 # For hash-pinned install see requirements.lock.txt.
 cryptography>=44.0.1,<45
--- a/sigil.py
+++ b/sigil.py
@ -6,8 +6,7 @@
    sigil.py enrol <agent> <pubkey.b64> <secret>
    sigil.py challenge <agent>                nonce to sign
    sigil.py checkpoint <key> <agent> [label] sign the current audit head
-                                              (also publishes off-box when SIGIL_ANCHOR is set)
+    sigil.py verify-chain                     check the chain and its checkpoints
    sigil.py verify-chain                     check chain, checkpoints, and off-box tip
    sigil.py revoke <agent> [--until UNIX] [--reason R] [--actor A]
                                         --until omitted means permanent
    sigil.py reinstate <agent> [actor]       lift a revocation
@ -19,9 +18,7 @@
    sigil.py whoami <token>                   describe a session
    sigil.py events [actor]                   audit trail
-Uses SIGIL_DB (default ./sigil.db), SIGIL_KEY (default ./agent.key),
+Uses SIGIL_DB (default ./sigil.db) and SIGIL_KEY (default ./agent.key).
 optional SIGIL_ANCHOR (filesystem path for the off-box signed tip), and
 optional SIGIL_SEAL_PASS (when set, SIGIL_DB is an AES-GCM sealed file).
 """
 from __future__ import annotations
@ -32,13 +29,10 @@ import os
 import sys
 from sigil import errors, identity, keys, sessions, viewcodes
 from sigil.sealed import SealedStore, is_sealed_file
 from sigil.store import Store
 DB = os.environ.get("SIGIL_DB", "./sigil.db")
 KEY = os.environ.get("SIGIL_KEY", "./agent.key")
 ANCHOR = os.environ.get("SIGIL_ANCHOR")  # off-box tip path; unset = in-DB only
 SEAL_PASS = os.environ.get("SIGIL_SEAL_PASS")  # seal passphrase; unset = cleartext DB
 def out(obj) -> None:
@ -50,14 +44,7 @@ def main(argv) -> int:
        print(__doc__)
        return 2
    cmd = argv[1]
-    store = None
+    store = Store(DB)
    if SEAL_PASS or (os.path.exists(DB) and is_sealed_file(DB)):
        if not SEAL_PASS:
            print('{"error":"ValueError","message":"SIGIL_DB is sealed; set SIGIL_SEAL_PASS"}')
            return 1
        store = SealedStore(DB, SEAL_PASS, anchor_path=ANCHOR)
    else:
        store = Store(DB, anchor_path=ANCHOR)
    try:
        if cmd == "seed":
@ -128,12 +115,9 @@ def main(argv) -> int:
        elif cmd == "verify-chain":
            chain = store.verify_chain()
-            checkpoints = store.verify_checkpoints()
+            anchors = store.verify_checkpoints()
-            external = store.verify_external_anchors()
+            out({"chain": chain, "checkpoints": anchors})
-            out({"chain": chain, "checkpoints": checkpoints,
+            return 0 if chain["ok"] and anchors["ok"] else 1
                 "external_anchors": external})
            ok = chain["ok"] and checkpoints["ok"] and external["ok"]
            return 0 if ok else 1
        elif cmd == "revoke":
            # Positional optional args are a trap here: "revoke alice please"
@ -195,26 +179,15 @@ def main(argv) -> int:
    except errors.SigilError as exc:
        print(json.dumps({"error": type(exc).__name__, "message": str(exc)}))
        _flush_if_sealed(store)
        return 1
    except (ValueError, OSError) as exc:
        # Wrong passphrase, unreadable key file. Report it as a plain result
        # rather than a traceback: it is an ordinary operator mistake.
        print(json.dumps({"error": type(exc).__name__, "message": str(exc)[:200]}))
        _flush_if_sealed(store)
        return 1
    _flush_if_sealed(store)
    return 0
 def _flush_if_sealed(store) -> None:
    if store is not None and isinstance(store, SealedStore):
        try:
            store.flush()
        except Exception:
            pass
 if __name__ == "__main__":
    raise SystemExit(main(sys.argv))
--- a/sigil/init.py
+++ b/sigil/init.py
@ -31,20 +31,16 @@ without the private key, which never leaves the agent.
 import time
-from . import anchor, errors, identity, keys, sealed, sessions, store, viewcodes
+from . import errors, identity, keys, sessions, store, viewcodes
 from .store import Store
 from .sealed import SealedStore
-__version__ = "0.4.1"
+__version__ = "0.3.0"
 __all__ = [
    "Store",
    "SealedStore",
    "anchor",
    "errors",
    "identity",
    "keys",
    "sealed",
    "sessions",
    "store",
    "viewcodes",
--- a/sigil/anchor.py
+++ b/sigil/anchor.py
@ -1,122 +0,0 @@
 """Off-box anchors for the audit chain tip.
 A signed checkpoint inside the same SQLite file as the events is still only as
 strong as that file: an attacker who can rewrite the database can delete the
 checkpoints table along with everything else. Publishing the signed tip to a
 path *outside* the database closes that. Rewriting the DB alone cannot erase
 the tip the operator already published.
 The notice is the same canonical payload the in-DB checkpoint signs, plus the
 signature. verify_anchor() checks the signature against the enrolled public
 key and reports whether the live chain still ends at that head.
 """
 from __future__ import annotations
 import base64
 import json
 import os
 from typing import Any, Dict, List, Optional
 def notice_bytes(notice: Dict[str, Any]) -> bytes:
    """Canonical encoding of a published tip, excluding the signature itself."""
    body = {
        "at": int(notice["at"]),
        "label": notice.get("label") or "",
        "head": notice["head"],
        "events": int(notice["events"]),
        "signer": notice["signer"],
        "fingerprint": notice["fingerprint"],
    }
    return json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8")
 def write_tip(path: str, notice: Dict[str, Any]) -> str:
    """Atomically write the latest tip notice to `path`.
    Also appends one JSON line to `path + ".log"` so earlier tips remain
    auditable after a later checkpoint overwrites the tip file.
    """
    if not path or path == ":memory:":
        raise ValueError("anchor path must be a real filesystem path")
    payload = {
        "at": int(notice["at"]),
        "label": notice.get("label") or "",
        "head": notice["head"],
        "events": int(notice["events"]),
        "signer": notice["signer"],
        "fingerprint": notice["fingerprint"],
        "signature": notice["signature"],
    }
    text = json.dumps(payload, sort_keys=True, indent=2) + "\n"
    directory = os.path.dirname(os.path.abspath(path))
    if directory:
        os.makedirs(directory, exist_ok=True)
    fd, tmp = _tmp_beside(path)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            handle.write(text)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(tmp, path)
    except Exception:
        try:
            os.unlink(tmp)
        except OSError:
            pass
        raise
    # Append-only history beside the tip. Best-effort: tip write already landed.
    try:
        with open(path + ".log", "a", encoding="utf-8") as log:
            log.write(json.dumps(payload, sort_keys=True, separators=(",", ":")) + "\n")
    except OSError:
        pass
    return path
 def read_tip(path: str) -> Optional[Dict[str, Any]]:
    """Load the tip notice, or None if the file is missing."""
    if not path or not os.path.exists(path):
        return None
    with open(path, "r", encoding="utf-8") as handle:
        return json.loads(handle.read())
 def read_log(path: str) -> List[Dict[str, Any]]:
    """Load every historically published tip from the sibling .log file."""
    log_path = path + ".log"
    if not os.path.exists(log_path):
        tip = read_tip(path)
        return [tip] if tip else []
    out: List[Dict[str, Any]] = []
    with open(log_path, "r", encoding="utf-8") as handle:
        for line in handle:
            line = line.strip()
            if line:
                out.append(json.loads(line))
    return out
 def verify_notice(notice: Dict[str, Any], public_key) -> Dict[str, Any]:
    """Check one notice's signature. `public_key` is an Ed25519 public key."""
    from . import keys as _keys
    try:
        payload = notice_bytes(notice)
        sig = base64.b64decode(notice["signature"])
        public_key.verify(sig, payload)
    except Exception as exc:
        return {"ok": False, "reason": f"signature check failed: {exc}"}
    expected_fp = _keys.fingerprint(public_key)
    if notice.get("fingerprint") != expected_fp:
        return {"ok": False, "reason": "fingerprint does not match signer key"}
    return {"ok": True, "head": notice["head"], "events": notice["events"],
            "signer": notice["signer"]}
 def _tmp_beside(path: str):
    directory = os.path.dirname(os.path.abspath(path)) or "."
    return __import__("tempfile").mkstemp(
        prefix=".sigil-anchor-", suffix=".tmp", dir=directory
    )
--- a/sigil/identity.py
+++ b/sigil/identity.py
@ -56,9 +56,7 @@ def seed_bootstrap(store: Store, agent: str, secret: Optional[str] = None) -> st
    """
    _validate_agent_id(agent)
    row = store.get_agent(agent)
-    # Timed suspension: refuse. Permanent revoke (no live key): allow so ops
+    if row is not None and is_revoked(row):
    # can stage a rekey before or after reinstate.
    if row is not None and is_revoked(row) and has_live_key(row):
        raise errors.UnknownAgent(f"{agent} is revoked")
    secret = secret or secrets.token_urlsafe(24)
    store.put_bootstrap(_hash_secret(secret), agent)
@ -78,10 +76,6 @@ def enrol(
    _validate_agent_id(agent)
    if not store.allow(f"enrol:{agent}"):
        raise errors.RateLimited(f"too many enrolment attempts for {agent}")
    # Already enrolled (e.g. after permanent revoke): refuse *before* spending
    # the seed so ops following a wrong "re-enrol" path do not burn it.
    if store.get_agent(agent) is not None:
        raise errors.EnrolmentFailed(f"{agent} is already enrolled; use enrol_rekey")
    # Spend the secret first so a burned secret cannot be retried. The agent
    # row insert below is INSERT OR IGNORE: if two secrets race for the same
    # unenrolled agent, exactly one put_agent wins and the loser fails closed
@ -94,8 +88,6 @@ def enrol(
    fingerprint = keys.fingerprint(public)
    if not store.put_agent(agent, keys.public_bytes(public), fingerprint):
        raise errors.EnrolmentFailed(f"{agent} is already enrolled")
    # Spend any sibling seeds left over from double-seed / unused chat leaks.
    store.clear_bootstrap_for_agent(agent)
    store.record(
        "identity.enrolled",
        actor=agent,
@ -120,18 +112,6 @@ def is_revoked(row, now: Optional[float] = None) -> bool:
    return (now if now is not None else time.time()) < until
 def has_live_key(row) -> bool:
    """Does this agent row still hold an enrolled public key?
    Permanent revoke clears pubkey/fingerprint so reinstate alone cannot bring
    a compromised key back. Callers that mint challenges or sessions must
    refuse until bootstrap + enrol_rekey restores a real key.
    """
    fp = row["fingerprint"] if "fingerprint" in row.keys() else None
    pubkey = row["pubkey"] if "pubkey" in row.keys() else None
    return bool(fp) and isinstance(pubkey, (bytes, bytearray)) and len(pubkey) == 32
 def revoke(store: Store, agent: str, until: Optional[int] = None,
           reason: Optional[str] = None, actor: str = "system") -> int:
    """Revoke an agent, optionally until a time.
@ -162,14 +142,7 @@ def revoke(store: Store, agent: str, until: Optional[int] = None,
 def reinstate(store: Store, agent: str, actor: str = "system") -> None:
-    """Lift a revocation. Audited, because it undoes a security decision.
+    """Lift a revocation. Audited, because it undoes a security decision."""
    After a *permanent* revoke the enrolled key was cleared. Reinstate lifts
    the ban flags only — the agent still cannot authenticate until ops seed a
    bootstrap secret and call enrol_rekey with a fresh key. Timed suspensions
    keep the key, so reinstate (or waiting out the window) restores the same
    identity without rotation.
    """
    _validate_agent_id(agent)
    if store.get_agent(agent) is None:
        raise errors.UnknownAgent(f"{agent} is not enrolled")
@ -188,10 +161,6 @@ def challenge(store: Store, agent: str, ttl: int = DEFAULT_NONCE_TTL,
        raise errors.UnknownAgent(f"{agent} is not enrolled")
    if is_revoked(row):
        raise errors.UnknownAgent(f"{agent} is revoked")
    if not has_live_key(row):
        raise errors.UnknownAgent(
            f"{agent} has no live key; bootstrap and enrol_rekey required"
        )
    nonce = secrets.token_urlsafe(24)
    expires = int(time.time()) + ttl
    store.put_nonce(nonce, agent, expires)
@ -250,10 +219,6 @@ def verify(
        raise errors.UnknownAgent(f"{agent} is not enrolled")
    if is_revoked(agent_row):
        raise errors.UnknownAgent(f"{agent} is revoked")
    if not has_live_key(agent_row):
        raise errors.UnknownAgent(
            f"{agent} has no live key; bootstrap and enrol_rekey required"
        )
    public = keys.from_public_bytes(agent_row["pubkey"])
    try:
@ -292,17 +257,10 @@ def enrol_rekey(store: Store, agent: str, public_b64: str, secret: str) -> str:
    row = store.get_agent(agent)
    if row is None:
        raise errors.UnknownAgent(f"{agent} is not enrolled")
-    # Permanent revoke clears the key. Allow rekey while still revoked so ops
+    if is_revoked(row):
    # can rotate before reinstate; a revoked agent that still has a live key
    # (timed suspend) must not rekey as a back door.
    if is_revoked(row) and has_live_key(row):
        raise errors.UnknownAgent(f"{agent} is revoked")
    owner = store.spend_bootstrap(_hash_secret(secret))
    if owner != agent:
        # Rate-limit only failed attempts so junk cannot burn the ops path, and
        # use a dedicated rekey: bucket so exhausting enrol: cannot lock rekey.
        if not store.allow(f"rekey:{agent}"):
            raise errors.RateLimited(f"too many rekey attempts for {agent}")
        raise errors.EnrolmentFailed("bootstrap secret invalid or already spent")
    public = keys.import_public_b64(public_b64)
    fingerprint = keys.fingerprint(public)
@ -312,8 +270,6 @@ def enrol_rekey(store: Store, agent: str, public_b64: str, secret: str) -> str:
    # A new key must not leave old sessions alive, or rotating after a
    # compromise buys nothing.
    killed = store.revoke_sessions_for_agent(agent, "key rotated")
    # Leftover bootstrap seeds must not remain a standing rekey capability.
    store.clear_bootstrap_for_agent(agent)
    store.record(
        "identity.rekeyed",
        actor=agent,
--- a/sigil/sealed.py
+++ b/sigil/sealed.py
@ -1,227 +0,0 @@
 """Sealed store: encryption at rest for the Sigil database file.
 Tokens, codes and bootstrap secrets are already stored as digests, so a stolen
 cleartext SQLite file does not yield live credentials. It does still yield the
 full agent roster, revocation state, and audit history. A sealed store wraps
 the on-disk bytes in AES-256-GCM so that material is unreadable without the
 passphrase.
 This is an incremental path short of SQLCipher (no page-level encryption, the
 working copy is plaintext in memory / a temp file while open). It is a real
 working seal: the file on disk is ciphertext, and tests assert a stolen seal
 has no SQLite header and no recoverable agent rows without the passphrase.
 The plaintext working copy is created mode 0600 inside a private temp dir.
 SQLCipher (or equivalent page-level encryption) remains a deferred prod
 follow-up for multi-tenant / hostile co-resident hosts — never treat seal-on
 as "encrypted DB" while the process is running.
 Format on disk (version 1):
    b"SIGILSEAL1" + 16-byte salt + 12-byte nonce + AESGCM(ciphertext)
    key = PBKDF2-HMAC-SHA256(passphrase, salt, 480_000 iterations, 32 bytes)
 """
 from __future__ import annotations
 import os
 import sqlite3
 import tempfile
 from typing import Optional
 from cryptography.hazmat.primitives.ciphers.aead import AESGCM
 from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
 from cryptography.hazmat.primitives import hashes
 from .store import Store
 MAGIC = b"SIGILSEAL1"
 SALT_LEN = 16
 NONCE_LEN = 12
 KEY_LEN = 32
 PBKDF2_ITERS = 480_000
 MIN_PASSPHRASE = 12
 def derive_key(passphrase: str, salt: bytes) -> bytes:
    if len(passphrase) < MIN_PASSPHRASE:
        raise ValueError(
            f"seal passphrase must be at least {MIN_PASSPHRASE} characters"
        )
    kdf = PBKDF2HMAC(
        algorithm=hashes.SHA256(),
        length=KEY_LEN,
        salt=salt,
        iterations=PBKDF2_ITERS,
    )
    return kdf.derive(passphrase.encode("utf-8"))
 def seal_bytes(plaintext: bytes, passphrase: str, *, salt: Optional[bytes] = None) -> bytes:
    """Encrypt plaintext DB bytes into a sealed blob."""
    salt = salt or os.urandom(SALT_LEN)
    key = derive_key(passphrase, salt)
    nonce = os.urandom(NONCE_LEN)
    ct = AESGCM(key).encrypt(nonce, plaintext, MAGIC)
    return MAGIC + salt + nonce + ct
 def unseal_bytes(blob: bytes, passphrase: str) -> bytes:
    """Decrypt a sealed blob back to plaintext DB bytes."""
    if not blob.startswith(MAGIC):
        raise ValueError("not a Sigil sealed store (bad magic)")
    salt = blob[len(MAGIC):len(MAGIC) + SALT_LEN]
    nonce_start = len(MAGIC) + SALT_LEN
    nonce = blob[nonce_start:nonce_start + NONCE_LEN]
    ct = blob[nonce_start + NONCE_LEN:]
    key = derive_key(passphrase, salt)
    try:
        return AESGCM(key).decrypt(nonce, ct, MAGIC)
    except Exception as exc:
        raise ValueError("wrong passphrase or corrupted sealed store") from exc
 def is_sealed_file(path: str) -> bool:
    if not os.path.exists(path):
        return False
    with open(path, "rb") as handle:
        return handle.read(len(MAGIC)) == MAGIC
 def _write_private_file(path: str, data: bytes) -> None:
    """Write `data` to `path` with mode 0600, fsync, unlink on failure."""
    flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
    fd = os.open(path, flags, 0o600)
    try:
        with os.fdopen(fd, "wb") as handle:
            fd = -1  # ownership transferred
            handle.write(data)
            handle.flush()
            os.fsync(handle.fileno())
        os.chmod(path, 0o600)
    except Exception:
        if fd >= 0:
            try:
                os.close(fd)
            except OSError:
                pass
        try:
            os.unlink(path)
        except OSError:
            pass
        raise
 class SealedStore(Store):
    """Store whose on-disk file is AES-GCM sealed.
    Open decrypts into a temp SQLite file (or creates an empty one). Call
    `flush()` or `close()` to re-encrypt and write the seal back to `path`.
    The passphrase never touches the SQLite file; only the seal blob does.
    """
    def __init__(
        self,
        path: str,
        passphrase: str,
        *,
        anchor_path: Optional[str] = None,
    ):
        if path == ":memory:":
            raise ValueError("SealedStore needs a filesystem path to seal")
        if len(passphrase) < MIN_PASSPHRASE:
            raise ValueError(
                f"seal passphrase must be at least {MIN_PASSPHRASE} characters"
            )
        self.seal_path = path
        self._passphrase = passphrase
        self._tmpdir = tempfile.TemporaryDirectory(prefix="sigil-seal-")
        # TemporaryDirectory is already 0700; pin the dir mode explicitly.
        os.chmod(self._tmpdir.name, 0o700)
        self._plain_path = os.path.join(self._tmpdir.name, "store.db")
        if os.path.exists(path):
            with open(path, "rb") as handle:
                blob = handle.read()
            if not blob.startswith(MAGIC):
                raise ValueError(
                    f"{path} exists but is not a Sigil sealed store; "
                    "refusing to treat a cleartext DB as sealed"
                )
            plain = unseal_bytes(blob, passphrase)
            _write_private_file(self._plain_path, plain)
        else:
            # Touch an empty 0600 file so SQLite does not create a world-readable DB.
            _write_private_file(self._plain_path, b"")
        super().__init__(self._plain_path, anchor_path=anchor_path)
        # Re-pin after SQLite open (WAL/SHM sidecars inherit umask otherwise).
        try:
            os.chmod(self._plain_path, 0o600)
        except OSError:
            pass
    def flush(self) -> str:
        """Checkpoint WAL, read plaintext bytes, write sealed blob to seal_path."""
        with self._write_lock:
            self.conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
            self.conn.commit()
            # Copy via SQLite backup so we don't race readers mid-page.
            dest_path = os.path.join(self._tmpdir.name, "export.db")
            if os.path.exists(dest_path):
                os.unlink(dest_path)
            # Pre-create 0600 so SQLite backup does not leave a world-readable export.
            _write_private_file(dest_path, b"")
            dest = sqlite3.connect(dest_path)
            try:
                self.conn.backup(dest)
                dest.commit()
            finally:
                dest.close()
            try:
                os.chmod(dest_path, 0o600)
            except OSError:
                pass
            with open(dest_path, "rb") as handle:
                plain = handle.read()
            try:
                os.unlink(dest_path)
            except OSError:
                pass
            blob = seal_bytes(plain, self._passphrase)
            directory = os.path.dirname(os.path.abspath(self.seal_path))
            if directory:
                os.makedirs(directory, exist_ok=True)
            fd, tmp = tempfile.mkstemp(
                prefix=".sigil-seal-", suffix=".tmp", dir=directory or "."
            )
            try:
                with os.fdopen(fd, "wb") as handle:
                    handle.write(blob)
                    handle.flush()
                    os.fsync(handle.fileno())
                os.replace(tmp, self.seal_path)
            except Exception:
                try:
                    os.unlink(tmp)
                except OSError:
                    pass
                raise
            os.chmod(self.seal_path, 0o600)
        return self.seal_path
    def close(self) -> None:
        """Flush the seal and release the temp working copy."""
        try:
            self.flush()
        finally:
            try:
                self.conn.close()
            except Exception:
                pass
            self._tmpdir.cleanup()
    def __enter__(self):
        return self
    def __exit__(self, exc_type, exc, tb):
        self.close()
        return False
--- a/sigil/sessions.py
+++ b/sigil/sessions.py
@ -105,16 +105,9 @@ class GrantedCode:
    agent session (issue_code) or a valid code (redeem_code).
    """
-    __slots__ = ("agent", "scopes", "ttl", "fingerprint", "_sigil_internal")
+    __slots__ = ("agent", "scopes", "ttl", "_sigil_internal")
-    def __init__(
+    def __init__(self, agent: str, scopes: List[str], ttl: int, _internal=None):
        self,
        agent: str,
        scopes: List[str],
        ttl: int,
        _internal=None,
        fingerprint: Optional[str] = None,
    ):
        if _internal is not _ISSUER:
            raise errors.SigilError(
                "GrantedCode cannot be constructed directly; it comes from"
@ -123,7 +116,6 @@ class GrantedCode:
        self.agent = agent
        self.scopes = list(scopes)
        self.ttl = ttl
        self.fingerprint = fingerprint
        self._sigil_internal = _internal
    def __repr__(self) -> str:
@ -159,7 +151,6 @@ def issue_view_session(store: Store, grant: "GrantedCode") -> dict:
        granted,
        expires,
        granted_by=agent,
        fingerprint=grant.fingerprint,
    )
    store.record(
        "session.issued",
@ -185,14 +176,7 @@ def issue_view_session(store: Store, grant: "GrantedCode") -> dict:
 def resolve(store: Store, token: str) -> dict:
-    """Look up a session and reject it if revoked, expired, or the agent died.
+    """Look up a session and reject it if revoked or expired."""
    The session row alone is not enough: an in-flight verify/issue_code can
    insert a session or code after revoke/rekey has already run. Re-check the
    agent on every resolve so a race-surviving bearer cannot keep acting.
    """
    from . import identity as _identity
    row = store.get_session(hash_token(token))
    if row is None:
        raise errors.SessionUnknown("no such session")
@ -200,24 +184,6 @@ def resolve(store: Store, token: str) -> dict:
        raise errors.SessionExpired("session revoked")
    if time.time() >= row["expires_at"]:
        raise errors.SessionExpired("session expired")
    agent = row["on_behalf_of"]
    agent_row = store.get_agent(agent)
    if agent_row is None or _identity.is_revoked(agent_row):
        raise errors.SessionExpired("agent revoked")
    if not _identity.has_live_key(agent_row):
        raise errors.SessionExpired("agent has no live key")
    # Every session that carries a fingerprint is bound to that key epoch.
    # After rekey, race-surviving agent:act *and* human_via_agent bearers die
    # even if revoke_sessions missed them (N7).
    sess_fp = row["fingerprint"] or ""
    live_fp = agent_row["fingerprint"] or ""
    if sess_fp or row["kind"] == AGENT:
        # Agent sessions always require a matching fingerprint; human sessions
        # require a match whenever a fingerprint was recorded at mint time.
        if not sess_fp or sess_fp != live_fp:
            raise errors.SessionExpired("session key epoch no longer current")
    return {
        "token": token,
        "kind": row["kind"],
--- a/sigil/store.py
+++ b/sigil/store.py
@ -36,20 +36,6 @@ CREATE TABLE IF NOT EXISTS agents (
    revoked_reason TEXT
 );
 -- Append-only public-key history. Revoke and rekey *retire* a key instead of
 -- destroying it, so historical checkpoints and off-box anchors stay verifiable.
 -- Never stores private keys.
 CREATE TABLE IF NOT EXISTS agent_keys (
    agent          TEXT NOT NULL,
    fingerprint    TEXT NOT NULL,
    pubkey         BLOB NOT NULL,
    active_from    INTEGER NOT NULL,
    retired_at     INTEGER,
    retired_reason TEXT,
    PRIMARY KEY (agent, fingerprint)
 );
 CREATE INDEX IF NOT EXISTS agent_keys_agent ON agent_keys(agent, active_from);
 -- Enrolment bootstrap secrets. Spent secrets are deleted, not marked, so a
 -- spent secret cannot be replayed even if the table is read.
 CREATE TABLE IF NOT EXISTS bootstrap (
@ -91,11 +77,7 @@ CREATE TABLE IF NOT EXISTS codes (
    redeem_expires_at INTEGER NOT NULL,  -- short: the code is a key
    view_ttl         INTEGER NOT NULL,   -- the door opens a full hour, but only
                                        -- once the key is used
-    used_at         INTEGER,
+    used_at         INTEGER
    -- Fingerprint of the agent key at issue time. Redeem refuses if the agent
    -- has since rekeyed (or been wiped), so an in-flight issue_code cannot
    -- outlive a revoke/rekey that lands before put_code commits.
    fingerprint     TEXT
 );
 -- Append-only, hash-chained audit log. The chain is what makes this an
@ -148,9 +130,6 @@ RATE_LIMITS = {
    "challenge": (30, 60),      # 30 per 60s
    "verify": (60, 60),
    "enrol": (10, 300),
    # Separate from enrol: junk pre-auth enrol_rekey must not lock out ops
    # rotating a compromised agent during incident response.
    "rekey": (10, 300),
    "code_issue": (20, 300),
    "code_redeem": (20, 60),
    "code_redeem_fail": (10, 300),
@ -158,15 +137,7 @@ RATE_LIMITS = {
 class Store:
-    def __init__(self, path: str = ":memory:", anchor_path: Optional[str] = None):
+    def __init__(self, path: str = ":memory:"):
        """Open a Sigil store.
        `anchor_path`, when set, is a filesystem path *outside* this database
        where every checkpoint also publishes its signed tip. Rewriting or
        deleting rows in the SQLite file cannot erase that published tip.
        """
        self.path = path
        self.anchor_path = anchor_path
        self.conn = sqlite3.connect(path, check_same_thread=False, timeout=10.0)
        self.conn.row_factory = sqlite3.Row
        # Writers wait rather than failing instantly. Without this a second
@ -238,22 +209,6 @@ class Store:
            self.conn.execute(
                "UPDATE codes SET view_ttl = 3600 WHERE view_ttl IS NULL"
            )
        code_cols = {
            row["name"] for row in self.conn.execute("PRAGMA table_info(codes)")
        }
        if "fingerprint" not in code_cols:
            self.conn.execute("ALTER TABLE codes ADD COLUMN fingerprint TEXT")
        # Backfill agent_keys from current agents rows (live or wiped-empty skipped).
        for row in self.conn.execute(
            "SELECT agent, pubkey, fingerprint, enrolled_at FROM agents"
            " WHERE fingerprint IS NOT NULL AND fingerprint != ''"
            " AND length(pubkey) = 32"
        ):
            self.conn.execute(
                "INSERT OR IGNORE INTO agent_keys"
                " (agent, fingerprint, pubkey, active_from) VALUES (?, ?, ?, ?)",
                (row["agent"], row["fingerprint"], row["pubkey"], row["enrolled_at"]),
            )
        self.conn.commit()
    def close(self) -> None:
@ -290,19 +245,12 @@ class Store:
        which preserves revocation columns.
        """
        with self._write_lock:
            now = int(time.time())
            cur = self.conn.execute(
                "INSERT OR IGNORE INTO agents"
                " (agent, pubkey, fingerprint, enrolled_at)"
                " VALUES (?, ?, ?, ?)",
-                (agent, pubkey, fingerprint, now),
+                (agent, pubkey, fingerprint, int(time.time())),
            )
            if cur.rowcount == 1:
                self.conn.execute(
                    "INSERT OR IGNORE INTO agent_keys"
                    " (agent, fingerprint, pubkey, active_from) VALUES (?, ?, ?, ?)",
                    (agent, fingerprint, pubkey, now),
                )
            self.conn.commit()
            return cur.rowcount == 1
@ -314,33 +262,10 @@ class Store:
        used to un-revoke. An UPDATE of only the key columns keeps the ban intact.
        """
        with self._write_lock:
            now = int(time.time())
            prev = self.conn.execute(
                "SELECT pubkey, fingerprint, enrolled_at FROM agents WHERE agent = ?",
                (agent,),
            ).fetchone()
            cur = self.conn.execute(
                "UPDATE agents SET pubkey = ?, fingerprint = ? WHERE agent = ?",
                (pubkey, fingerprint, agent),
            )
            if cur.rowcount == 1 and prev is not None:
                old_fp = prev["fingerprint"] or ""
                if old_fp and len(prev["pubkey"] or b"") == 32:
                    self.conn.execute(
                        "INSERT OR IGNORE INTO agent_keys"
                        " (agent, fingerprint, pubkey, active_from) VALUES (?, ?, ?, ?)",
                        (agent, old_fp, prev["pubkey"], prev["enrolled_at"] or now),
                    )
                    self.conn.execute(
                        "UPDATE agent_keys SET retired_at = ?, retired_reason = ?"
                        " WHERE agent = ? AND fingerprint = ? AND retired_at IS NULL",
                        (now, "rekey", agent, old_fp),
                    )
                self.conn.execute(
                    "INSERT OR IGNORE INTO agent_keys"
                    " (agent, fingerprint, pubkey, active_from) VALUES (?, ?, ?, ?)",
                    (agent, fingerprint, pubkey, now),
                )
            self.conn.commit()
            return cur.rowcount == 1
@ -351,47 +276,14 @@ class Store:
                     reason: Optional[str] = None) -> int:
        """Suspend an agent. `until` of None is permanent.
        A permanent revoke clears the enrolled public key so `reinstate` alone
        cannot restore a compromised identity — ops must bootstrap + enrol_rekey
        before the agent can authenticate again. Timed suspension keeps the
        key: when the pause lapses, the same key works.
        Returns how many live sessions were killed.
        """
        with self._write_lock:
-            if until is None:
+            self.conn.execute(
-                # Permanent: retire the public key into agent_keys (so historical
+                "UPDATE agents SET revoked_at = ?, revoked_until = ?, revoked_reason = ?"
-                # checkpoints stay verifiable), then wipe the live slot.
+                " WHERE agent = ?",
-                now = int(time.time())
+                (int(time.time()), until, reason, agent),
-                prev = self.conn.execute(
+            )
                    "SELECT pubkey, fingerprint, enrolled_at FROM agents WHERE agent = ?",
                    (agent,),
                ).fetchone()
                if prev is not None:
                    old_fp = prev["fingerprint"] or ""
                    if old_fp and len(prev["pubkey"] or b"") == 32:
                        self.conn.execute(
                            "INSERT OR IGNORE INTO agent_keys"
                            " (agent, fingerprint, pubkey, active_from)"
                            " VALUES (?, ?, ?, ?)",
                            (agent, old_fp, prev["pubkey"], prev["enrolled_at"] or now),
                        )
                        self.conn.execute(
                            "UPDATE agent_keys SET retired_at = ?, retired_reason = ?"
                            " WHERE agent = ? AND fingerprint = ? AND retired_at IS NULL",
                            (now, reason or "permanent revoke", agent, old_fp),
                        )
                self.conn.execute(
                    "UPDATE agents SET revoked_at = ?, revoked_until = ?, revoked_reason = ?,"
                    " pubkey = ?, fingerprint = ? WHERE agent = ?",
                    (now, until, reason, b"", "", agent),
                )
            else:
                self.conn.execute(
                    "UPDATE agents SET revoked_at = ?, revoked_until = ?, revoked_reason = ?"
                    " WHERE agent = ?",
                    (int(time.time()), until, reason, agent),
                )
            # Drop outstanding challenges so an in-flight exchange cannot
            # complete after the revocation lands.
            self.conn.execute(
@ -412,68 +304,31 @@ class Store:
    # -- bootstrap ---------------------------------------------------------
    def put_bootstrap(self, secret_hash: str, agent: str) -> None:
-        """Register one unspent bootstrap secret for an agent.
+        self._exec(
-
+            "INSERT INTO bootstrap (secret_hash, agent, created_at) VALUES (?, ?, ?)",
-        Any prior unspent secrets for the same agent are spent first so only
+            (secret_hash, agent, int(time.time())),
-        one live seed exists at a time (ops double-seed or a chat leak of an
+        )
        unused seed must not leave a standing rekey capability).
        """
        with self._write_lock:
            now = int(time.time())
            self.conn.execute(
                "DELETE FROM bootstrap WHERE agent = ? AND spent_at IS NULL",
                (agent,),
            )
            self.conn.execute(
                "INSERT INTO bootstrap (secret_hash, agent, created_at) VALUES (?, ?, ?)",
                (secret_hash, agent, now),
            )
            self.conn.commit()
    def clear_bootstrap_for_agent(self, agent: str, *, keep_hash: Optional[str] = None) -> int:
        """Delete every unspent bootstrap secret for an agent.
        After a successful enrol / enrol_rekey, leftover seeds must not remain
        usable. keep_hash, when set, is left alone (already consumed by CAS).
        """
        with self._write_lock:
            if keep_hash is None:
                cur = self.conn.execute(
                    "DELETE FROM bootstrap WHERE agent = ? AND spent_at IS NULL",
                    (agent,),
                )
            else:
                cur = self.conn.execute(
                    "DELETE FROM bootstrap WHERE agent = ? AND spent_at IS NULL"
                    " AND secret_hash != ?",
                    (agent, keep_hash),
                )
            self.conn.commit()
            return cur.rowcount
    def spend_bootstrap(self, secret_hash: str) -> Optional[str]:
        """Atomically consume an enrolment secret. Returns the agent or None.
-        Compare-and-set then delete: exactly one concurrent caller can win, and
+        Compare-and-set: the UPDATE only matches an unspent row, so exactly one
-        the spent row is removed so a table dump cannot offline-crack a
+        concurrent caller can win. Check-then-update would let a secret enrol
-        low-entropy custom secret. Matches the schema comment.
+        more than one key.
        """
        with self._write_lock:
            row = self.conn.execute(
                "SELECT agent FROM bootstrap WHERE secret_hash = ? AND spent_at IS NULL",
                (secret_hash,),
            ).fetchone()
            if row is None:
                return None
            agent = row["agent"]
            cur = self.conn.execute(
-                "DELETE FROM bootstrap WHERE secret_hash = ? AND spent_at IS NULL",
+                "UPDATE bootstrap SET spent_at = ? WHERE secret_hash = ?"
-                (secret_hash,),
+                " AND spent_at IS NULL",
                (int(time.time()), secret_hash),
            )
            self.conn.commit()
            if cur.rowcount != 1:
                return None
-            return agent
+            row = self.conn.execute(
                "SELECT agent FROM bootstrap WHERE secret_hash = ?", (secret_hash,)
            ).fetchone()
            return row["agent"] if row else None
    # -- nonces ------------------------------------------------------------
    def put_nonce(self, nonce: str, agent: str, expires_at: int) -> None:
@ -539,30 +394,12 @@ class Store:
            (int(time.time()), token_hash),
        )
    def invalidate_codes_for_agent(self, agent: str) -> int:
        """Mark every unused view code for an agent as spent.
        Revoke, suspend, and rekey must close outstanding codes the same way they
        kill sessions. Leaving a pre-issued code redeemable after compromise
        revoke or key rotation would mint a fresh human_via_agent session while
        the agent is supposed to be dead.
        """
        with self._write_lock:
            cur = self.conn.execute(
                "UPDATE codes SET used_at = ? WHERE agent = ? AND used_at IS NULL",
                (int(time.time()), agent),
            )
            self.conn.commit()
            return cur.rowcount
    def revoke_sessions_for_agent(self, agent: str, reason: str) -> int:
        """Kill every live session for an agent. Used on revoke and rekey.
        Rotating a key must not leave old sessions usable, otherwise the new key
-        buys nothing after a compromise. Outstanding view codes are invalidated
+        buys nothing after a compromise.
        in the same breath so a pre-issued code cannot mint a fresh session.
        """
        self.invalidate_codes_for_agent(agent)
        cur = self._exec(
            "UPDATE sessions SET revoked_at = ?, revoked_reason = ?"
            " WHERE on_behalf_of = ? AND revoked_at IS NULL",
@ -581,11 +418,10 @@ class Store:
        scopes: List[str],
        redeem_expires_at: int,
        view_ttl: int,
        fingerprint: Optional[str] = None,
    ) -> None:
        self._exec(
            "INSERT INTO codes (code_hash, agent, scopes, issued_at, redeem_expires_at,"
-            " view_ttl, fingerprint) VALUES (?, ?, ?, ?, ?, ?, ?)",
+            " view_ttl) VALUES (?, ?, ?, ?, ?, ?)",
            (
                code_hash,
                agent,
@ -593,7 +429,6 @@ class Store:
                int(time.time()),
                redeem_expires_at,
                view_ttl,
                fingerprint,
            ),
        )
@ -601,17 +436,14 @@ class Store:
        return self._one("SELECT * FROM codes WHERE code_hash = ?", (code_hash,))
    def claim_code(self, code_hash: str) -> bool:
-        """Consume a one-time code exactly once. False if already used or expired.
+        """Consume a one-time code exactly once. False if already used.
-        Atomic CAS includes the expiry predicate so a race at the TTL boundary
+        Atomic, so a code cannot mint two sessions under a race.
        cannot mint a session slightly after redeem_expires_at.
        """
        with self._write_lock:
            now = int(time.time())
            cur = self.conn.execute(
-                "UPDATE codes SET used_at = ? WHERE code_hash = ? AND used_at IS NULL"
+                "UPDATE codes SET used_at = ? WHERE code_hash = ? AND used_at IS NULL",
-                " AND redeem_expires_at > ?",
+                (int(time.time()), code_hash),
                (now, code_hash, now),
            )
            self.conn.commit()
            return cur.rowcount == 1
@ -740,26 +572,7 @@ class Store:
                }
            )
        return out
    def get_key_by_fingerprint(self, agent: str, fingerprint: str):
        """Return a historical (or live) public key row for verify.
        Prefers agent_keys history. Falls back to the live agents row when the
        fingerprint still matches (covers DBs created before agent_keys).
        Never returns private key material — pubkey only.
        """
        row = self._one(
            "SELECT * FROM agent_keys WHERE agent = ? AND fingerprint = ?",
            (agent, fingerprint),
        )
        if row is not None:
            return row
        live = self.get_agent(agent)
        if live is not None and (live["fingerprint"] or "") == fingerprint:
            return live
        return None
    # -- checkpoints -------------------------------------------------------
    @staticmethod
    def _checkpoint_payload(at: int, label: str, head: str, events: int,
                            signer: str, fingerprint: str) -> bytes:
@ -778,29 +591,13 @@ class Store:
        digest, so later verifying that the chain still ends at the same
        committed head is a check against anyone who cannot forge a signature
        with this key.
        When this Store was opened with `anchor_path`, the same signed tip is
        also published to that path. The published tip survives a rewrite of
        this database file.
        """
        from . import keys as _keys
        from . import anchor as _anchor
        public = _keys.public_from_private(private)
        fingerprint = _keys.fingerprint(public)
-        from . import identity as _identity
+        if self.get_agent(signer) is None:
        agent_row = self.get_agent(signer)
        if agent_row is None:
            raise ValueError(f"{signer} is not enrolled, so cannot sign a checkpoint")
        if _identity.is_revoked(agent_row):
            raise ValueError(f"{signer} is revoked, so cannot sign a checkpoint")
        enrolled_fp = agent_row["fingerprint"] or ""
        if not enrolled_fp or fingerprint != enrolled_fp:
            raise ValueError(
                f"private key fingerprint does not match enrolled key for {signer}; "
                "refusing to sign a checkpoint with a stranger key"
            )
        with self._write_lock:
            last = self.conn.execute(
                "SELECT id, hash FROM events ORDER BY id DESC LIMIT 1"
@ -817,97 +614,8 @@ class Store:
                (at, label or "", head, events, signer, fingerprint, signature),
            )
            self.conn.commit()
-        result = {"at": at, "label": label, "head": head, "events": events,
+        return {"at": at, "label": label, "head": head, "events": events,
-                  "signer": signer, "fingerprint": fingerprint,
+                "signer": signer, "fingerprint": fingerprint}
                  "signature": signature}
        if self.anchor_path:
            published = _anchor.write_tip(self.anchor_path, result)
            result["anchor_path"] = published
        return result
    def publish_anchor(self, private, signer: str, label: Optional[str] = None) -> dict:
        """Checkpoint and require an off-box publish.
        Same as checkpoint(), but refuses if this Store has no anchor_path, so
        a deployment that intends to keep the tip off-box cannot silently fall
        back to in-DB-only.
        """
        if not self.anchor_path:
            raise ValueError(
                "publish_anchor requires Store(..., anchor_path=...); "
                "a checkpoint inside the database alone is not off-box"
            )
        return self.checkpoint(private, signer, label)
    def verify_external_anchors(self) -> Dict[str, Any]:
        """Verify the published tip (and history) against enrolled keys + chain.
        Returns ok=False when:
        - the tip file is missing but an anchor_path was configured
        - a notice signature fails
        - the tip head is no longer present at the recorded event count
          (the DB was rewritten or truncated past the published tip)
        """
        from . import keys as _keys
        from . import anchor as _anchor
        if not self.anchor_path:
            return {"ok": True, "anchors": 0, "skipped": True,
                    "reason": "no anchor_path configured"}
        notices = _anchor.read_log(self.anchor_path)
        tip = _anchor.read_tip(self.anchor_path)
        if tip is None:
            return {"ok": False, "anchors": 0, "failures": [
                {"reason": f"anchor tip missing at {self.anchor_path}"}
            ]}
        # Ensure the tip itself is represented even if the log write failed.
        if not notices or notices[-1].get("head") != tip.get("head"):
            notices = list(notices) + [tip]
        results = []
        warnings = []
        for notice in notices:
            if self.get_agent(notice["signer"]) is None:
                results.append({"ok": False, "head": notice.get("head"),
                                "reason": f"signer {notice['signer']} no longer enrolled"})
                continue
            key_row = self.get_key_by_fingerprint(
                notice["signer"], notice.get("fingerprint") or ""
            )
            if key_row is None:
                results.append({"ok": False, "head": notice.get("head"),
                                "reason": "signer key for this fingerprint is unknown"})
                continue
            public = _keys.from_public_bytes(key_row["pubkey"])
            checked = _anchor.verify_notice(notice, public)
            if not checked["ok"]:
                results.append({"ok": False, "head": notice.get("head"),
                                "reason": checked["reason"]})
                continue
            found = self.conn.execute(
                "SELECT hash FROM events WHERE id = ?", (notice["events"],)
            ).fetchone()
            actual = GENESIS if notice["events"] == 0 else (
                found["hash"] if found else None
            )
            if actual != notice["head"]:
                results.append({
                    "ok": False,
                    "head": notice["head"],
                    "reason": "published tip head no longer matches the chain "
                              "(database rewrite cannot erase the off-box tip)",
                })
                continue
            entry = {"ok": True, "head": notice["head"],
                     "events": notice["events"], "signer": notice["signer"]}
            retired_at = key_row["retired_at"] if "retired_at" in key_row.keys() else None
            if retired_at is not None and notice.get("at", 0) > retired_at:
                entry["flagged"] = True
                entry["reason"] = "notice timestamp is after the signing key was retired"
                warnings.append(entry)
            results.append(entry)
        bad = [r for r in results if not r["ok"]]
        return {"ok": not bad, "anchors": len(results), "failures": bad,
                "warnings": warnings, "tip": tip, "path": self.anchor_path}
    def checkpoints(self) -> List[Dict[str, Any]]:
        rows = self.conn.execute("SELECT * FROM checkpoints ORDER BY id ASC").fetchall()
@ -923,23 +631,22 @@ class Store:
        from . import keys as _keys
        results = []
        warnings = []
        for row in self.conn.execute("SELECT * FROM checkpoints ORDER BY id ASC"):
-            if self.get_agent(row["signer"]) is None:
+            agent_row = self.get_agent(row["signer"])
            if agent_row is None:
                results.append({"id": row["id"], "ok": False,
                                "reason": f"signer {row['signer']} no longer enrolled"})
                continue
-            key_row = self.get_key_by_fingerprint(row["signer"], row["fingerprint"])
+            if agent_row["fingerprint"] != row["fingerprint"]:
            if key_row is None:
                results.append({"id": row["id"], "ok": False,
-                                "reason": "signer key for this fingerprint is unknown"})
+                                "reason": "signer key changed since the checkpoint was made"})
                continue
            payload = self._checkpoint_payload(
                row["at"], row["label"], row["head"], row["events"],
                row["signer"], row["fingerprint"],
            )
            try:
-                public = _keys.from_public_bytes(key_row["pubkey"])
+                public = _keys.from_public_bytes(agent_row["pubkey"])
                # verify() returns None on success and raises on failure, so it
                # must be called, not inspected.
                public.verify(base64.b64decode(row["signature"]), payload)
@ -957,16 +664,9 @@ class Store:
                results.append({"id": row["id"], "ok": False,
                                "reason": "chain head at that position no longer matches"})
                continue
-            entry = {"id": row["id"], "ok": True}
+            results.append({"id": row["id"], "ok": True})
            retired_at = key_row["retired_at"] if "retired_at" in key_row.keys() else None
            if retired_at is not None and row["at"] > retired_at:
                entry["flagged"] = True
                entry["reason"] = "checkpoint timestamp is after the signing key was retired"
                warnings.append(entry)
            results.append(entry)
        bad = [r for r in results if not r["ok"]]
-        return {"ok": not bad, "checkpoints": len(results),
+        return {"ok": not bad, "checkpoints": len(results), "failures": bad}
                "failures": bad, "warnings": warnings}
    # -- housekeeping ------------------------------------------------------
    def sweep(self) -> Dict[str, int]:
--- a/sigil/viewcodes.py
+++ b/sigil/viewcodes.py
@ -91,18 +91,9 @@ def issue_code(
    if not store.allow(f"code_issue:{agent}", limit=limit, window=window):
        raise errors.RateLimited(f"too many view codes for {agent}")
    from . import identity as _identity
    agent_row = store.get_agent(agent)
    if agent_row is None:
        raise errors.UnknownAgent(f"{agent} is not enrolled")
    # Re-check after resolve: revoke/rekey can land between resolve and put_code.
    if _identity.is_revoked(agent_row) or not _identity.has_live_key(agent_row):
        raise errors.UnknownAgent(f"{agent} is revoked or has no live key")
    live_fp = agent_row["fingerprint"] or ""
    sess_fp = principal.get("fingerprint") or ""
    if not live_fp or sess_fp != live_fp:
        raise errors.SessionExpired("session key epoch no longer current")
    granted = list(scopes if scopes is not None else sessions.HUMAN_VIEW_SCOPES)
    # Refuse to bake write or act scopes into a human code at issue time.
@ -124,13 +115,12 @@ def issue_code(
        granted,
        redeem_expires_at=now + code_ttl,
        view_ttl=view_ttl,
        fingerprint=live_fp,
    )
    store.record(
        "view_code.issued",
        actor=agent,
        subject=f"{agent}'s human",
-        fingerprint=live_fp,
+        fingerprint=agent_row["fingerprint"],
        detail={
            "code_ref": _code_ref(code),
            "scopes": granted,
@ -144,6 +134,7 @@ def issue_code(
        "code_ttl": code_ttl,
        "view_ttl": view_ttl,
        "redeem_expires_at": now + code_ttl,
        "view_ttl": view_ttl,
    }
@ -164,28 +155,6 @@ def redeem_code(store: Store, code: str) -> dict:
    agent = row["agent"]
    # Defense in depth: refuse if agent revoked, wiped, or rekeyed since issue.
    from . import identity as _identity
    agent_row = store.get_agent(agent)
    code_fp = row["fingerprint"] if "fingerprint" in row.keys() else None
    live_fp = (agent_row["fingerprint"] or "") if agent_row is not None else ""
    bad = (
        agent_row is None
        or _identity.is_revoked(agent_row)
        or not _identity.has_live_key(agent_row)
        or (code_fp is not None and code_fp != live_fp)
    )
    if bad:
        if not store.allow(f"code_redeem_fail:{agent}"):
            raise errors.RateLimited("too many redemption attempts")
        store.record(
            "view_code.redeem_refused",
            actor=agent,
            subject=f"{agent}'s human",
            detail={"reason": "agent revoked, wiped, or rekeyed"},
        )
        raise errors.CodeInvalid("that code is not redeemable")
    # Known-agent attempts use the success quota (code_redeem), not the fail
    # bucket. Failures for this agent (expired / already used) charge a
    # per-agent fail bucket so one noisy human cannot lock every agent out.
@ -231,7 +200,6 @@ def redeem_code(store: Store, code: str) -> dict:
        _json.loads(row["scopes"]),
        row["view_ttl"],
        sessions._ISSUER,
        fingerprint=live_fp,
    )
    return sessions.issue_view_session(store, grant)
--- a/tests/test_anchor.py
+++ b/tests/test_anchor.py
@ -1,137 +0,0 @@
 """Off-box audit chain anchor.
 Finding: a signed checkpoint inside the same SQLite file is still only as
 strong as that file. An attacker who can rewrite the database can delete the
 checkpoints table and recompute the chain. Publishing the signed tip to a path
 outside the database means rewriting the DB alone cannot erase the tip.
 """
 from __future__ import annotations
 import json
 import os
 import tempfile
 import unittest
 from sigil import identity, keys
 from sigil.store import Store, GENESIS
 from sigil import anchor as anchor_mod
 from tests import helpers
 class TestOffBoxAnchor(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.db = os.path.join(self.tmp.name, "sigil.db")
        self.tip = os.path.join(self.tmp.name, "offbox", "chain.tip")
        self.store = Store(self.db, anchor_path=self.tip)
        self.private, _ = helpers.enrol(self.store, "bigpickle")
        identity.challenge(self.store, "bigpickle")  # leaves an event
    def test_checkpoint_publishes_signed_tip_outside_the_db(self):
        result = self.store.checkpoint(self.private, "bigpickle", "daily")
        self.assertEqual(result["anchor_path"], self.tip)
        self.assertTrue(os.path.exists(self.tip))
        # Tip must not live inside the SQLite file.
        with open(self.db, "rb") as handle:
            raw = handle.read()
        tip_text = open(self.tip, "r", encoding="utf-8").read()
        notice = json.loads(tip_text)
        self.assertNotIn(notice["signature"].encode(), raw)
        self.assertEqual(notice["head"], result["head"])
        self.assertEqual(notice["signer"], "bigpickle")
        self.assertTrue(self.store.verify_external_anchors()["ok"])
    def test_publish_anchor_refuses_without_path(self):
        bare, private = helpers.fresh()
        with self.assertRaises(ValueError):
            bare.publish_anchor(private, "bigpickle")
    def test_deleting_in_db_checkpoints_does_not_erase_the_tip(self):
        self.store.checkpoint(self.private, "bigpickle", "kept-offbox")
        self.store.conn.execute("DELETE FROM checkpoints")
        self.store.conn.commit()
        # In-DB check has nothing left to verify.
        self.assertTrue(self.store.verify_checkpoints()["ok"])
        self.assertEqual(self.store.verify_checkpoints()["checkpoints"], 0)
        # Off-box tip still present and valid against the live chain.
        external = self.store.verify_external_anchors()
        self.assertTrue(external["ok"], external)
        self.assertGreaterEqual(external["anchors"], 1)
        self.assertTrue(os.path.exists(self.tip))
    def test_rewriting_the_db_is_caught_by_the_offbox_tip(self):
        self.store.checkpoint(self.private, "bigpickle", "baseline")
        tip_before = json.loads(open(self.tip).read())
        # Attacker deletes history, forges an event, rebuilds hashes, and
        # clears the in-DB checkpoints so verify_checkpoints is vacuously ok.
        self.store.conn.execute("DELETE FROM events")
        self.store.conn.execute("DELETE FROM checkpoints")
        self.store.conn.commit()
        self.store.record("forged", "attacker", subject="bigpickle",
                          detail={"note": "never happened"})
        rows = self.store.conn.execute("SELECT * FROM events ORDER BY id").fetchall()
        prev = GENESIS
        for row in rows:
            digest = Store._event_hash(
                prev, row["at"], row["kind"], row["actor"],
                row["subject"], row["fingerprint"], row["detail"],
            )
            self.store.conn.execute(
                "UPDATE events SET prev_hash = ?, hash = ? WHERE id = ?",
                (prev, digest, row["id"]),
            )
            prev = digest
        self.store.conn.commit()
        self.assertTrue(self.store.verify_chain()["ok"])
        self.assertTrue(self.store.verify_checkpoints()["ok"])
        # Tip file still holds the pre-attack head.
        tip_after = json.loads(open(self.tip).read())
        self.assertEqual(tip_after["head"], tip_before["head"])
        external = self.store.verify_external_anchors()
        self.assertFalse(external["ok"], external)
        self.assertIn("off-box tip", external["failures"][0]["reason"])
    def test_forged_offbox_signature_is_rejected(self):
        self.store.checkpoint(self.private, "bigpickle", "honest")
        notice = json.loads(open(self.tip).read())
        notice["signature"] = __import__("base64").b64encode(b"\x00" * 64).decode()
        open(self.tip, "w").write(json.dumps(notice))
        # Truncate the log so verify only sees the forged tip.
        log = self.tip + ".log"
        if os.path.exists(log):
            os.unlink(log)
        external = self.store.verify_external_anchors()
        self.assertFalse(external["ok"])
        self.assertIn("signature", external["failures"][0]["reason"])
    def test_missing_configured_tip_fails_closed(self):
        # Configure a path that was never written.
        lonely = Store(self.db, anchor_path=os.path.join(self.tmp.name, "missing.tip"))
        external = lonely.verify_external_anchors()
        self.assertFalse(external["ok"])
        self.assertIn("missing", external["failures"][0]["reason"])
    def test_no_anchor_path_skips_cleanly(self):
        store, private = helpers.fresh()
        store.checkpoint(private, "bigpickle", "in-db-only")
        external = store.verify_external_anchors()
        self.assertTrue(external["ok"])
        self.assertTrue(external.get("skipped"))
    def test_anchor_log_keeps_prior_tips(self):
        first = self.store.checkpoint(self.private, "bigpickle", "one")
        identity.challenge(self.store, "bigpickle")
        second = self.store.checkpoint(self.private, "bigpickle", "two")
        self.assertNotEqual(first["head"], second["head"])
        history = anchor_mod.read_log(self.tip)
        self.assertGreaterEqual(len(history), 2)
        self.assertEqual(history[-1]["head"], second["head"])
        tip = anchor_mod.read_tip(self.tip)
        self.assertEqual(tip["head"], second["head"])
        self.assertTrue(self.store.verify_external_anchors()["ok"])
 if __name__ == "__main__":
    unittest.main()
--- a/tests/test_hardening.py
+++ b/tests/test_hardening.py
@ -359,216 +359,4 @@ class TestAgentIdValidation(unittest.TestCase):
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()
 class TestClaimCodeExpiryCAS(unittest.TestCase):
    """R6: claim_code CAS must include redeem_expires_at predicate."""
    def test_claim_code_after_expiry_returns_false(self):
        store, private = helpers.fresh()
        issued = helpers.issue(store, private, code_ttl=1, view_ttl=60)
        digest = viewcodes._hash_code(issued["code"])
        # Force expiry in the row.
        store.conn.execute(
            "UPDATE codes SET redeem_expires_at = ? WHERE code_hash = ?",
            (int(time.time()) - 1, digest),
        )
        store.conn.commit()
        self.assertFalse(store.claim_code(digest))
        row = store.get_code(digest)
        self.assertIsNone(row["used_at"])
 class TestEnrolRekeyRateLimit(unittest.TestCase):
    """N3: enrol_rekey uses a dedicated rekey: bucket; junk cannot lock ops."""
    def test_exhausting_enrol_bucket_does_not_block_rekey(self):
        store, _ = helpers.fresh("limited")
        for _ in range(20):
            store.allow("enrol:limited")
        secret = identity.seed_bootstrap(store, "limited")
        # Valid rekey must succeed even if enrol: is exhausted.
        identity.enrol_rekey(
            store, "limited", keys.export_public_b64(keys.generate()[1]), secret
        )
    def test_junk_rekey_fills_rekey_bucket_not_enrol(self):
        store, _ = helpers.fresh("victim")
        for _ in range(15):
            try:
                identity.enrol_rekey(
                    store, "victim", keys.export_public_b64(keys.generate()[1]), "junk"
                )
            except (errors.EnrolmentFailed, errors.RateLimited):
                pass
        # enrol: bucket still usable
        self.assertTrue(store.allow("enrol:victim"))
        # Valid ops rekey still works: failures charge rekey: only after bad secret,
        # and a valid secret bypasses the limiter.
        secret = identity.seed_bootstrap(store, "victim")
        identity.enrol_rekey(
            store, "victim", keys.export_public_b64(keys.generate()[1]), secret
        )
    def test_failed_rekey_eventually_rate_limited(self):
        store, _ = helpers.fresh("hammer")
        hit = False
        for _ in range(30):
            try:
                identity.enrol_rekey(
                    store, "hammer", keys.export_public_b64(keys.generate()[1]), "nope"
                )
            except errors.RateLimited:
                hit = True
                break
            except errors.EnrolmentFailed:
                pass
        self.assertTrue(hit, "junk rekey attempts should eventually RateLimited")
 class TestBootstrapRowDeletedOnSpend(unittest.TestCase):
    """R8: spent bootstrap secrets are deleted, not retained as hashes."""
    def test_spend_deletes_the_row(self):
        store = Store(":memory:")
        secret = identity.seed_bootstrap(store, "dave")
        digest = __import__("hashlib").sha256(secret.encode()).hexdigest()
        self.assertIsNotNone(store.conn.execute(
            "SELECT 1 FROM bootstrap WHERE secret_hash = ?", (digest,)
        ).fetchone())
        owner = store.spend_bootstrap(digest)
        self.assertEqual(owner, "dave")
        self.assertIsNone(store.conn.execute(
            "SELECT 1 FROM bootstrap WHERE secret_hash = ?", (digest,)
        ).fetchone())
 class TestResolveRechecksAgentState(unittest.TestCase):
    """N1: resolve must refuse sessions that outran revoke/rekey."""
    def _plant_agent_session(self, store, agent, fingerprint):
        """Insert a live session as if put_session won a race against revoke."""
        import time as _t
        token = "race-survivor-token"
        store.put_session(
            sessions.hash_token(token),
            sessions.AGENT,
            agent,
            agent,
            list(sessions.AGENT_SCOPES),
            int(_t.time()) + 900,
            fingerprint=fingerprint,
        )
        return token
    def test_resolve_refuses_race_survivor_after_permanent_revoke(self):
        store, private = helpers.fresh()
        fp = store.get_agent("bigpickle")["fingerprint"]
        identity.revoke(store, "bigpickle", reason="compromise")
        # Session inserted *after* revoke (race); not marked revoked_at.
        token = self._plant_agent_session(store, "bigpickle", fp)
        with self.assertRaises(errors.SessionExpired):
            sessions.resolve(store, token)
    def test_resolve_refuses_stale_fingerprint_after_rekey(self):
        store, private = helpers.fresh()
        old_fp = store.get_agent("bigpickle")["fingerprint"]
        new_pub = keys.generate()[1]
        secret = identity.seed_bootstrap(store, "bigpickle")
        identity.enrol_rekey(
            store, "bigpickle", keys.export_public_b64(new_pub), secret
        )
        token = self._plant_agent_session(store, "bigpickle", old_fp)
        with self.assertRaises(errors.SessionExpired):
            sessions.resolve(store, token)
    def test_redeem_refuses_code_after_rekey_even_if_not_invalidated(self):
        """Fingerprint binding: plant a code with a stale fingerprint."""
        store, private = helpers.fresh()
        agent_row = store.get_agent("bigpickle")
        old_fp = agent_row["fingerprint"]
        # Issue normally, then surgically restore used_at and keep old fp after rekey
        issued = helpers.issue(store, private)
        digest = viewcodes._hash_code(issued["code"])
        new_pub = keys.generate()[1]
        secret = identity.seed_bootstrap(store, "bigpickle")
        identity.enrol_rekey(
            store, "bigpickle", keys.export_public_b64(new_pub), secret
        )
        # Rekey invalidated the code; un-burn it to simulate put_code after invalidate.
        store.conn.execute(
            "UPDATE codes SET used_at = NULL, fingerprint = ? WHERE code_hash = ?",
            (old_fp, digest),
        )
        store.conn.commit()
        with self.assertRaises(errors.CodeInvalid):
            viewcodes.redeem_code(store, issued["code"])
    def test_redeem_refuses_without_live_key(self):
        store, private = helpers.fresh()
        issued = helpers.issue(store, private)
        digest = viewcodes._hash_code(issued["code"])
        identity.revoke(store, "bigpickle", reason="compromise")
        identity.reinstate(store, "bigpickle")  # ban lifted, key still wiped
        store.conn.execute(
            "UPDATE codes SET used_at = NULL WHERE code_hash = ?", (digest,)
        )
        store.conn.commit()
        with self.assertRaises(errors.CodeInvalid):
            viewcodes.redeem_code(store, issued["code"])
    def test_verify_then_revoke_session_unusable(self):
        """Session minted just before revoke must not resolve."""
        store, private = helpers.fresh()
        chal = identity.challenge(store, "bigpickle")
        sig = helpers.sign(private, chal["nonce"])
        sess = identity.verify(store, chal["nonce"], sig)
        identity.revoke(store, "bigpickle", reason="compromise")
        with self.assertRaises(errors.SessionExpired):
            sessions.resolve(store, sess["token"])
 class TestHumanSessionFingerprintEpoch(unittest.TestCase):
    """N7: rekey must kill race-surviving human_via_agent sessions too."""
    def test_resolve_refuses_stale_human_fingerprint_after_rekey(self):
        store, private = helpers.fresh()
        old_fp = store.get_agent("bigpickle")["fingerprint"]
        # Plant a human session as if redeem's put_session won a race against rekey.
        import time as _t
        token = "human-race-survivor"
        store.put_session(
            sessions.hash_token(token),
            sessions.HUMAN_VIA_AGENT,
            "bigpickle's human",
            "bigpickle",
            list(sessions.HUMAN_VIEW_SCOPES),
            int(_t.time()) + 3600,
            granted_by="bigpickle",
            fingerprint=old_fp,
        )
        new_pub = keys.generate()[1]
        secret = identity.seed_bootstrap(store, "bigpickle")
        identity.enrol_rekey(
            store, "bigpickle", keys.export_public_b64(new_pub), secret
        )
        # Clear revoked_at that rekey set, to isolate the fingerprint check.
        store.conn.execute(
            "UPDATE sessions SET revoked_at = NULL, revoked_reason = NULL"
            " WHERE token_hash = ?",
            (sessions.hash_token(token),),
        )
        store.conn.commit()
        with self.assertRaises(errors.SessionExpired) as ctx:
            sessions.resolve(store, token)
        self.assertIn("key epoch", str(ctx.exception))
    def test_issue_code_return_has_single_view_ttl(self):
        """N8: issue_code return dict must not duplicate view_ttl."""
        import inspect
        from sigil import viewcodes as vc
        src = inspect.getsource(vc.issue_code)
        # Count literal "view_ttl" keys in the return dict construction region.
        self.assertEqual(src.count('"view_ttl":'), 1, src[src.rfind("return"):])
--- a/tests/test_identity.py
+++ b/tests/test_identity.py
@ -6,8 +6,6 @@ import unittest
 from sigil import errors, identity, keys
 from sigil.store import Store
 import helpers
 class Base(unittest.TestCase):
    def setUp(self):
@ -146,59 +144,5 @@ class TestRekey(Base):
        )
 class TestBootstrapHygiene(unittest.TestCase):
    """R2: only one live bootstrap secret; leftovers unusable after enrol."""
    def test_leftover_seed_fails_after_enrol(self):
        store = Store(":memory:")
        private, public = keys.generate()
        s1 = identity.seed_bootstrap(store, "alice")
        # Direct insert of a second hash simulates a pre-hygiene double-seed
        # (put_bootstrap now clears siblings; plant one after s1 is recorded).
        import hashlib
        s2 = "leftover-seed-that-must-die"
        store.conn.execute(
            "INSERT INTO bootstrap (secret_hash, agent, created_at) VALUES (?, ?, ?)",
            (hashlib.sha256(s2.encode()).hexdigest(), "alice", 1),
        )
        store.conn.commit()
        identity.enrol(store, "alice", keys.export_public_b64(public), s1)
        with self.assertRaises(errors.EnrolmentFailed):
            identity.enrol_rekey(
                store, "alice", keys.export_public_b64(public), s2
            )
    def test_put_bootstrap_keeps_only_one_live_secret(self):
        store = Store(":memory:")
        s1 = identity.seed_bootstrap(store, "bob")
        s2 = identity.seed_bootstrap(store, "bob")
        private, public = keys.generate()
        with self.assertRaises(errors.EnrolmentFailed):
            identity.enrol(store, "bob", keys.export_public_b64(public), s1)
        # Only the latest seed works.
        identity.enrol(store, "bob", keys.export_public_b64(public), s2)
    def test_leftover_seed_fails_after_rekey(self):
        store = Store(":memory:")
        helpers.enrol(store, "carol")
        import hashlib
        leftover = "another-unused-seed"
        new_pub = keys.generate()[1]
        secret = identity.seed_bootstrap(store, "carol")
        # Plant a sibling after the live seed (simulates a concurrent leak).
        store.conn.execute(
            "INSERT INTO bootstrap (secret_hash, agent, created_at) VALUES (?, ?, ?)",
            (hashlib.sha256(leftover.encode()).hexdigest(), "carol", 2),
        )
        store.conn.commit()
        identity.enrol_rekey(
            store, "carol", keys.export_public_b64(new_pub), secret
        )
        newer_pub = keys.generate()[1]
        with self.assertRaises(errors.EnrolmentFailed):
            identity.enrol_rekey(
                store, "carol", keys.export_public_b64(newer_pub), leftover
            )
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()
--- a/tests/test_production.py
+++ b/tests/test_production.py
@ -133,25 +133,15 @@ class TestSignedCheckpoints(unittest.TestCase):
        with self.assertRaises(ValueError):
            stranger.checkpoint(self.private, "not-enrolled")
-    def test_checkpoint_refuses_enrolled_name_with_wrong_key(self):
+    def test_checkpoint_after_rekey_is_flagged(self):
        # R4: signer name is enrolled, but private key is not the enrolled one.
        before = len(self.store.checkpoints())
        other_private, _ = keys.generate()
        with self.assertRaises(ValueError) as ctx:
            self.store.checkpoint(other_private, "bigpickle")
        self.assertIn("does not match enrolled key", str(ctx.exception))
        self.assertEqual(len(self.store.checkpoints()), before)
    def test_checkpoint_after_rekey_still_verifies_via_history(self):
        # N2: historical checkpoints verify against retired public keys.
        secret = identity.seed_bootstrap(self.store, "bigpickle")
        identity.enrol_rekey(
            self.store, "bigpickle",
            keys.export_public_b64(keys.generate()[1]), secret,
        )
        result = self.store.verify_checkpoints()
-        self.assertTrue(result["ok"], result)
+        self.assertFalse(result["ok"])
-        self.assertEqual(result["failures"], [])
+        self.assertIn("key changed", result["failures"][0]["reason"])
    def test_empty_log_checkpoints_cleanly(self):
        store, private = helpers.fresh()
@ -203,19 +193,10 @@ class TestRevocationSemantics(unittest.TestCase):
    def test_reinstating_works_and_is_audited(self):
        identity.revoke(self.store, "bigpickle")
        identity.reinstate(self.store, "bigpickle")
        # Permanent revoke cleared the key; reinstate alone is not enough.
        with self.assertRaises(errors.UnknownAgent):
            identity.challenge(self.store, "bigpickle")
        new_pub = keys.generate()[1]
        secret = identity.seed_bootstrap(self.store, "bigpickle")
        identity.enrol_rekey(
            self.store, "bigpickle", keys.export_public_b64(new_pub), secret
        )
        identity.challenge(self.store, "bigpickle")
        kinds = [e["kind"] for e in self.store.events(limit=50)]
        self.assertIn("identity.revoked", kinds)
        self.assertIn("identity.reinstated", kinds)
        self.assertIn("identity.rekeyed", kinds)
    def test_revoking_an_unknown_agent_refused(self):
        with self.assertRaises(errors.UnknownAgent):
@ -249,33 +230,6 @@ class TestRevocationSemantics(unittest.TestCase):
                            helpers.sign(self.private, chal["nonce"]))
 class TestCompromiseReinstateRequiresRekey(unittest.TestCase):
    """R3: reinstate after permanent revoke must not restore the same key."""
    def setUp(self):
        self.store, self.private = helpers.fresh()
    def test_reinstate_alone_cannot_use_old_key(self):
        old_token_path_chal = identity.challenge(self.store, "bigpickle")
        self.assertTrue(old_token_path_chal["fingerprint"])
        identity.revoke(self.store, "bigpickle", reason="key leak")
        identity.reinstate(self.store, "bigpickle")
        with self.assertRaises(errors.UnknownAgent):
            identity.challenge(self.store, "bigpickle")
        # Old private key must not verify even if a nonce were planted.
        self.assertFalse(identity.has_live_key(self.store.get_agent("bigpickle")))
    def test_timed_suspend_reinstate_keeps_same_key(self):
        import time
        fp = self.store.get_agent("bigpickle")["fingerprint"]
        identity.revoke(
            self.store, "bigpickle", until=int(time.time()) + 60, reason="pause"
        )
        identity.reinstate(self.store, "bigpickle")
        chal = identity.challenge(self.store, "bigpickle")
        self.assertEqual(chal["fingerprint"], fp)
 class TestViewSessionsCannotBeForged(unittest.TestCase):
    """Finding: human sessions were minted from an agent name.
@ -346,14 +300,12 @@ class TestRekeyDoesNotUnrevoke(unittest.TestCase):
    def setUp(self):
        self.store, self.private = helpers.fresh()
-    def test_rekey_refuses_while_timed_suspended(self):
+    def test_rekey_refuses_while_revoked(self):
-        import time
+        identity.revoke(self.store, "bigpickle", reason="incident")
-        # Timed suspend keeps the key — rekey must not be a back door.
+        secret = None
        identity.revoke(
            self.store, "bigpickle", until=int(time.time()) + 3600, reason="pause"
        )
        with self.assertRaises(errors.UnknownAgent):
-            identity.seed_bootstrap(self.store, "bigpickle")
+            secret = identity.seed_bootstrap(self.store, "bigpickle")
        # Even with a secretly planted bootstrap, rekey must refuse.
        planted = secrets_module.token_urlsafe(24)
        self.store.put_bootstrap(
            __import__("hashlib").sha256(planted.encode()).hexdigest(),
@ -368,28 +320,6 @@ class TestRekeyDoesNotUnrevoke(unittest.TestCase):
            )
        row = self.store.get_agent("bigpickle")
        self.assertTrue(identity.is_revoked(row))
        self.assertTrue(identity.has_live_key(row))
    def test_permanent_revoke_allows_rekey_but_stays_revoked(self):
        """Compromise recovery: rotate key while banned, then reinstate."""
        old_fp = self.store.get_agent("bigpickle")["fingerprint"]
        identity.revoke(self.store, "bigpickle", reason="compromise")
        row = self.store.get_agent("bigpickle")
        self.assertTrue(identity.is_revoked(row))
        self.assertFalse(identity.has_live_key(row))
        new_pub = keys.generate()[1]
        secret = identity.seed_bootstrap(self.store, "bigpickle")
        new_fp = identity.enrol_rekey(
            self.store, "bigpickle", keys.export_public_b64(new_pub), secret
        )
        row = self.store.get_agent("bigpickle")
        self.assertTrue(identity.is_revoked(row))
        self.assertTrue(identity.has_live_key(row))
        self.assertNotEqual(new_fp, old_fp)
        with self.assertRaises(errors.UnknownAgent):
            identity.challenge(self.store, "bigpickle")
        identity.reinstate(self.store, "bigpickle")
        identity.challenge(self.store, "bigpickle")
    def test_rekey_preserves_revocation_columns(self):
        # Reinstate path: if somehow update ran under a revoked row, columns stay.
@ -415,17 +345,10 @@ class TestDualBootstrapEnrolRace(unittest.TestCase):
        import threading
        store = Store(":memory:")
-        # Plant two live secrets directly: put_bootstrap now keeps only one
+        secrets = [
-        # live seed, so the dual-enrol race must bypass that hygiene.
+            identity.seed_bootstrap(store, "racer"),
-        import hashlib, time as _time
+            identity.seed_bootstrap(store, "racer"),
-        secrets = ["race-secret-alpha", "race-secret-beta"]
+        ]
        now = int(_time.time())
        for s in secrets:
            store.conn.execute(
                "INSERT INTO bootstrap (secret_hash, agent, created_at) VALUES (?, ?, ?)",
                (hashlib.sha256(s.encode()).hexdigest(), "racer", now),
            )
        store.conn.commit()
        pubs = [keys.generate()[1] for _ in range(2)]
        results = []
        lock = threading.Lock()
@ -492,75 +415,3 @@ class TestRedeemRateLimitIsolation(unittest.TestCase):
            self.fail("successes must not share the fail bucket")
        except errors.SigilError:
            pass
 class TestKeyHistoryKeepsCheckpointsVerifiable(unittest.TestCase):
    """N2: permanent revoke / rekey must not make honest checkpoints unverifiable."""
    def test_checkpoints_verify_after_signer_permanent_revoke(self):
        import tempfile, os
        d = tempfile.mkdtemp()
        anchor = os.path.join(d, "tip")
        store = Store(":memory:", anchor_path=anchor)
        private, _ = helpers.enrol(store, "signer")
        store.record("identity.enrolled", actor="signer", subject="signer")
        store.checkpoint(private, "signer", "pre-compromise")
        before = store.verify_checkpoints()
        self.assertTrue(before["ok"], before)
        anchors_before = store.verify_external_anchors()
        self.assertTrue(anchors_before["ok"], anchors_before)
        identity.revoke(store, "signer", reason="compromise")
        after = store.verify_checkpoints()
        self.assertTrue(after["ok"], after)
        self.assertEqual(after["failures"], [])
        anchors_after = store.verify_external_anchors()
        self.assertTrue(anchors_after["ok"], anchors_after)
        # Live key wiped; history retained.
        self.assertFalse(identity.has_live_key(store.get_agent("signer")))
        hist = store.conn.execute(
            "SELECT fingerprint, retired_at FROM agent_keys WHERE agent = ?",
            ("signer",),
        ).fetchall()
        self.assertEqual(len(hist), 1)
        self.assertIsNotNone(hist[0]["retired_at"])
    def test_private_key_never_in_agent_keys(self):
        store, private = helpers.fresh("carol")
        cols = [r["name"] for r in store.conn.execute("PRAGMA table_info(agent_keys)")]
        self.assertNotIn("private", cols)
        self.assertNotIn("privkey", cols)
        row = store.conn.execute("SELECT * FROM agent_keys WHERE agent = ?", ("carol",)).fetchone()
        self.assertEqual(len(row["pubkey"]), 32)
 class TestEnrolAfterPermanentRevoke(unittest.TestCase):
    """N4: enrol must not burn the seed when the agent row still exists."""
    def test_enrol_after_permanent_revoke_does_not_burn_seed(self):
        store, private = helpers.fresh()
        identity.revoke(store, "bigpickle", reason="compromise")
        identity.reinstate(store, "bigpickle")
        secret = identity.seed_bootstrap(store, "bigpickle")
        with self.assertRaises(errors.EnrolmentFailed) as ctx:
            identity.enrol(
                store, "bigpickle", keys.export_public_b64(keys.generate()[1]), secret
            )
        self.assertIn("enrol_rekey", str(ctx.exception))
        # Seed still usable for the correct path.
        identity.enrol_rekey(
            store, "bigpickle", keys.export_public_b64(keys.generate()[1]), secret
        )
 class TestSuspendedSignerCannotCheckpoint(unittest.TestCase):
    """N5: timed-suspended signer must not checkpoint."""
    def test_timed_suspend_refuses_checkpoint(self):
        import time
        store, private = helpers.fresh()
        identity.revoke(
            store, "bigpickle", until=int(time.time()) + 3600, reason="pause"
        )
        with self.assertRaises(ValueError) as ctx:
            store.checkpoint(private, "bigpickle")
        self.assertIn("revoked", str(ctx.exception).lower())
--- a/tests/test_sealed.py
+++ b/tests/test_sealed.py
@ -1,153 +0,0 @@
 """Sealed store: encryption at rest for the Sigil database file.
 Finding: digests stop a stolen DB from yielding live credentials, but the file
 still holds the agent roster, revocation state and audit history in the clear.
 SealedStore wraps on-disk bytes in AES-256-GCM so that material is unreadable
 without the passphrase. Short of SQLCipher (no page-level encryption while
 open); real working seal of the sensitive file.
 """
 from __future__ import annotations
 import os
 import sqlite3
 import tempfile
 import unittest
 from sigil import identity, keys
 from sigil.sealed import (
    MAGIC,
    SealedStore,
    is_sealed_file,
    seal_bytes,
    unseal_bytes,
 )
 from sigil.store import Store
 from tests import helpers
 PASS = "test-seal-passphrase-ok"
 class TestSealBytes(unittest.TestCase):
    def test_round_trip(self):
        blob = seal_bytes(b"SQLite format 3\x00hello", PASS)
        self.assertTrue(blob.startswith(MAGIC))
        self.assertEqual(unseal_bytes(blob, PASS), b"SQLite format 3\x00hello")
    def test_wrong_passphrase_refused(self):
        blob = seal_bytes(b"secret-db-bytes", PASS)
        with self.assertRaises(ValueError):
            unseal_bytes(blob, "wrong-passphrase!!")
    def test_short_passphrase_refused(self):
        with self.assertRaises(ValueError):
            seal_bytes(b"x", "short")
    def test_corrupt_magic_refused(self):
        with self.assertRaises(ValueError):
            unseal_bytes(b"NOTSEALED" + b"\x00" * 40, PASS)
 class TestSealedStore(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.path = os.path.join(self.tmp.name, "sigil.seal")
    def test_flush_writes_ciphertext_without_sqlite_header(self):
        store = SealedStore(self.path, PASS)
        helpers.enrol(store, "bigpickle")
        store.flush()
        with open(self.path, "rb") as handle:
            raw = handle.read()
        self.assertTrue(raw.startswith(MAGIC))
        self.assertNotIn(b"SQLite format 3", raw)
        self.assertTrue(is_sealed_file(self.path))
        # Agent id must not appear in the clear on disk.
        self.assertNotIn(b"bigpickle", raw)
        store.close()
    def test_reopen_with_passphrase_recovers_agents(self):
        with SealedStore(self.path, PASS) as store:
            private, _ = helpers.enrol(store, "bigpickle")
            store.flush()
        with SealedStore(self.path, PASS) as store:
            row = store.get_agent("bigpickle")
            self.assertIsNotNone(row)
            self.assertEqual(row["fingerprint"], keys.fingerprint(
                keys.public_from_private(private)
            ))
    def test_reopen_with_wrong_passphrase_fails(self):
        with SealedStore(self.path, PASS) as store:
            helpers.enrol(store, "bigpickle")
            store.flush()
        with self.assertRaises(ValueError):
            SealedStore(self.path, "definitely-wrong-passphrase")
    def test_stolen_seal_is_not_a_sqlite_database(self):
        with SealedStore(self.path, PASS) as store:
            helpers.enrol(store, "bigpickle")
            identity.challenge(store, "bigpickle")
            store.flush()
        with self.assertRaises(sqlite3.DatabaseError):
            sqlite3.connect(self.path).execute("SELECT name FROM sqlite_master")
    def test_refuses_to_open_cleartext_db_as_sealed(self):
        clear = os.path.join(self.tmp.name, "clear.db")
        Store(clear)  # creates a normal sqlite file
        with self.assertRaises(ValueError) as ctx:
            SealedStore(clear, PASS)
        self.assertIn("not a Sigil sealed store", str(ctx.exception))
    def test_seal_permissions_are_owner_only(self):
        with SealedStore(self.path, PASS) as store:
            helpers.enrol(store, "bigpickle")
            store.flush()
        mode = os.stat(self.path).st_mode & 0o777
        self.assertEqual(mode, 0o600)
    def test_sealed_store_works_with_offbox_anchor(self):
        tip = os.path.join(self.tmp.name, "chain.tip")
        with SealedStore(self.path, PASS, anchor_path=tip) as store:
            private, _ = helpers.enrol(store, "bigpickle")
            identity.challenge(store, "bigpickle")
            result = store.checkpoint(private, "bigpickle", "sealed-daily")
            self.assertEqual(result.get("anchor_path"), tip)
            self.assertTrue(os.path.exists(tip))
            self.assertTrue(store.verify_external_anchors()["ok"])
            store.flush()
        self.assertTrue(is_sealed_file(self.path))
        self.assertTrue(os.path.exists(tip))
 if __name__ == "__main__":
    unittest.main()
 class TestPlaintextWorkingCopyHardening(unittest.TestCase):
    """R5: plaintext tmp must be 0600; SQLCipher remains deferred."""
    def test_plaintext_working_copy_is_owner_only(self):
        import tempfile
        from sigil.sealed import SealedStore
        from sigil import identity, keys
        d = tempfile.mkdtemp()
        path = os.path.join(d, "sealed.db")
        store = SealedStore(path, "test-passphrase-ok")
        try:
            mode = os.stat(store._plain_path).st_mode & 0o777
            self.assertEqual(mode, 0o600, oct(mode))
            dir_mode = os.stat(store._tmpdir.name).st_mode & 0o777
            self.assertEqual(dir_mode, 0o700, oct(dir_mode))
            secret = identity.seed_bootstrap(store, "alice")
            _, pub = keys.generate()
            identity.enrol(store, "alice", keys.export_public_b64(pub), secret)
            store.flush()
            # export.db must not linger world-readable after flush
            export = os.path.join(store._tmpdir.name, "export.db")
            self.assertFalse(os.path.exists(export))
        finally:
            store.close()
--- a/tests/test_viewcodes.py
+++ b/tests/test_viewcodes.py
@ -142,49 +142,4 @@ class TestAuditTrail(Base):
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()
 class TestRevokeAndRekeyCloseViewCodes(Base):
    """R1: revoke / rekey / suspend must close outstanding view codes."""
    def test_redeem_refused_after_permanent_revoke(self):
        issued = helpers.issue(self.store, self.private)
        identity.revoke(self.store, "bigpickle", reason="compromise")
        with self.assertRaises((errors.CodeInvalid, errors.CodeUsed)):
            viewcodes.redeem_code(self.store, issued["code"])
    def test_redeem_refused_during_timed_suspension(self):
        import time
        issued = helpers.issue(self.store, self.private)
        identity.revoke(
            self.store, "bigpickle", until=int(time.time()) + 3600, reason="pause"
        )
        with self.assertRaises((errors.CodeInvalid, errors.CodeUsed)):
            viewcodes.redeem_code(self.store, issued["code"])
    def test_redeem_refused_after_rekey(self):
        issued = helpers.issue(self.store, self.private)
        new_private, new_public = keys.generate()
        secret = identity.seed_bootstrap(self.store, "bigpickle")
        identity.enrol_rekey(
            self.store,
            "bigpickle",
            keys.export_public_b64(new_public),
            secret,
        )
        # Invalidate-on-rekey marks the code used; redeem must not mint a session.
        with self.assertRaises((errors.CodeInvalid, errors.CodeUsed)):
            viewcodes.redeem_code(self.store, issued["code"])
        # New agent can still mint and redeem a fresh code.
        fresh = helpers.issue(self.store, new_private)
        session = viewcodes.redeem_code(self.store, fresh["code"])
        self.assertEqual(session["kind"], "human_via_agent")
    def test_unused_codes_marked_used_on_revoke(self):
        issued = helpers.issue(self.store, self.private)
        digest = viewcodes._hash_code(issued["code"])
        row = self.store.get_code(digest)
        self.assertIsNone(row["used_at"])
        identity.revoke(self.store, "bigpickle")
        row = self.store.get_code(digest)
        self.assertIsNotNone(row["used_at"])