saymrwulf/proof-aware-crypto-tooling-agent
1
0
Fork 0
mirror of https://github.com/saymrwulf/proof-aware-crypto-tooling-agent.git synced 2026-09-04 20:03:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

cockpit: the bridge — six role stations, the crew law, live liveness

Browse source 
Operator verdict on round two: better, but 'still no coherent
application'. The cockpit must provide everything a human crew would
need if no AI were around — as distinct roles that cooperate through
handoffs and never melt into each other. This rebuilds the IA as a
bridge with six stations over shared instruments, in the control-room
tradition (overview -> station -> instrument -> raw files/CLI), with
maker-checker separation of duties encoded in the UI itself.

- / is now the Bridge: whole-system verdict strip, six crew cards with
  live data, and the dispatch (andon) board 'if this happens, who acts'
- /station/{proposer,quorum,operator,cryptographer,architect,newcomer}:
  each console has a fixed anatomy: Mission -> Duties (every duty a
  runnable, verified-real CLI command - the no-AI drill) -> embedded
  live instruments -> 'This station never...' (separation of duties) ->
  Handoffs (receives/delivers)
- Operator gets a real liveness board: on-demand parallel probes (HTTP
  GET on log head/paper/blog/mirror with observed facts + latency; git
  HEAD/cleanliness on all 9 local repos). Never probes on ordinary page
  loads. Verified live: caught this very repo as 'alive, dirty' while
  building it, and confirmed log 13/3488a2d0 + paper 7f140356
- Architect gets the live drift tripwire (ESTATE.md vs estate view)
- modularized per the standing separation-of-concerns order:
  uikit.py (primitives+style), stations.py (role model, pure),
  liveness.py (probes), walletui.py (collectors, instruments, routes)
- crew law test-enforced: bridge crew+dispatch, per-station role
  contract, station distinctness (signature phrases must not bleed
  across roles), explicit-probe semantics; read-only byte sweep now
  covers all 13 routes incl. the probe route
- narrow-viewport fix: breakany for unbreakable paths in headings

Suite 135 -> 139 green. Read-only guarantee unchanged: no mutating
routes; every custody act is a printed command, never a button.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
mrwulf 2026-07-21 16:47:43 +02:00
parent 00be6a396e
commit a18877612d
7 changed files with 1299 additions and 218 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
WALLET.md
View file

@ -146,18 +146,23 @@ pacta wallet mcp --wallet ./my-warden # stdio JSON-RPC MCP server
## The custody cockpit (human surface, read-only) ## The custody cockpit (human surface, read-only)
`pacta wallet cockpit --wallet <dir>` (or `--demo` for a throwaway, `pacta wallet cockpit --wallet <dir>` (or `--demo` for a throwaway,
custody-inert wallet to explore from zero) serves a local web UI for the custody-inert wallet to explore from zero) serves a local web UI
operator: posture (verdict banner, latch, ledger chain re-verified, organized as a **bridge with six role stations** — Proposer, Quorum
pinned quorum), the airgap signature queue (observed, never operated), bench, Operator (with an on-demand liveness board probing every public
the incident and refusal browser, a receipt inspector driven by the service and local repo), Cryptographer, Architect, Newcomer — over
deployed verifier, the estate map, and a plain-language guide with a shared evidence instruments: posture (verdict banner, latch, ledger
full glossary. Design law: it renders recomputed evidence with chain re-verified, pinned quorum), the airgap signature queue (observed,
provenance lines, never cached status; it cannot approve, sign, unlatch, never operated), the incident and refusal browser, a receipt inspector
or modify custody state (byte-level read-only guarantee in driven by the deployed verifier, the estate map, and a plain-language
`tests/test_walletui.py`). UX law: every page explains itself — verdict guide with a full glossary. Three test-enforced laws: the design law (it
in words, per-panel "how to read this" expanders, jargon linked to the renders recomputed evidence with provenance lines, never cached status;
`/guide` glossary (also test-enforced). Details: it cannot approve, sign, unlatch, or modify custody state — byte-level
[docs/cockpit.md](docs/cockpit.md). read-only guarantee in `tests/test_walletui.py`), the UX law (every page
explains itself — verdict in words, per-panel "how to read this"
expanders, jargon linked to the `/guide` glossary), and the crew law
(each station states its mission, duties as runnable commands — the
no-AI drill — its "never" list, and explicit handoffs; roles stay
distinct). Details: [docs/cockpit.md](docs/cockpit.md).
## Agent-native surface (MCP) ## Agent-native surface (MCP)

67
docs/cockpit.md
View file

@ -1,9 +1,14 @@
# The custody cockpit — a read-only surface for the human operator # The custody cockpit — a bridge for the human crew
`pacta wallet cockpit --wallet <dir>` serves a local web UI `pacta wallet cockpit --wallet <dir>` serves a local web UI
(default `http://127.0.0.1:8471`) over an existing warden wallet. (default `http://127.0.0.1:8471`) over an existing warden wallet.
warden has always been agent-native (MCP) and CLI-native; the cockpit is warden has always been agent-native (MCP) and CLI-native; the cockpit is
the third surface — for the human who ultimately answers for the money. the third surface — for the humans who ultimately answer for the money.
It is organized as a **bridge with six role stations** over shared
evidence instruments, in the control-room tradition (overview → station
→ instrument → raw files/CLI): the cockpit provides everything a human
crew would need to run this estate **if no AI were around**.
## The design law ## The design law
@ -36,16 +41,56 @@ page is built from the same anatomy, top to bottom:
5. **The provenance line** — the dashed footer naming the exact function 5. **The provenance line** — the dashed footer naming the exact function
and timestamp that recomputed the panel. and timestamp that recomputed the panel.
The `/guide` view is the manual: what warden is, how to read any page, The `/guide` view is the manual: what warden is, the crew model, how to
the color code, a five-minute tour, a glossary of every term (capsule, read any page, the color code, a five-minute tour, a glossary of every
member, pinning, evidence grades R0–R5, ledger, latch, incident, refusal term (capsule, member, pinning, evidence grades R0–R5, ledger, latch,
receipt, air-gap, attestation/receipt, provenance, DEMO), and an honest incident, refusal receipt, air-gap, attestation/receipt, provenance,
"what this cockpit cannot tell you" section. Navigation tabs state the station, DEMO), and an honest "what this cockpit cannot tell you"
question each view answers. This contract is enforced by tests section. Navigation tabs state the question each view answers. This
(`test_guide_view_explains_every_term`, contract is enforced by tests (`test_guide_view_explains_every_term`,
`test_every_view_carries_lead_nav_and_explainers`, `test_every_view_carries_lead_nav_and_explainers`,
`test_empty_states_are_explained`). `test_empty_states_are_explained`).
## The crew law (roles, not a blur)
**The crew is a team of distinct roles.** Running the estate takes six
roles; in production one financial agent can play every one of them —
but the roles stay separate, cooperate through explicit handoffs, and
never melt into each other. Separation of duties is a custody control:
the one who proposes never approves, the one who verifies never
proposes, the one who watches never overrides the bench.
The **Bridge** (`/`) is the Level-1 overview: the whole-system verdict
strip (custody verdict in words + quorum/ledger/incident/queue chips),
the six crew cards with live data, and the dispatch (andon) board — "if
this happens, who acts". Each **station** (`/station/<id>`) is one
role's console with a fixed anatomy: *Mission* → *Duties* (every duty a
runnable command — the no-AI drill) → live embedded instruments →
*"This station never…"* (the separation-of-duties list) → *Handoffs*
(receives ← / delivers →).
| station | question | live instruments on the console |
|---|---|---|
| **Proposer** (`/station/proposer`) | I need something signed — how do I ask, and what do I do with the answer? | the Queue |
| **Quorum bench** (`/station/quorum`) | Would I stake custody on this evidence? Four seats, one answer each. | the live bench roster (capsule members) |
| **Operator** (`/station/operator`) | Is everything that should be running, running — and is custody unfrozen? | the **liveness board** (on-demand probes of every public service + every local repo), latch, recorded history |
| **Cryptographer** (`/station/cryptographer`) | Does the evidence really prove what it claims — no more, no less? | the Inspect verifier |
| **Architect** (`/station/architect`) | Does the map still match the territory? | the live **drift tripwire** (ESTATE.md vs estate view) |
| **Newcomer** (`/station/newcomer`) | What is all this? Where do I start? | the first-hour checklist |
The liveness board probes **only when the operator presses «Probe
now»** — the cockpit never phones home on an ordinary page load. Probes
are read-only observations (HTTP GET on the public services, `git
rev-parse`/`status` on local checkouts) and report observed facts with
latency; liveness is pulses, not honesty — honesty is the
Cryptographer's replay.
The crew law is test-enforced: `test_bridge_shows_crew_and_dispatch`,
`test_every_station_defines_role_contract` (mission/duties/commands/
never-list/handoffs on all six), `test_stations_are_distinct_roles`
(each role's signature phrase appears on its own station and on no
other — no melting), `test_operator_probe_is_explicit_and_live`.
## The read-only guarantee ## The read-only guarantee
The cockpit cannot approve, sign, unlatch, or modify custody state. It The cockpit cannot approve, sign, unlatch, or modify custody state. It
@ -57,11 +102,11 @@ wallet directory hash-identical. Human approve/deny is deliberately NOT
here — that would be a custody-semantics change, which belongs to a here — that would be a custody-semantics change, which belongs to a
separate, explicitly reviewed milestone. separate, explicitly reviewed milestone.
## The six views ## The instruments (shared evidence views)
| view | answers | recomputed by | | view | answers | recomputed by |
|---|---|---| |---|---|---|
| **Posture** (`/`) | *Is custody healthy right now?* Verdict banner, then: custody latch, ledger with full hash-chain re-verification, the pinned quorum members (backend, component, evidence grade, source commit, binary fingerprint), signing rules verbatim, incident/refusal counts | `Wallet.posture()` / `Wallet.verify_ledger()` | | **Posture** (`/posture`) | *Is custody healthy right now?* Verdict banner, then: custody latch, ledger with full hash-chain re-verification, the pinned quorum members (backend, component, evidence grade, source commit, binary fingerprint), signing rules verbatim, incident/refusal counts | `Wallet.posture()` / `Wallet.verify_ledger()` |
| **Queue** (`/queue`) | *What awaits the offline signer?* Parked air-gap signing requests (outbox) and whether the device has answered (inbox) — observed, never operated | airgap outbox/inbox listing | | **Queue** (`/queue`) | *What awaits the offline signer?* Parked air-gap signing requests (outbox) and whether the device has answered (inbox) — observed, never operated | airgap outbox/inbox listing |
| **Incidents** (`/incidents`) | *What has ever gone wrong?* Incident records and signed refusal receipts, verbatim, newest first — with the page explaining why empty is the good state | `incidents/*.json`, `receipts/*.json` | | **Incidents** (`/incidents`) | *What has ever gone wrong?* Incident records and signed refusal receipts, verbatim, newest first — with the page explaining why empty is the good state | `incidents/*.json`, `receipts/*.json` |
| **Inspect** (`/inspect`) | *Can I check a receipt myself?* Paste an attestation + transparency receipt + log public key; the verdict, per-signature results, and diagnostics come verbatim from the deployed verifier | `pacta.transparency.verify_receipt` | | **Inspect** (`/inspect`) | *Can I check a receipt myself?* Paste an attestation + transparency receipt + log public key; the verdict, per-signature results, and diagnostics come verbatim from the deployed verifier | `pacta.transparency.verify_receipt` |

191
src/pacta/liveness.py Normal file
View file

@ -0,0 +1,191 @@
"""liveness - the Operator station's board: is everything that should be
running actually running?
Probes are READ-ONLY observations (HTTP GET on the public services, `git`
queries on local working copies) and run ONLY when the operator presses
«Probe now» - the cockpit never phones home on an ordinary page load.
Nothing here touches wallet state; the byte-level read-only guarantee in
tests/test_walletui.py covers the probe route too.
Each probe answers exactly one question - reachable? present? - and shows
the observed facts (status, head, latency). Whether the observed facts
are HONEST is a different station's job (the Cryptographer replays; this
board only watches pulses).
"""
from __future__ import annotations
import hashlib
import json
import subprocess
import time
import urllib.request
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import Any
from .uikit import esc, explain, provenance
_TIMEOUT = 4 # seconds per probe; the board is on-demand, not a pager
# (label, url, expectation as stated on the estate map)
SERVICES: list[tuple[str, str, str]] = [
("log head", "https://ltl.zkdefi.org/v1/sth", "ALWAYS ON (provider, read-only)"),
("paper", "https://ltl.zkdefi.org/paper", "ALWAYS ON (static, behind caddy)"),
("blog", "https://blog.zkdefi.org", "ALWAYS ON (static, behind caddy)"),
("public mirror (Forgejo)", "https://zkdefi.org", "ALWAYS ON"),
]
# local working copies expected as siblings of this repo checkout
LOCAL_REPOS: list[str] = [
"lean-transparency-log",
"ltl-accumulator-verified",
"verifying-crypto-with-lean",
"dalek-ed25519-verified",
"anza-ed25519-verified",
"risc0-ed25519-verified",
"betrusted-ed25519-verified",
"pasta-pallas-verified",
"proof-aware-crypto-tooling-agent",
]
def default_repos_root() -> Path:
"""Siblings of this repo checkout (…/FormalVerification)."""
return Path(__file__).resolve().parents[3]
def _probe_service(label: str, url: str, expect: str) -> dict[str, Any]:
start = time.monotonic()
try:
with urllib.request.urlopen(url, timeout=_TIMEOUT) as resp:
body = resp.read()
status = resp.status
ms = int((time.monotonic() - start) * 1000)
fact = f"HTTP {status}"
if label == "log head":
try:
sth = json.loads(body)
fact = (f"HTTP {status} · tree_size {sth.get('tree_size')} · "
f"root {str(sth.get('root_hash', ''))[:8]}…")
except Exception: # noqa: BLE001 - fact stays the bare status
pass
elif label == "paper":
fact = f"HTTP {status} · sha256 {hashlib.sha256(body).hexdigest()[:8]}… · {len(body)//1024} KiB"
return {"label": label, "expect": expect, "ok": status == 200,
"fact": fact, "ms": ms}
except Exception as error: # noqa: BLE001 - a dead service is a result, not a crash
ms = int((time.monotonic() - start) * 1000)
return {"label": label, "expect": expect, "ok": False,
"fact": f"UNREACHABLE — {type(error).__name__}: {error}", "ms": ms}
def _probe_repo(root: Path, name: str) -> dict[str, Any]:
path = root / name
if not path.is_dir():
return {"label": name, "expect": "local working copy", "ok": False,
"fact": f"MISSING — no directory at {path}", "ms": 0}
start = time.monotonic()
try:
head = subprocess.run(
["git", "-C", str(path), "rev-parse", "--short", "HEAD"],
capture_output=True, text=True, timeout=10, check=True).stdout.strip()
dirty = subprocess.run(
["git", "-C", str(path), "status", "--porcelain"],
capture_output=True, text=True, timeout=10, check=True).stdout
ms = int((time.monotonic() - start) * 1000)
n_dirty = len([line for line in dirty.splitlines() if line.strip()])
if n_dirty:
return {"label": name, "expect": "local working copy", "ok": True,
"warn": True, "fact": f"HEAD {head} · {n_dirty} uncommitted change(s)",
"ms": ms}
return {"label": name, "expect": "local working copy", "ok": True,
"fact": f"HEAD {head} · clean", "ms": ms}
except Exception as error: # noqa: BLE001
ms = int((time.monotonic() - start) * 1000)
return {"label": name, "expect": "local working copy", "ok": False,
"fact": f"NOT READABLE AS GIT — {type(error).__name__}: {error}", "ms": ms}
def collect_liveness(repos_root: Path | None = None) -> dict[str, Any]:
"""Run every probe, in parallel, once — called only on explicit demand."""
root = repos_root or default_repos_root()
try:
with ThreadPoolExecutor(max_workers=8) as pool:
services = list(pool.map(lambda s: _probe_service(*s), SERVICES))
repos = list(pool.map(lambda n: _probe_repo(root, n), LOCAL_REPOS))
return {"ok": True, "via": "live HTTP GET + git rev-parse/status, on demand",
"data": {"services": services, "repos": repos, "root": str(root)}}
except Exception as error: # noqa: BLE001
return {"ok": False, "via": "live HTTP GET + git", "error": error}
def _row(p: dict[str, Any]) -> str:
if p["ok"] and p.get("warn"):
pill = '<span class="pill warn">alive, dirty</span>'
elif p["ok"]:
pill = '<span class="pill ok">alive</span>'
else:
pill = '<span class="pill bad">DOWN / MISSING</span>'
return (f"<tr><td>{esc(p['label'])}</td><td>{pill}</td>"
f"<td class='mono'>{esc(p['fact'])}</td>"
f"<td class='muted'>{esc(p['expect'])}</td>"
f"<td class='mono'>{p['ms']} ms</td></tr>")
def render_liveness(result: dict[str, Any] | None) -> str:
"""The board. result=None means: not probed yet this page load."""
from .uikit import failed_panel
if result is None:
body = (
"<p class='empty'>This board has not probed yet — probes run only when "
"you press the button, so the cockpit never phones home on an ordinary "
"page load. Press «Probe now» to check every service and repo live.</p>"
"<p><a class='btnlink' href='/station/operator?probe=1'>Probe now</a></p>")
return (f"<div class='panel'><h3 style='margin-top:0'>Liveness board — what is "
f"actually running?</h3>{body}"
+ explain(
"<ul><li>Service probes are plain HTTP GETs (4-second timeout) "
"reporting status, observed facts, and latency.</li>"
"<li>Local repos are checked with <code>git rev-parse</code> / "
"<code>git status</code>: present, at which commit, clean or "
"dirty.</li>"
"<li><span class='pill ok'>alive</span> = responded / present · "
"<span class='pill warn'>alive, dirty</span> = present with "
"uncommitted changes · <span class='pill bad'>DOWN / MISSING</span> "
"= no response or not found.</li>"
"<li>Liveness is not honesty: this board only checks pulses. "
"Whether the answers are cryptographically true is the "
"Cryptographer's replay work.</li></ul>")
+ provenance("no probe run this page load (press the button)")
+ "</div>")
if not result["ok"]:
return failed_panel("Liveness board", result["via"], result["error"])
d = result["data"]
service_rows = "".join(_row(p) for p in d["services"])
repo_rows = "".join(_row(p) for p in d["repos"])
down = [p["label"] for p in d["services"] + d["repos"] if not p["ok"]]
verdict = ('<p class="plain"><span class="pill bad">ATTENTION</span> not alive: '
+ ", ".join(f"<code>{esc(x)}</code>" for x in down) + "</p>"
if down else
'<p class="plain"><span class="pill ok">all probed targets alive</span></p>')
return (
f"<div class='panel'><h3 style='margin-top:0'>Liveness board — what is "
f"actually running?</h3>{verdict}"
"<h3>Public services</h3>"
f"<div class='tablewrap'><table><tr><th>service</th><th>state</th>"
f"<th>observed</th><th>expected</th><th>latency</th></tr>{service_rows}</table></div>"
f"<h3>Local working copies <span class='muted mono breakany'>under {esc(d['root'])}</span></h3>"
f"<div class='tablewrap'><table><tr><th>repo</th><th>state</th>"
f"<th>observed</th><th>expected</th><th>took</th></tr>{repo_rows}</table></div>"
"<p><a class='btnlink' href='/station/operator?probe=1'>Probe again</a></p>"
+ explain(
"<ul><li>«observed» is what the probe just saw: HTTP status and payload "
"facts for services, HEAD commit and cleanliness for repos.</li>"
"<li>«expected» is what the estate map says this target should be — a "
"target can be alive and still wrong (that is the Cryptographer's "
"beat).</li>"
"<li>DOWN on a public service: check your own network first, then the "
"server. MISSING on a repo: this machine simply has no checkout — clone "
"it if this machine should hold one.</li></ul>")
+ provenance(result["via"]) + "</div>"
)

452
src/pacta/stations.py Normal file
View file

@ -0,0 +1,452 @@
"""stations - the role model of the warden bridge.
The cockpit's third law (after the design law and the UX law): THE CREW
IS A TEAM OF DISTINCT ROLES, NOT A BLUR. If no AI were around, running
this estate would take a crew: a proposer, a quorum bench, an operator,
a cryptographer, an architect, and a newcomer finding their feet. In
production one financial agent can play every role - but the roles stay
separate, hand work to each other through explicit interfaces, and never
melt into one another. Separation of duties is a custody control, not a
metaphor: the one who proposes never approves, the one who verifies
never proposes, the one who watches never overrides the bench.
Design lineage, deliberately borrowed:
- control-room HMI hierarchy (overview -> station -> instrument -> raw
detail; color reserved for state that needs attention),
- mission-control discipline consoles (one role, one console, explicit
handoffs on the loop),
- banking's maker-checker (four-eyes) separation of duties,
- checklist culture: every duty is a runnable command, not advice.
This module is pure presentation over passed-in data: no wallet imports,
no I/O. Live evidence panels are built by walletui and handed in as
`embeds`, so the read-only guarantee and the provenance discipline stay
in one place.
"""
from __future__ import annotations
from typing import Any
from .uikit import cmd_block, esc, explain
# ---------------------------------------------------------------------------
# the crew
# ---------------------------------------------------------------------------
STATIONS: list[dict[str, Any]] = [
{
"id": "proposer", "name": "Proposer", "monogram": "PR",
"hue": "#a86a10", "tint": "#fdf0da",
"question": "I need something signed — how do I ask, and what do I do with the answer?",
"card": "Originates signing requests; consumes signatures and refusal receipts.",
"lead": ("You are at the <strong>Proposer</strong> station. You originate signing "
"requests and you live with the answers — a signature or a written "
"refusal. You never approve anything yourself; that separation is what "
"makes your requests trustworthy."),
"mission": ("Turn an intent (\"pay this\", \"sign that\") into a precise, "
"fingerprinted request; submit it through the wallet's front door; "
"read every refusal receipt as instructions, not rejection."),
"duties": [
("Form the request precisely",
"Know the exact bytes you want signed and compute their fingerprint — the "
"device will sign those bytes and nothing else, and every later dispute is "
"settled by this hash.",
"sha256sum payload.bin"),
("Submit through the wallet's front door",
"The proposal interface is agent-native: the <code>request_signature</code> "
"tool on the wallet's MCP surface. A human drives the same surface over "
"stdio (the JSON-RPC lines are in WALLET.md).",
"pacta wallet mcp --wallet WALLET_DIR\n# then call the request_signature tool"),
("Track your request",
"Your request appears in the Queue instrument (embedded below) as "
"«awaiting device» until the offline signer answers.",
None),
("On refusal: read the receipt, fix, retry",
"A refusal receipt names the rule (<code>code</code>), what was missing, and "
"the <code>remediation</code>. It is a to-do list, not a verdict on you.",
None),
("On success: verify what you got",
"Never forward a signature you have not checked. For a Solana transaction, "
"the quorum re-verifies every signature locally:",
"pacta wallet treasury-verify --wallet WALLET_DIR --tx-file tx.bin"),
],
"never": [
"Approve or verify the evidence behind your own proposal — that is the "
"Quorum bench's seat, and the whole point is that it is not yours.",
"Touch the air-gap device or its outbox/inbox files — the Operator owns "
"that walk.",
"Clear a latch, or edit <code>policy.json</code> to make your own request "
"fit — policy changes are the Operator's deliberate, recorded act.",
],
"receives": ["a signature, or a refusal receipt — from the wallet",
"custody posture answers — from the Operator"],
"delivers": ["fingerprinted signing requests — to the signing firewall",
"escalations after repeated refusals — to the Operator"],
"instruments": [("/queue", "Queue"), ("/incidents", "Incidents (refusals)")],
},
{
"id": "quorum", "name": "Quorum bench", "monogram": "QM",
"hue": "#3b4d8f", "tint": "#eef0f7",
"question": "Would I stake custody on this evidence? Four seats, one answer each.",
"card": "Four independent verifier seats; unanimity admits, one dissent latches.",
"lead": ("You are at the <strong>Quorum bench</strong> — four seats, one per "
"independently built verifier (dalek, anza, risc0, betrusted). Each seat "
"answers for itself. The bench never averages: unanimity admits a "
"component, a single dissent freezes custody. Your value is your "
"independence."),
"mission": ("Hold an independent verdict on every piece of cryptographic "
"evidence the wallet is asked to trust. Your seat's judgment must "
"come from your seat's binary, built from your seat's verified "
"sources — nobody else's."),
"duties": [
("Know your seat",
"Which verified repository you are built from, at which commit, with which "
"binary fingerprint — the live bench roster is embedded below.",
None),
("Rebuild your member from pinned proven sources",
"When sources or toolchains move, rebuild from the pinned fork checkouts "
"and let the capsule re-pin your fingerprint:",
"pacta wallet build-quorum --sources-root FORK_CHECKOUTS_DIR"),
("Guard your independence",
"A shared toolchain is a shared bug. Do not copy another seat's build "
"artifacts, caches, or patches — four seats that agree because they are "
"secretly one seat protect nothing.",
None),
("On divergence: your dissent worked",
"If your seat says INVALID while others say OK, custody latches and an "
"incident is written. That is the system succeeding, not you failing. Read "
"the incident (embedded on the Incidents instrument) and defend your "
"verdict to the Operator.",
None),
],
"never": [
"Propose a request — the bench judges evidence, it never originates spends.",
"Vote another seat's verdict, or harmonize before answering — the bench "
"never averages; unanimity or latch.",
"Clear a latch your own dissent caused — the Operator investigates; you are "
"a witness, not the judge of your own alarm.",
],
"receives": ["component evidence (attestations + receipts) — from the wallet's "
"inbound boundary",
"rebuilt source workspaces — from the Architect's pinned forks"],
"delivers": ["a unanimous admit, or a latch-tripping dissent — to the wallet",
"divergence incidents — to the Operator"],
"instruments": [("/posture", "Posture (bench roster)"), ("/incidents", "Incidents")],
},
{
"id": "operator", "name": "Operator", "monogram": "OP",
"hue": "#1e7f4f", "tint": "#e2f2e9",
"question": "Is everything that should be running, running — and is custody unfrozen?",
"card": "Watches liveness of every service and repo; owns latch recovery.",
"lead": ("You are at the <strong>Operator</strong> station. You watch the "
"liveness of everything — every service, every repo, the wallet's own "
"health — and you own the emergency procedures. When the latch trips, "
"everyone else stops and you start."),
"mission": ("Keep the estate observably alive: probe the public services, check "
"the local repos, re-verify the wallet daily, advance the log pin, "
"and run latch recovery by the book. You are the only station that "
"may clear a latch — deliberately, with a permanent written note."),
"duties": [
("Daily watch",
"Press «Probe now» on the liveness board below — it re-checks every "
"service and repo live, on demand. Then re-verify the wallet from the "
"command line:",
"pacta wallet status --wallet WALLET_DIR\npacta wallet verify-ledger --wallet WALLET_DIR"),
("Advance the log pin",
"Fetch the latest signed head of the transparency log, verify signature "
"and consistency against your pinned size, and advance the pin — this is "
"the split-view/rollback defense:",
"pacta sth-refresh --url https://ltl.zkdefi.org \\\n --sth-store sth-store.json --log-public-key log.pub"),
("Witness the published log",
"Audit a full clone: recompute every prefix root, check every historical "
"signed head. An unwatched log is safe only on paper — you are the watcher:",
"pacta witness-audit --published-dir path/to/lean-transparency-log"),
("On latch: run the book",
"Follow <code>docs/runbook-latch.md</code> step by step. Only when the "
"cause is understood and fixed, clear the latch — the note is recorded "
"permanently in the ledger:",
'pacta wallet unlatch --wallet WALLET_DIR --note "root cause and fix"'),
],
"never": [
"Clear a latch without a written root cause — the CLI refuses an empty "
"note, and the ledger keeps it forever.",
"Rewrite log history: never force-push the log repo, never edit or delete "
"under <code>entries/</code>, never re-sign a published head, never "
"backdate. The log is append-only or it is nothing.",
"Override the bench: if the quorum diverged, the answer is investigation, "
"not a fifth vote.",
"Sign or propose — you hold the brakes, not the pen.",
],
"receives": ["incidents and latch events — from the wallet and the Quorum bench",
"escalations — from the Proposer"],
"delivers": ["unlatch decisions with permanent notes — to the wallet ledger",
"outage and repair notes — to the Architect (map updates)"],
"instruments": [("/posture", "Posture"), ("/incidents", "Incidents"),
("/queue", "Queue")],
},
{
"id": "cryptographer", "name": "Cryptographer", "monogram": "CR",
"hue": "#6d4a8f", "tint": "#f0e8f7",
"question": "Does the evidence really prove what it claims — no more, no less?",
"card": "Re-verifies receipts, replays the log offline, guards claim boundaries.",
"lead": ("You are at the <strong>Cryptographer</strong> station. You take "
"nothing on trust that you can recompute: receipts, inclusion proofs, "
"signed heads, the whole log. And you guard the boundary of every "
"claim — what is proven, and exactly where the proof stops."),
"mission": ("Independently re-verify any evidence artifact anyone hands you, "
"replay the public log offline, and keep everyone honest about what "
"the mathematics does and does not cover."),
"duties": [
("Verify a receipt end to end",
"Use the Inspect instrument (embedded below) or the CLI with the hardened "
"flags — pin store, freshness policy, verified-verifier requirement:",
"pacta receipt-verify --attestation a.json --receipt r.json \\\n"
" --log-public-key log.pub --sth-store sth-store.json \\\n"
" --max-sth-age-seconds 604800 --require-verified-verifier"),
("Fetch fresh evidence yourself",
"Never verify only what you were handed — fetch from the live log and "
"verify locally:",
"pacta log-fetch --url https://ltl.zkdefi.org --component dalek-ed25519-verified"),
("Replay the whole log offline",
"In a clone of the published log repo: a fail-closed, standard-library "
"verifier re-checks every leaf, every signed head, every receipt — with "
"its adversarial self-test shipped beside it:",
"python3 verify.py --all\npython3 verify_selftest.py"),
("Guard the claim boundary",
"The evidence grades (R0–R5) are exact: R4 means machine-checked proofs on "
"the documented boundary — SHA-512 opaque, reproducible builds out of "
"scope (R5). A claim stretched past its boundary is a false claim. The "
"grading tools are <code>pacta claims</code>, <code>pacta score</code>, "
"<code>pacta report</code>.",
None),
],
"never": [
"Accept a green light you did not recompute — including this cockpit's.",
"Extend a claim beyond its stated boundary — «verified» never means more "
"than the certificate says.",
"Treat the log as a truth oracle — it is an accountability ledger: leaves "
"can lie, and only independent replay catches a fabricated claim.",
],
"receives": ["evidence artifacts to audit — from anyone",
"fresh attestations and receipts — from the live log"],
"delivers": ["audit verdicts — to the Operator and the Proposer",
"claim-boundary corrections — to the Architect (docs and map)"],
"instruments": [("/inspect", "Inspect"), ("/guide", "Guide (limits)")],
},
{
"id": "architect", "name": "Architect", "monogram": "AR",
"hue": "#2b5b78", "tint": "#e8eef2",
"question": "Does the map still match the territory — every repo, service, loop?",
"card": "Keeps the estate map true; watches the loops and the public boundary.",
"lead": ("You are at the <strong>Architect</strong> station. The estate is many "
"repos, services, mirrors, and two self-referential loops — too much "
"for anyone's head, which is why the map exists. Your job is that the "
"map never lies: about what exists, what runs, and what is public."),
"mission": ("Keep the estate map congruent with reality after every change, "
"keep its two renderings from drifting, watch the loops, and gate "
"what may be named in public."),
"duties": [
("Walk the map after every landed change",
"Open the Estate instrument and check the changed entity's card — "
"runtime (always-on / on-demand / not-running / static), mutability, "
"custody lane. The live drift tripwire is embedded below.",
None),
("Keep the two renderings synced",
"The map exists twice: <code>ESTATE.md</code> (canonical, committed) and "
"the cockpit's estate view. A name-level tripwire test fails the suite if "
"they drift — run it after map edits:",
"python3 scripts/mini_pytest.py"),
("Watch the loops",
"Loop 1: the dogfood signer is attested at leaf 8 of the very log it "
"signs. Loop 2: entry 13 — the log carries the kernel-checked mechanization "
"of its own accumulator's soundness. Both must stay tellable in one "
"breath; if an explanation of a loop stops being crisp, the estate has "
"drifted somewhere.",
None),
("Gate the public boundary",
"Public documents list only entities whose existence is already public or "
"must be public for trust. Private infrastructure stays unnamed, "
"everywhere, always.",
None),
],
"never": [
"Name private infrastructure in a public artifact — not in maps, not in "
"docs, not in commit messages.",
"Let a generated file drift from its canonical source — when you fix a "
"published file, find and fix its generator in the same change.",
"Redraw the map from memory — the map is recomputed from the repos and "
"services, never from recollection.",
],
"receives": ["outage and repair notes — from the Operator",
"claim-boundary corrections — from the Cryptographer"],
"delivers": ["the updated, drift-guarded map — to everyone "
"(ESTATE.md + the estate view)"],
"instruments": [("/estate", "Estate map"), ("/guide", "Guide")],
},
{
"id": "newcomer", "name": "Newcomer", "monogram": "NC",
"hue": "#0f766e", "tint": "#e0f2f0",
"question": "What is all this? Where do I even start?",
"card": "Learns the system hands-on with the DEMO wallet; supplies fresh eyes.",
"lead": ("You are at the <strong>Newcomer</strong> station — everyone's first "
"station, including the people now sitting at the other five. Your "
"first hour is mapped out below, and your confusion is valuable: it "
"finds the gaps the veterans stopped seeing."),
"mission": ("Learn the system hands-on, with a wallet that cannot hurt "
"anything, until you can read the Bridge at a glance — then pick a "
"station and shadow it."),
"duties": [
("Your first hour, step 1: run the demo",
"A throwaway, custody-inert wallet with fake members — every view has "
"content, nothing can sign anything real:",
"pacta wallet cockpit --demo"),
("Step 2: read the Guide",
"All of it — ten minutes. What warden is, how to read any page, the color "
"code, and a glossary of every term you will meet.",
None),
("Step 3: take the five-minute tour",
"Posture → Incidents → Queue → Inspect → Estate, in that order, reading "
"each page's verdict first.",
None),
("Step 4: verify something real",
"Paste the sample evidence from <code>examples/wallet-evidence/</code> "
"into the Inspect instrument and watch the deployed verifier accept it — "
"then break one character and watch it refuse.",
None),
("Step 5: pick a station and shadow it",
"Read that station's mission, duties, and its «never» list — the never "
"list is the fastest way to understand a role.",
None),
],
"never": [
"Pretend to understand — every other station once sat exactly here.",
"Assume a confusing page is your fault: if a page confuses you, that is a "
"bug in the page, not in you. Report it.",
],
"receives": ["the Guide, the demo wallet, and patient answers — from every "
"station"],
"delivers": ["fresh eyes: every page that confuses you, reported — to the "
"Architect and the Operator"],
"instruments": [("/guide", "Guide"), ("/inspect", "Inspect"),
("/estate", "Estate map")],
},
]
STATION_BY_ID = {s["id"]: s for s in STATIONS}
# the andon board: event -> who acts, with what
DISPATCH: list[tuple[str, str, str]] = [
("A request was refused", "proposer",
"read the refusal receipt: code → missing → remediation, then retry"),
("CUSTODY FROZEN, or a new incident", "operator",
"runbook-latch, root cause, permanent unlatch note (plus the Quorum bench if it "
"was a divergence)"),
("A new component wants to be trusted", "quorum",
"evidence through the inbound boundary; unanimity admits, one dissent latches"),
("Someone handed you evidence", "cryptographer",
"Inspect, or receipt-verify with the hardened flags — recompute, never trust"),
("A service or repo looks dead", "operator",
"liveness board: Probe now"),
("The map feels wrong", "architect",
"estate view + drift tripwire; fix the generator, not just the page"),
("“I don’t understand any of this”", "newcomer",
"the Newcomer station is the entry point, not an insult"),
]
# ---------------------------------------------------------------------------
# renderers (pure)
# ---------------------------------------------------------------------------
def _role_vars(s: dict[str, Any]) -> str:
return f"--role:{s['hue']};--roletint:{s['tint']}"
def render_bridge(strip_html: str, live: dict[str, str]) -> str:
"""The Level-1 overview: whole-system verdict, then the crew."""
cards = "".join(
f'<div class="stationcard" style="{_role_vars(s)}">'
f'<div><span class="monogram">{s["monogram"]}</span> '
f'<strong>{esc(s["name"])}</strong></div>'
f'<div class="q">{esc(s["question"])}</div>'
f'<div class="plain">{esc(s["card"])}</div>'
f'{live.get(s["id"], "")}'
f'<a class="take" href="/station/{s["id"]}">Take this station →</a></div>'
for s in STATIONS)
dispatch_rows = "".join(
f'<tr><td>{event}</td>'
f'<td><a href="/station/{sid}">{esc(STATION_BY_ID[sid]["name"])}</a></td>'
f'<td>{action}</td></tr>'
for event, sid, action in DISPATCH)
return (
strip_html
+ "<h2>The crew — who does what</h2>"
"<p class='plain'>Running this estate without an AI takes six roles. Each "
"station page states its mission, its duties as runnable commands, what it "
"hands to whom — and what it never does. Roles cooperate through those "
"handoffs; they do not blur into each other.</p>"
f"<div class='crew'>{cards}</div>"
"<h2>If this happens, who acts</h2>"
"<div class='tablewrap'><table>"
"<tr><th>event</th><th>station</th><th>first move</th></tr>"
f"{dispatch_rows}</table></div>"
+ explain(
"<ul><li>This bridge is the one-glance overview: the verdict strip on top "
"is the whole system's state, recomputed on load.</li>"
"<li>The crew cards are the six roles; «Take this station» opens that "
"role's console with its duties and handoffs.</li>"
"<li>The dispatch table is the andon board: when something happens, it "
"names the station that acts first — nobody improvises ownership during an "
"incident.</li>"
"<li>The layout follows control-room practice: overview (this bridge) → "
"station (a role's console) → instrument (shared evidence panels) → raw "
"files and CLI. Deeper is always one click, never a guess.</li></ul>")
)
def render_station(s: dict[str, Any], embeds: list[str]) -> str:
"""One role's console: mission, duties (with commands), live instruments,
the never-list, and explicit handoffs."""
duties = "".join(
f'<div class="duty"><strong>{i}. {title}</strong>'
f'<div class="why">{why}</div>'
f'{cmd_block(command) if command else ""}</div>'
for i, (title, why, command) in enumerate(s["duties"], start=1))
nevers = "".join(f"<li>{item}</li>" for item in s["never"])
receives = "".join(f"<li>{item}</li>" for item in s["receives"])
delivers = "".join(f"<li>{item}</li>" for item in s["delivers"])
instruments = " · ".join(
f'<a href="{href}">{esc(label)}</a>' for href, label in s["instruments"])
embedded = "".join(embeds)
return (
f'<div class="rolehead" style="{_role_vars(s)}">'
f'<span class="monogram">{s["monogram"]}</span>'
f'<div><h2>{esc(s["name"])} station</h2>'
f'<div class="q">{esc(s["question"])}</div></div></div>'
f'<div class="panel"><h3 style="margin-top:0">Mission</h3>'
f'<p class="plain">{s["mission"]}</p>'
f'<p class="muted">Instruments this station works with: {instruments}</p></div>'
f'<div class="panel"><h3 style="margin-top:0">Duties — the no-AI drill</h3>'
"<p class='plain'>Every duty is a runnable command or a concrete act — this is "
"the work, not advice about the work.</p>"
f'{duties}</div>'
f'{embedded}'
f'<div class="panel" style="border-left:4px solid var(--bad)">'
f'<h3 style="margin-top:0">This station never…</h3>'
"<p class='plain'>Separation of duties is the control that makes the team "
"trustworthy — these lines are load-bearing, not etiquette.</p>"
f'<ul class="never">{nevers}</ul></div>'
f'<div class="panel"><h3 style="margin-top:0">Handoffs</h3>'
f'<div class="hand">'
f'<div class="hcol"><b>RECEIVES</b><ul class="diag">{receives}</ul></div>'
f'<div class="hcol"><b>DELIVERS</b><ul class="diag">{delivers}</ul></div>'
f'</div>'
+ explain(
"<ul><li>Handoffs are the team's interfaces: what this station takes in, "
"what it hands out, and to whom. Work moves between stations only through "
"these — that is how distinct roles cooperate without merging.</li>"
"<li>If you are alone (or you are the AI), you may hold several stations — "
"but you switch between them explicitly, one at a time, and the handoffs "
"still apply to yourself.</li></ul>")
+ "</div>"
)

162
src/pacta/uikit.py Normal file
View file

@ -0,0 +1,162 @@
"""uikit - shared presentation primitives for the warden cockpit.
Pure string builders and the one stylesheet. No wallet imports, no I/O:
this module can be reasoned about (and tested) as text in, text out.
The cockpit's two laws live in walletui's docstring; every helper here
exists to serve them - provenance lines, loud failure panels, glossary
links, per-panel explainers.
"""
from __future__ import annotations
import html
from datetime import datetime, timezone
from typing import Any
STYLE = """
:root{--ink:#1c2430;--ink2:#5a6675;--line:#dde2e9;--ok:#1e7f4f;--okbg:#e2f2e9;
--bad:#a3242c;--badbg:#fbe4e6;--warn:#a86a10;--warnbg:#fdf0da;
--accent:#3b4d8f;--accentbg:#eef0f7;--bg:#f8f9fa}
*{box-sizing:border-box}
body{font-family:system-ui,sans-serif;max-width:62rem;margin:0 auto;
padding:1.4rem 1.2rem 4rem;color:var(--ink);line-height:1.55;background:var(--bg)}
h1{font-size:1.35rem;margin:.2rem 0 0}
h2{font-size:1.05rem;margin:1.6rem 0 .5rem}
h3{font-size:.95rem;margin:1rem 0 .3rem}
code{font-family:ui-monospace,Menlo,Consolas,monospace;background:#eef0f3;
border-radius:4px;padding:.08rem .3rem;font-size:.88em}
.sub{color:var(--ink2);font-size:.85rem;margin:.3rem 0 .6rem}
.navrow{display:flex;gap:.5rem;flex-wrap:wrap;align-items:stretch;margin:.45rem 0}
.navtag{font-size:.62rem;font-weight:700;letter-spacing:.08em;color:var(--ink2);
align-self:center;min-width:6.2rem}
.navrow a{color:var(--accent);text-decoration:none;border:1px solid var(--line);
background:#fff;border-radius:6px;padding:.3rem .6rem;font-size:.82rem;
display:flex;flex-direction:column;line-height:1.25;min-width:6.4rem}
.navrow a.here{border-color:var(--accent);font-weight:600;background:var(--accentbg)}
.navrow a .navsub{font-size:.65rem;color:var(--ink2);font-weight:400}
.banner{background:var(--warnbg);border:1px solid var(--warn);color:var(--warn);
border-radius:6px;padding:.45rem .8rem;font-size:.82rem;font-weight:600}
.banner a{color:var(--warn)}
.lead{font-size:.92rem;margin:.9rem 0 .2rem}
.verdict{border-radius:8px;padding:.8rem 1.1rem;margin:.8rem 0;border:1px solid}
.verdict strong{font-size:1.05rem;letter-spacing:.02em}
.verdict p{margin:.3rem 0 0;font-size:.88rem;font-weight:400}
.verdict.ok{background:var(--okbg);border-color:var(--ok);color:var(--ok)}
.verdict.warn{background:var(--warnbg);border-color:var(--warn);color:var(--warn)}
.verdict.bad{background:var(--badbg);border-color:var(--bad);color:var(--bad)}
.panel{background:#fff;border:1px solid var(--line);border-radius:8px;
padding:.9rem 1.1rem;margin:.7rem 0}
.panel.bad{border-color:var(--bad);background:var(--badbg)}
.prov{color:var(--ink2);font-size:.72rem;margin-top:.6rem;border-top:1px dashed var(--line);
padding-top:.35rem}
.pill{display:inline-block;border-radius:9px;padding:.06rem .55rem;font-size:.76rem;
font-weight:700}
.pill.ok{background:var(--okbg);color:var(--ok)}
.pill.bad{background:var(--badbg);color:var(--bad)}
.pill.warn{background:var(--warnbg);color:var(--warn)}
a.help{display:inline-block;width:1.05rem;height:1.05rem;line-height:1.05rem;text-align:center;
border-radius:50%;background:var(--accentbg);color:var(--accent);font-size:.72rem;
font-weight:700;text-decoration:none;vertical-align:.15em}
details.explain{margin-top:.55rem;font-size:.82rem}
details.explain summary{cursor:pointer;color:var(--accent);font-weight:600;font-size:.78rem}
details.explain .expl{color:var(--ink2);margin:.4rem 0 0;padding:.5rem .7rem;
background:var(--accentbg);border-radius:6px}
details.explain .expl ul{margin:.3rem 0;padding-left:1.1rem}
details.explain .expl li{margin:.15rem 0}
.plain{font-size:.88rem;margin:.3rem 0 .6rem}
.empty{color:var(--ink2);font-size:.88rem;background:var(--accentbg);border-radius:6px;
padding:.5rem .8rem}
pre{overflow-x:auto}
pre.cmd{background:#1c2430;color:#e8ecf2;border-radius:6px;padding:.55rem .8rem;
font-size:.78rem;line-height:1.5;font-family:ui-monospace,Menlo,Consolas,monospace}
.tablewrap{overflow-x:auto}
table{border-collapse:collapse;width:100%;font-size:.88rem;background:#fff}
td,th{border:1px solid var(--line);padding:.4rem .6rem;text-align:left;vertical-align:top}
th{background:var(--accentbg)}
ul.diag{margin:.4rem 0 0;padding-left:1.2rem}
ul.diag li{font-size:.85rem;margin:.2rem 0}
dl.gloss dt{font-weight:700;margin-top:.8rem}
dl.gloss dd{margin:.15rem 0 0 0;font-size:.88rem;color:var(--ink)}
textarea{width:100%;min-height:7.5rem;font-family:ui-monospace,monospace;font-size:.8rem;
border:1px solid var(--line);border-radius:6px;padding:.5rem}
button{background:var(--accent);color:#fff;border:0;border-radius:6px;
padding:.5rem 1.1rem;font-size:.9rem;cursor:pointer}
.btnlink{display:inline-block;background:var(--accent);color:#fff;border-radius:6px;
padding:.45rem 1rem;font-size:.88rem;text-decoration:none}
.muted{color:var(--ink2);font-size:.85rem}
.mono{font-family:ui-monospace,monospace}
.breakany{overflow-wrap:anywhere}
/* bridge */
.strip{display:flex;gap:.45rem;flex-wrap:wrap;margin:.8rem 0}
.chip{border:1px solid var(--line);background:#fff;border-radius:8px;
padding:.3rem .7rem;font-size:.82rem}
.chip b{font-weight:700}
.crew{display:grid;grid-template-columns:repeat(auto-fill,minmax(17rem,1fr));
gap:.7rem;margin:.7rem 0}
.stationcard{background:#fff;border:1px solid var(--line);border-radius:8px;
padding:.8rem .95rem;border-left:4px solid var(--role,#3b4d8f);
display:flex;flex-direction:column;gap:.35rem}
.stationcard .q{color:var(--ink2);font-size:.82rem;font-style:italic}
.stationcard .live{font-size:.8rem}
.stationcard .take{margin-top:auto;font-size:.84rem;font-weight:600;
color:var(--accent);text-decoration:none}
.monogram{display:inline-block;min-width:1.9rem;height:1.9rem;line-height:1.9rem;
text-align:center;border-radius:6px;font-weight:800;font-size:.8rem;
background:var(--roletint,#eef0f7);color:var(--role,#3b4d8f)}
.rolehead{display:flex;align-items:center;gap:.6rem;margin:.9rem 0 .2rem;
padding:.7rem .9rem;background:#fff;border:1px solid var(--line);
border-radius:8px;border-left:5px solid var(--role,#3b4d8f)}
.rolehead h2{margin:0;font-size:1.15rem}
.rolehead .q{color:var(--ink2);font-size:.85rem;font-style:italic}
.duty{margin:.55rem 0 .9rem}
.duty .why{font-size:.85rem;color:var(--ink2);margin:.15rem 0 .3rem}
.never li{margin:.25rem 0;font-size:.88rem}
.hand{display:grid;grid-template-columns:1fr 1fr;gap:.7rem}
@media(max-width:40rem){.hand{grid-template-columns:1fr}}
.hand .hcol{background:var(--accentbg);border-radius:6px;padding:.5rem .8rem;
font-size:.86rem}
.hand .hcol b{display:block;margin-bottom:.2rem;font-size:.78rem;
letter-spacing:.05em}
"""
def now_utc() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def esc(value: Any) -> str:
return html.escape(str(value))
def help_link(anchor: str) -> str:
"""A small ? that jumps to the glossary entry for a term."""
return (f'<a class="help" href="/guide#{anchor}" '
f'title="what does this mean? — explained in the guide">?</a>')
def explain(body: str) -> str:
"""The per-panel interpretation aid: always present, opt-in detail."""
return (f'<details class="explain"><summary>How to read this panel</summary>'
f'<div class="expl">{body}</div></details>')
def provenance(via: str) -> str:
return (f'<div class="prov">recomputed {esc(now_utc())} via <code>{esc(via)}</code>'
f' — nothing on this panel is cached or asserted.</div>')
def failed_panel(what: str, via: str, error: Exception) -> str:
return (
f'<div class="panel bad"><span class="pill bad">FAILED TO VERIFY</span> '
f"<strong>{esc(what)}</strong> could not be recomputed: "
f"<code>{esc(f'{type(error).__name__}: {error}')}</code>. "
f"A cockpit that cannot verify shows red, never a stale green. "
f"<span class='muted'>What to do: check that the wallet directory still exists and is "
f"readable, then reload. If this persists, inspect from the command line with "
f"<code>pacta wallet status</code>.</span>"
f"{provenance(via)}</div>"
)
def cmd_block(command: str) -> str:
"""A copy-paste command block: the no-AI drill in executable form."""
return f'<pre class="cmd">{esc(command)}</pre>'

514
src/pacta/walletui.py
View file

@ -1,153 +1,68 @@
"""walletui - the warden custody cockpit (local, read-only). """walletui - the warden custody cockpit (local, read-only): a bridge
with six role stations over shared evidence instruments.
A localhost web surface over an existing wallet directory, for the human Information architecture (control-room style, four levels):
operator who ultimately answers for the money. Six views: posture, the 1. BRIDGE (/) - whole-system verdict + the crew of six roles
pending-signature queue (airgap outbox), the incident & refusal browser, 2. STATIONS (/station/*) - one console per role: mission, duties as
a receipt inspector, the estate map, and a plain-language guide. runnable commands, live instruments, the
never-list, explicit handoffs
3. INSTRUMENTS - shared evidence views: posture, queue,
incidents, inspect, estate, guide
4. RAW - the wallet files and the CLI themselves
Design law: THE COCKPIT RENDERS EVIDENCE, IT NEVER ASSERTS IT. Every Three laws, each enforced by tests:
DESIGN LAW - THE COCKPIT RENDERS EVIDENCE, IT NEVER ASSERTS IT. Every
panel is recomputed from wallet state or submitted artifacts at request panel is recomputed from wallet state or submitted artifacts at request
time by the same functions the wallet itself uses, and every panel names time by the same functions the wallet itself uses, and every panel names
the function and timestamp that produced it. Anything that cannot be the function and timestamp that produced it. Anything that cannot be
recomputed renders as a loud FAILED-TO-VERIFY panel - there is no cached recomputed renders as a loud FAILED-TO-VERIFY panel - no cached green,
green and no neutral gray. no neutral gray.
UX law (the design law's twin): THE COCKPIT NEVER LEAVES A HUMAN IN THE UX LAW - THE COCKPIT NEVER LEAVES A HUMAN IN THE DARK. Every page opens
DARK. Every page opens with a plain-language statement of what it shows; with a plain-language lead; every verdict is stated in words; every
every verdict is stated in words, not just color; every panel carries a panel carries a "how to read this" explainer; every jargon term links to
"how to read this" explainer; every piece of jargon links to the /guide the /guide glossary.
glossary. A person who has never heard of warden must be able to read
every screen. CREW LAW - ROLES ARE DISTINCT AND COOPERATE THROUGH HANDOFFS. The bridge
presents everything a human crew would need if no AI were around: six
stations (proposer, quorum bench, operator, cryptographer, architect,
newcomer), each with runnable duties and a "this station never" list.
Separation of duties is a custody control; the stations do not melt into
each other. (Role content lives in stations.py; liveness probes in
liveness.py; shared primitives in uikit.py.)
Read-only guarantee: this module calls only read paths (``Wallet.posture``, Read-only guarantee: this module calls only read paths (``Wallet.posture``,
``verify_ledger``, directory listings) and ``transparency.verify_receipt`` ``verify_ledger``, directory listings), ``transparency.verify_receipt`` on
on submitted artifacts (parsed in memory / temp files outside the wallet). submitted artifacts (parsed in memory / temp files outside the wallet),
It cannot approve, sign, unlatch, or modify custody state; the HTTP surface and - only when the operator explicitly presses "Probe now" - outbound
exposes no mutating route. Human approve/deny is deliberately NOT here - liveness observations (HTTP GET, git queries). It cannot approve, sign,
that would be a custody-semantics change, which belongs to a separate, unlatch, or modify custody state; the HTTP surface exposes no mutating
explicitly reviewed milestone. route. Human approve/deny is deliberately NOT here - that would be a
custody-semantics change, which belongs to a separate, explicitly
reviewed milestone. The mutating acts a human crew needs are provided as
exact CLI commands on the stations instead.
The server binds 127.0.0.1 by default and is not meant to be exposed. The server binds 127.0.0.1 by default and is not meant to be exposed.
""" """
from __future__ import annotations from __future__ import annotations
import html
import json import json
import tempfile import tempfile
import urllib.parse import urllib.parse
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path from pathlib import Path
from typing import Any, Callable from typing import Any, Callable
from .attestation import load_attestation from .attestation import load_attestation
from .liveness import collect_liveness, render_liveness
from .stations import STATION_BY_ID, STATIONS, render_bridge, render_station
from .transparency import load_receipt, verify_receipt from .transparency import load_receipt, verify_receipt
from .uikit import (STYLE, esc as _esc, explain as _explain,
failed_panel as _failed_panel, help_link as _help,
now_utc as _now, provenance as _provenance)
from .wallet import Wallet from .wallet import Wallet
_STYLE = """
:root{--ink:#1c2430;--ink2:#5a6675;--line:#dde2e9;--ok:#1e7f4f;--okbg:#e2f2e9;
--bad:#a3242c;--badbg:#fbe4e6;--warn:#a86a10;--warnbg:#fdf0da;
--accent:#3b4d8f;--accentbg:#eef0f7;--bg:#f8f9fa}
*{box-sizing:border-box}
body{font-family:system-ui,sans-serif;max-width:62rem;margin:0 auto;
padding:1.4rem 1.2rem 4rem;color:var(--ink);line-height:1.55;background:var(--bg)}
h1{font-size:1.35rem;margin:.2rem 0 0}
h2{font-size:1.05rem;margin:1.6rem 0 .5rem}
code{font-family:ui-monospace,Menlo,Consolas,monospace;background:#eef0f3;
border-radius:4px;padding:.08rem .3rem;font-size:.88em}
.sub{color:var(--ink2);font-size:.85rem;margin:.3rem 0 .6rem}
nav{margin:.7rem 0 1rem;display:flex;gap:.5rem;flex-wrap:wrap}
nav a{color:var(--accent);text-decoration:none;border:1px solid var(--line);
background:#fff;border-radius:6px;padding:.3rem .7rem;font-size:.85rem;
display:flex;flex-direction:column;line-height:1.25;min-width:7.5rem}
nav a.here{border-color:var(--accent);font-weight:600;background:var(--accentbg)}
nav a .navsub{font-size:.67rem;color:var(--ink2);font-weight:400}
.banner{background:var(--warnbg);border:1px solid var(--warn);color:var(--warn);
border-radius:6px;padding:.45rem .8rem;font-size:.82rem;font-weight:600}
.banner a{color:var(--warn)}
.lead{font-size:.92rem;margin:.9rem 0 .2rem}
.verdict{border-radius:8px;padding:.8rem 1.1rem;margin:.8rem 0;border:1px solid}
.verdict strong{font-size:1.05rem;letter-spacing:.02em}
.verdict p{margin:.3rem 0 0;font-size:.88rem;font-weight:400}
.verdict.ok{background:var(--okbg);border-color:var(--ok);color:var(--ok)}
.verdict.warn{background:var(--warnbg);border-color:var(--warn);color:var(--warn)}
.verdict.bad{background:var(--badbg);border-color:var(--bad);color:var(--bad)}
.panel{background:#fff;border:1px solid var(--line);border-radius:8px;
padding:.9rem 1.1rem;margin:.7rem 0}
.panel.bad{border-color:var(--bad);background:var(--badbg)}
.prov{color:var(--ink2);font-size:.72rem;margin-top:.6rem;border-top:1px dashed var(--line);
padding-top:.35rem}
.pill{display:inline-block;border-radius:9px;padding:.06rem .55rem;font-size:.76rem;
font-weight:700}
.pill.ok{background:var(--okbg);color:var(--ok)}
.pill.bad{background:var(--badbg);color:var(--bad)}
.pill.warn{background:var(--warnbg);color:var(--warn)}
a.help{display:inline-block;width:1.05rem;height:1.05rem;line-height:1.05rem;text-align:center;
border-radius:50%;background:var(--accentbg);color:var(--accent);font-size:.72rem;
font-weight:700;text-decoration:none;vertical-align:.15em}
details.explain{margin-top:.55rem;font-size:.82rem}
details.explain summary{cursor:pointer;color:var(--accent);font-weight:600;font-size:.78rem}
details.explain .expl{color:var(--ink2);margin:.4rem 0 0;padding:.5rem .7rem;
background:var(--accentbg);border-radius:6px}
details.explain .expl ul{margin:.3rem 0;padding-left:1.1rem}
details.explain .expl li{margin:.15rem 0}
.plain{font-size:.88rem;margin:.3rem 0 .6rem}
.empty{color:var(--ink2);font-size:.88rem;background:var(--accentbg);border-radius:6px;
padding:.5rem .8rem}
pre{overflow-x:auto}
.tablewrap{overflow-x:auto}
table{border-collapse:collapse;width:100%;font-size:.88rem;background:#fff}
td,th{border:1px solid var(--line);padding:.4rem .6rem;text-align:left;vertical-align:top}
th{background:var(--accentbg)}
ul.diag{margin:.4rem 0 0;padding-left:1.2rem}
ul.diag li{font-size:.85rem;margin:.2rem 0}
dl.gloss dt{font-weight:700;margin-top:.8rem}
dl.gloss dd{margin:.15rem 0 0 0;font-size:.88rem;color:var(--ink)}
textarea{width:100%;min-height:7.5rem;font-family:ui-monospace,monospace;font-size:.8rem;
border:1px solid var(--line);border-radius:6px;padding:.5rem}
button{background:var(--accent);color:#fff;border:0;border-radius:6px;
padding:.5rem 1.1rem;font-size:.9rem;cursor:pointer}
.muted{color:var(--ink2);font-size:.85rem}
.mono{font-family:ui-monospace,monospace}
"""
def _now() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def _esc(value: Any) -> str:
return html.escape(str(value))
def _help(anchor: str) -> str:
"""A small ? that jumps to the glossary entry for a term."""
return (f'<a class="help" href="/guide#{anchor}" '
f'title="what does this mean? — explained in the guide">?</a>')
def _explain(body: str) -> str:
"""The per-panel interpretation aid: always present, opt-in detail."""
return (f'<details class="explain"><summary>How to read this panel</summary>'
f'<div class="expl">{body}</div></details>')
def _provenance(via: str) -> str:
return f'<div class="prov">recomputed {_esc(_now())} via <code>{_esc(via)}</code> — nothing on this panel is cached or asserted.</div>'
def _failed_panel(what: str, via: str, error: Exception) -> str:
return (
f'<div class="panel bad"><span class="pill bad">FAILED TO VERIFY</span> '
f"<strong>{_esc(what)}</strong> could not be recomputed: "
f"<code>{_esc(f'{type(error).__name__}: {error}')}</code>. "
f"A cockpit that cannot verify shows red, never a stale green. "
f"<span class='muted'>What to do: check that the wallet directory still exists and is "
f"readable, then reload. If this persists, inspect from the command line with "
f"<code>pacta wallet posture</code>.</span>"
f"{_provenance(via)}</div>"
)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# collectors - read-only, one wallet function each, exceptions contained # collectors - read-only, one wallet function each, exceptions contained
@ -232,25 +147,55 @@ def inspect_receipt(attestation_text: str, receipt_text: str,
"error": f"{type(error).__name__}: {error}"} "error": f"{type(error).__name__}: {error}"}
def _collect_drift() -> dict[str, Any]:
"""The Architect's live tripwire: do the two estate renderings agree?"""
def read() -> dict[str, Any]:
from .estateview import ESTATE_HTML
estate_md = (Path(__file__).resolve().parents[2] / "ESTATE.md").read_text(
encoding="utf-8")
sentinels = ["lean-transparency-log", "ltl-accumulator-verified",
"proof-aware-crypto-tooling-agent", "verifying-crypto-with-lean",
"dalek-ed25519-verified", "pasta-pallas-verified",
"ltl.zkdefi.org", "Forgejo"]
missing = ([f"{n} (estate view)" for n in sentinels if n not in ESTATE_HTML]
+ [f"{n} (ESTATE.md)" for n in sentinels if n not in estate_md])
return {"sentinels": len(sentinels), "missing": missing,
"runtime_in_both": ("What is running" in estate_md
and "ALWAYS ON" in ESTATE_HTML)}
return collect("name-level comparison of estateview.ESTATE_HTML vs ESTATE.md", read)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# page shell - one navigation, one lead paragraph, on every view # page shell - two-row navigation (stations / instruments), lead on every view
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
_VIEWS = [ _STATION_TABS = [("/", "Bridge", "the whole system at a glance")] + [
("/", "Posture", "is custody healthy right now?"), (f"/station/{s['id']}", s["name"], sub) for s, sub in zip(STATIONS, [
"ask for signatures", "four seats, one answer each",
"liveness + latch recovery", "recompute everything",
"map = territory", "start here"])]
_INSTRUMENT_TABS = [
("/posture", "Posture", "is custody healthy right now?"),
("/queue", "Queue", "what awaits the offline signer?"), ("/queue", "Queue", "what awaits the offline signer?"),
("/incidents", "Incidents", "what has ever gone wrong?"), ("/incidents", "Incidents", "what has ever gone wrong?"),
("/inspect", "Inspect", "check a receipt yourself"), ("/inspect", "Inspect", "check a receipt yourself"),
("/estate", "Estate map", "the whole system, drawn"), ("/estate", "Estate map", "the territory, drawn"),
("/guide", "Guide", "every term, explained"), ("/guide", "Guide", "every term, explained"),
] ]
_LEADS = { _LEADS: dict[str, str] = {
"/": ('This page answers one question: <strong>is custody healthy right now?</strong> ' "/": ('This is the <strong>bridge</strong>: the whole estate at one glance, then '
'The verdict comes first, the evidence behind it below. Every panel ends with a ' 'the crew. The verdict strip is recomputed on load; each crew card opens a '
'dashed provenance line naming the exact function that just recomputed it — ' '<strong>station</strong> — one human role with its duties, commands, and '
'and every small <a class="help" href="/guide#glossary">?</a> jumps to a ' 'handoffs. If no AI were around, these six stations are how people would '
'plain-language explanation.'), 'run this system.'),
"/posture": ('This page answers one question: <strong>is custody healthy right '
'now?</strong> The verdict comes first, the evidence behind it below. '
'Every panel ends with a dashed provenance line naming the exact '
'function that just recomputed it — and every small '
'<a class="help" href="/guide#glossary">?</a> jumps to a '
'plain-language explanation.'),
"/queue": ('The wallet’s signing key can live on an <em>air-gapped</em> device — a ' "/queue": ('The wallet’s signing key can live on an <em>air-gapped</em> device — a '
'computer with no network connection. To get something signed, the wallet ' 'computer with no network connection. To get something signed, the wallet '
'parks a request file in an outbox; a human carries it to the device; the ' 'parks a request file in an outbox; a human carries it to the device; the '
@ -265,16 +210,25 @@ _LEADS = {
'own verifier on them, on your machine, without writing anything to the ' 'own verifier on them, on your machine, without writing anything to the '
'wallet. Use it to check evidence somebody handed you before trusting it.'), 'wallet. Use it to check evidence somebody handed you before trusting it.'),
"/guide": ('Plain-language explanations for everything this cockpit shows. Nothing on ' "/guide": ('Plain-language explanations for everything this cockpit shows. Nothing on '
'this page is live data — this is the manual. The other five tabs are the ' 'this page is live data — this is the manual. The Bridge and the stations '
'instruments.'), 'are the working surfaces; the instruments are the shared evidence.'),
} }
for _s in STATIONS:
_LEADS[f"/station/{_s['id']}"] = _s["lead"]
def _tabs(items: list[tuple[str, str, str]], active: str) -> str:
return "".join(
f'<a href="{href}"{" class=here" if href == active else ""}>{label}'
f'<span class="navsub">{sub}</span></a>'
for href, label, sub in items)
def _page(title: str, active: str, body: str, wallet_dir: str) -> str: def _page(title: str, active: str, body: str, wallet_dir: str) -> str:
nav = "".join( nav = (f'<div class="navrow"><span class="navtag">STATIONS</span>'
f'<a href="{href}"{" class=here" if href == active else ""}>{label}' f'{_tabs(_STATION_TABS, active)}</div>'
f'<span class="navsub">{sub}</span></a>' f'<div class="navrow"><span class="navtag">INSTRUMENTS</span>'
for href, label, sub in _VIEWS) f'{_tabs(_INSTRUMENT_TABS, active)}</div>')
demo_badge = ('<span class="pill warn" title="sealed by --demo; fake members; can sign ' demo_badge = ('<span class="pill warn" title="sealed by --demo; fake members; can sign '
'nothing real">DEMO WALLET — custody-inert</span> ' 'nothing real">DEMO WALLET — custody-inert</span> '
if "DEMO" in wallet_dir else "") if "DEMO" in wallet_dir else "")
@ -284,20 +238,22 @@ def _page(title: str, active: str, body: str, wallet_dir: str) -> str:
"<!doctype html><html><head><meta charset='utf-8'>" "<!doctype html><html><head><meta charset='utf-8'>"
"<meta name='viewport' content='width=device-width,initial-scale=1'>" "<meta name='viewport' content='width=device-width,initial-scale=1'>"
f"<title>warden cockpit — {_esc(title)}</title>" f"<title>warden cockpit — {_esc(title)}</title>"
f"<style>{_STYLE}</style></head><body>" f"<style>{STYLE}</style></head><body>"
f"<h1>warden custody cockpit {demo_badge}</h1>" f"<h1>warden custody cockpit {demo_badge}</h1>"
f"<p class='sub'>Watching wallet <code>{_esc(wallet_dir)}</code> — everything below is " f"<p class='sub'>Watching wallet <code>{_esc(wallet_dir)}</code> — everything below is "
"recomputed live from that directory each time a page loads; nothing is cached, " "recomputed live from that directory each time a page loads; nothing is cached, "
"nothing is taken on trust.</p>" "nothing is taken on trust.</p>"
"<div class='banner'>READ-ONLY. This cockpit observes and recomputes; it cannot " "<div class='banner'>READ-ONLY. This cockpit observes and recomputes; it cannot "
"approve, sign, unlatch, or modify custody state. First time here? Start with the " "approve, sign, unlatch, or modify custody state — the stations give you the "
"<a href='/guide'>Guide</a> — every term on these pages is explained there.</div>" "exact commands for every act instead. First time here? Start with the "
f"<nav>{nav}</nav>{lead_html}{body}</body></html>" "<a href='/guide'>Guide</a> or the <a href='/station/newcomer'>Newcomer "
"station</a>.</div>"
f"{nav}{lead_html}{body}</body></html>"
) )
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# renderers - pure string builders over collector output # posture instrument - verdict banner + five evidence panels
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
def _posture_verdict(p: dict[str, Any]) -> str: def _posture_verdict(p: dict[str, Any]) -> str:
@ -321,25 +277,10 @@ def _posture_verdict(p: dict[str, Any]) -> str:
"is re-checked — not remembered — in the panels below.</p></div>") "is re-checked — not remembered — in the panels below.</p></div>")
def render_posture(posture: dict[str, Any]) -> str: def _panel_latch(p: dict[str, Any]) -> str:
if not posture["ok"]:
return _failed_panel("Custody posture", posture["via"], posture["error"])
p = posture["data"]
latch = p["latch"] latch = p["latch"]
ledger = p["ledger"]
latch_pill = ('<span class="pill bad">LATCHED — outbound custody frozen</span>' latch_pill = ('<span class="pill bad">LATCHED — outbound custody frozen</span>'
if latch.get("latched") else '<span class="pill ok">unlatched</span>') if latch.get("latched") else '<span class="pill ok">unlatched</span>')
chain_pill = ('<span class="pill ok">chain verified</span>' if ledger["chain_ok"]
else '<span class="pill bad">CHAIN BROKEN</span>')
members = "".join(
f"<tr><td><code>{_esc(m['backend'])}</code></td>"
f"<td class='mono'>{_esc(m['component'])}</td>"
f"<td>{_esc(m['risk_tier'])}</td>"
f"<td class='mono'>{_esc(m['source_commit'][:12])}…</td>"
f"<td class='mono'>{_esc(m['binary_sha256'][:16])}…</td></tr>"
for m in p["members"])
problems = "".join(f"<li>{_esc(x)}</li>" for x in ledger["problems"]) or (
"<li>none — every link in the chain held</li>")
latch_detail = "" latch_detail = ""
if latch.get("latched"): if latch.get("latched"):
latch_detail = (f"<p class='plain'>Trigger: <code>{_esc(latch.get('reason'))}</code> · " latch_detail = (f"<p class='plain'>Trigger: <code>{_esc(latch.get('reason'))}</code> · "
@ -347,19 +288,8 @@ def render_posture(posture: dict[str, Any]) -> str:
f"frozen since {_esc(latch.get('at'))}. Read the incident in the " f"frozen since {_esc(latch.get('at'))}. Read the incident in the "
f"<a href='/incidents'>incident browser</a>, then follow " f"<a href='/incidents'>incident browser</a>, then follow "
f"<code>docs/runbook-latch.md</code> to recover.</p>") f"<code>docs/runbook-latch.md</code> to recover.</p>")
spending = p.get("spending_policy") or {}
no_rules = (not spending) or ("note" in spending and len(spending) == 1)
spending_note = (
"<p class='empty'>No spending rules are configured for this wallet: beyond the "
"quorum gate and the latch, outbound signing is unrestricted — the wallet's own "
"words below say so. A real deployment would define limits and allowlists in "
"<code>policy.json</code>.</p>" if no_rules else
"<p class='plain'>These rules are enforced by the signing firewall before anything "
"is signed. Shown verbatim from <code>policy.json</code>.</p>")
return ( return (
_posture_verdict(p) f"<div class='panel'><h2 style='margin-top:0'>Custody latch {_help('latch')} {latch_pill}</h2>"
# --- latch ---------------------------------------------------------
+ f"<div class='panel'><h2 style='margin-top:0'>Custody latch {_help('latch')} {latch_pill}</h2>"
"<p class='plain'>The latch is the wallet's emergency brake. It trips when the " "<p class='plain'>The latch is the wallet's emergency brake. It trips when the "
"quorum disagrees or tampering is suspected, and freezes all outbound signing " "quorum disagrees or tampering is suspected, and freezes all outbound signing "
"until an operator clears it through the wallet's own channels — never from " "until an operator clears it through the wallet's own channels — never from "
@ -371,8 +301,16 @@ def render_posture(posture: dict[str, Any]) -> str:
"<li><span class='pill bad'>LATCHED</span> — the brake is on; every signing " "<li><span class='pill bad'>LATCHED</span> — the brake is on; every signing "
"request is refused with a receipt until a human resolves the trigger. " "request is refused with a receipt until a human resolves the trigger. "
"Recovery steps live in <code>docs/runbook-latch.md</code>.</li></ul>") "Recovery steps live in <code>docs/runbook-latch.md</code>.</li></ul>")
+ f"{_provenance('Wallet.latch_state()')}</div>" + f"{_provenance('Wallet.latch_state()')}</div>")
# --- ledger --------------------------------------------------------
def _panel_ledger(p: dict[str, Any]) -> str:
ledger = p["ledger"]
chain_pill = ('<span class="pill ok">chain verified</span>' if ledger["chain_ok"]
else '<span class="pill bad">CHAIN BROKEN</span>')
problems = "".join(f"<li>{_esc(x)}</li>" for x in ledger["problems"]) or (
"<li>none — every link in the chain held</li>")
return (
f"<div class='panel'><h2 style='margin-top:0'>Ledger — has history been tampered with? {_help('ledger')} {chain_pill}</h2>" f"<div class='panel'><h2 style='margin-top:0'>Ledger — has history been tampered with? {_help('ledger')} {chain_pill}</h2>"
"<p class='plain'>The ledger is the wallet's append-only journal: every custody " "<p class='plain'>The ledger is the wallet's append-only journal: every custody "
"event, in order, each entry carrying the hash of the one before it. Editing, " "event, in order, each entry carrying the hash of the one before it. Editing, "
@ -387,8 +325,18 @@ def render_posture(posture: dict[str, Any]) -> str:
"<li><span class='pill bad'>CHAIN BROKEN</span> — at least one link failed: " "<li><span class='pill bad'>CHAIN BROKEN</span> — at least one link failed: "
"history was altered, truncated, or corrupted. The list above names the first " "history was altered, truncated, or corrupted. The list above names the first "
"entry that failed.</li></ul>") "entry that failed.</li></ul>")
+ f"{_provenance('Wallet.verify_ledger() — full hash-chain recomputation')}</div>" + f"{_provenance('Wallet.verify_ledger() — full hash-chain recomputation')}</div>")
# --- quorum --------------------------------------------------------
def _panel_quorum(p: dict[str, Any]) -> str:
members = "".join(
f"<tr><td><code>{_esc(m['backend'])}</code></td>"
f"<td class='mono'>{_esc(m['component'])}</td>"
f"<td>{_esc(m['risk_tier'])}</td>"
f"<td class='mono'>{_esc(m['source_commit'][:12])}…</td>"
f"<td class='mono'>{_esc(m['binary_sha256'][:16])}…</td></tr>"
for m in p["members"])
return (
f"<div class='panel'><h2 style='margin-top:0'>Quorum — who must agree before anything is trusted? {_help('member')} " f"<div class='panel'><h2 style='margin-top:0'>Quorum — who must agree before anything is trusted? {_help('member')} "
f"<span class='pill ok'>{len(p['members'])} pinned</span></h2>" f"<span class='pill ok'>{len(p['members'])} pinned</span></h2>"
f"<p class='plain'>These are the {len(p['members'])} verifier programs this wallet " f"<p class='plain'>These are the {len(p['members'])} verifier programs this wallet "
@ -418,22 +366,48 @@ def render_posture(posture: dict[str, Any]) -> str:
"ledger's first entry so it cannot be quietly swapped. What this table does NOT " "ledger's first entry so it cannot be quietly swapped. What this table does NOT "
"prove: that the binaries correspond to the attested sources (reproducible builds " "prove: that the binaries correspond to the attested sources (reproducible builds "
"are out of scope — that gap is grade R5 — stated in the paper and the claim cards)." "are out of scope — that gap is grade R5 — stated in the paper and the claim cards)."
f"{_provenance('Wallet.capsule() / Wallet.posture()')}</div>" f"{_provenance('Wallet.capsule() / Wallet.posture()')}</div>")
# --- signing rules -------------------------------------------------
def _panel_policy(p: dict[str, Any]) -> str:
spending = p.get("spending_policy") or {}
no_rules = (not spending) or ("note" in spending and len(spending) == 1)
spending_note = (
"<p class='empty'>No spending rules are configured for this wallet: beyond the "
"quorum gate and the latch, outbound signing is unrestricted — the wallet's own "
"words below say so. A real deployment would define limits and allowlists in "
"<code>policy.json</code>.</p>" if no_rules else
"<p class='plain'>These rules are enforced by the signing firewall before anything "
"is signed. Shown verbatim from <code>policy.json</code>.</p>")
return (
f"<div class='panel'><h2 style='margin-top:0'>Signing rules (spending policy)</h2>" f"<div class='panel'><h2 style='margin-top:0'>Signing rules (spending policy)</h2>"
f"{spending_note}" f"{spending_note}"
f"<pre style='margin:0;font-size:.8rem'>{_esc(json.dumps(spending, indent=2, sort_keys=True))}</pre>" f"<pre style='margin:0;font-size:.8rem'>{_esc(json.dumps(spending, indent=2, sort_keys=True))}</pre>"
f"{_provenance('Wallet.policy() (policy.json, verbatim)')}</div>" f"{_provenance('Wallet.policy() (policy.json, verbatim)')}</div>")
# --- recorded history ----------------------------------------------
def _panel_history(p: dict[str, Any]) -> str:
return (
f"<div class='panel'><h2 style='margin-top:0'>Recorded history</h2>" f"<div class='panel'><h2 style='margin-top:0'>Recorded history</h2>"
f"<p class='plain'>Incidents on file: <strong>{p['incidents']}</strong> · refusal " f"<p class='plain'>Incidents on file: <strong>{p['incidents']}</strong> · refusal "
f"receipts on file: <strong>{p['refusal_receipts']}</strong> — read every one, " f"receipts on file: <strong>{p['refusal_receipts']}</strong> — read every one, "
"verbatim, under <a href='/incidents'>Incidents</a>. An incident is the wallet " "verbatim, under <a href='/incidents'>Incidents</a>. An incident is the wallet "
"noticing something wrong; a refusal receipt is the wallet saying no, in writing.</p>" "noticing something wrong; a refusal receipt is the wallet saying no, in writing.</p>"
f"{_provenance('directory counts, recomputed')}</div>" f"{_provenance('directory counts, recomputed')}</div>")
)
def render_posture(posture: dict[str, Any]) -> str:
if not posture["ok"]:
return _failed_panel("Custody posture", posture["via"], posture["error"])
p = posture["data"]
return (_posture_verdict(p) + _panel_latch(p) + _panel_ledger(p)
+ _panel_quorum(p) + _panel_policy(p) + _panel_history(p))
# ---------------------------------------------------------------------------
# queue / incidents / inspect instruments
# ---------------------------------------------------------------------------
def render_queue(airgap: dict[str, Any]) -> str: def render_queue(airgap: dict[str, Any]) -> str:
if not airgap["ok"]: if not airgap["ok"]:
return _failed_panel("Signature queue", airgap["via"], airgap["error"]) return _failed_panel("Signature queue", airgap["via"], airgap["error"])
@ -468,7 +442,7 @@ def render_queue(airgap: dict[str, Any]) -> str:
def render_incidents(incidents: dict[str, Any], refusals: dict[str, Any]) -> str: def render_incidents(incidents: dict[str, Any], refusals: dict[str, Any]) -> str:
def block(title: str, coll: dict[str, Any], intro: str, empty: str, def block(title: str, coll: dict[str, Any], intro: str, empty: str,
explain: str, via_note: str) -> str: explain_body: str, via_note: str) -> str:
if not coll["ok"]: if not coll["ok"]:
return _failed_panel(title, coll["via"], coll["error"]) return _failed_panel(title, coll["via"], coll["error"])
items = coll["data"] items = coll["data"]
@ -482,7 +456,7 @@ def render_incidents(incidents: dict[str, Any], refusals: dict[str, Any]) -> str
f"<pre style='font-size:.76rem;overflow-x:auto'>{_esc(json.dumps({k: v for k, v in i.items() if k != '_file'}, indent=2, sort_keys=True))}</pre></div>" f"<pre style='font-size:.76rem;overflow-x:auto'>{_esc(json.dumps({k: v for k, v in i.items() if k != '_file'}, indent=2, sort_keys=True))}</pre></div>"
for i in items[:50]) for i in items[:50])
return (f"<div class='panel'><h2 style='margin-top:0'>{title}</h2>" return (f"<div class='panel'><h2 style='margin-top:0'>{title}</h2>"
f"<p class='plain'>{intro}</p>{body}{_explain(explain)}" f"<p class='plain'>{intro}</p>{body}{_explain(explain_body)}"
f"{_provenance(via_note)}</div>") f"{_provenance(via_note)}</div>")
return ( return (
block(f"Incidents — what the wallet noticed {_help('incident')}", incidents, block(f"Incidents — what the wallet noticed {_help('incident')}", incidents,
@ -575,6 +549,89 @@ def render_inspect(result: dict[str, Any] | None,
) )
# ---------------------------------------------------------------------------
# bridge assembly - verdict strip + live crew snippets
# ---------------------------------------------------------------------------
def _bridge_strip(posture: dict[str, Any], airgap: dict[str, Any]) -> str:
if not posture["ok"]:
return _failed_panel("Custody verdict", posture["via"], posture["error"])
p = posture["data"]
ledger = p["ledger"]
chain = ('<span class="pill ok">chain verified</span>' if ledger["chain_ok"]
else '<span class="pill bad">CHAIN BROKEN</span>')
pending: Any = "?"
if airgap["ok"]:
pending = sum(1 for r in airgap["data"] if not r["answered"])
return (
_posture_verdict(p)
+ "<div class='strip'>"
f"<span class='chip'>quorum <b>{len(p['members'])} pinned</b></span>"
f"<span class='chip'>ledger {chain}</span>"
f"<span class='chip'>incidents <b>{p['incidents']}</b> · refusals "
f"<b>{p['refusal_receipts']}</b></span>"
f"<span class='chip'>queue <b>{pending} awaiting device</b></span>"
"<span class='chip'>liveness — <a href='/station/operator?probe=1'>probe from "
"the Operator station</a></span>"
"</div>"
+ _provenance("Wallet.posture() + airgap listing (liveness only on demand)")
)
def _bridge_live(posture: dict[str, Any], airgap: dict[str, Any]) -> dict[str, str]:
live: dict[str, str] = {
"cryptographer": "<div class='live muted'>instrument ready: Inspect</div>",
"architect": "<div class='live muted'>estate view + drift tripwire ready</div>",
"newcomer": "<div class='live muted'>start: the Guide, then the demo</div>",
}
if posture["ok"]:
p = posture["data"]
seats = ", ".join(f"<span class='mono'>{_esc(m['component'])}</span>"
for m in p["members"])
latch_word = ("LATCHED" if p["latch"].get("latched") else "unlatched")
live["quorum"] = f"<div class='live'>seats: {seats}</div>"
live["operator"] = (f"<div class='live'>incidents on file: <b>{p['incidents']}</b>"
f" · latch: <b>{latch_word}</b></div>")
if airgap["ok"]:
pending = sum(1 for r in airgap["data"] if not r["answered"])
live["proposer"] = f"<div class='live'>queue: <b>{pending}</b> awaiting device</div>"
return live
def _render_drift_panel() -> str:
coll = _collect_drift()
if not coll["ok"]:
return _failed_panel("Estate drift tripwire", coll["via"], coll["error"])
d = coll["data"]
if d["missing"] or not d["runtime_in_both"]:
missing = "".join(f"<li><code>{_esc(x)}</code></li>" for x in d["missing"]) or ""
runtime = ("" if d["runtime_in_both"] else
"<li>the runtime dimension is missing from one rendering</li>")
body = (f"<p class='plain'><span class='pill bad'>DRIFT</span> the two renderings "
f"of the estate disagree:</p><ul class='diag'>{missing}{runtime}</ul>")
else:
body = (f"<p class='plain'><span class='pill ok'>renderings agree</span> all "
f"{d['sentinels']} sentinel names present in both <code>ESTATE.md</code> "
"and the cockpit estate view, and both carry the runtime dimension.</p>")
return (
"<div class='panel'><h3 style='margin-top:0'>Drift tripwire — map vs map</h3>"
"<p class='plain'>The estate map exists twice: the committed "
"<code>ESTATE.md</code> and the <a href='/estate'>estate view</a>. Two "
"renderings of one model need a tripwire — this panel compares them live, "
"name by name.</p>"
+ body
+ _explain(
"<ul><li>The comparison is name-level (sentinel entities + the runtime "
"dimension) — the same check the test suite runs.</li>"
"<li>On DRIFT: fix the stale rendering AND check its generator — a "
"published file that drifted from its source once will drift again.</li></ul>")
+ _provenance(coll["via"]) + "</div>")
# ---------------------------------------------------------------------------
# guide instrument - the manual
# ---------------------------------------------------------------------------
def render_guide() -> str: def render_guide() -> str:
"""The manual: static plain-language explanations, no live data.""" """The manual: static plain-language explanations, no live data."""
return ( return (
@ -589,12 +646,27 @@ def render_guide() -> str:
# --- what is the cockpit ------------------------------------------- # --- what is the cockpit -------------------------------------------
"<div class='panel'><h2 style='margin-top:0' id='cockpit'>What is this cockpit?</h2>" "<div class='panel'><h2 style='margin-top:0' id='cockpit'>What is this cockpit?</h2>"
"<p class='plain'>A local, read-only window onto one wallet directory, for the " "<p class='plain'>A local, read-only window onto one wallet directory, for the "
"human who ultimately answers for the money. Its design law: <strong>the cockpit " "humans who ultimately answer for the money. Its design law: <strong>the cockpit "
"renders evidence, it never asserts it</strong>. Every page is recomputed from " "renders evidence, it never asserts it</strong>. Every page is recomputed from "
"the wallet's files at the moment you load it, by the same functions the wallet " "the wallet's files at the moment you load it, by the same functions the wallet "
"itself uses. It cannot approve, sign, unlatch, or change anything — the server " "itself uses. It cannot approve, sign, unlatch, or change anything — the server "
"has no writing routes, and the test suite proves a full click-through changes " "has no writing routes, and the test suite proves a full click-through changes "
"not one byte of wallet state.</p></div>" "not one byte of wallet state. What it does provide is the <em>work</em>: the "
"<a href='/'>Bridge</a> organizes everything a human crew would do if no AI were "
"around, as six role stations with runnable commands.</p></div>"
# --- the crew ------------------------------------------------------
"<div class='panel'><h2 style='margin-top:0' id='crew'>The crew model</h2>"
"<p class='plain'>Six roles run this estate: the <strong>Proposer</strong> asks "
"for signatures; the <strong>Quorum bench</strong> holds four independent "
"verifier seats; the <strong>Operator</strong> watches liveness and owns latch "
"recovery; the <strong>Cryptographer</strong> recomputes every piece of "
"evidence; the <strong>Architect</strong> keeps the estate map true; the "
"<strong>Newcomer</strong> learns — and supplies fresh eyes. Roles cooperate "
"through explicit handoffs and never blur: the one who proposes never approves, "
"the one who verifies never proposes, the one who watches never overrides the "
"bench. One person (or one agent) may hold several stations — explicitly, one "
"at a time, with the handoffs still applying. Each station page states its "
"mission, duties as commands, and what it <em>never</em> does.</p></div>"
# --- how to read --------------------------------------------------- # --- how to read ---------------------------------------------------
"<div class='panel'><h2 style='margin-top:0' id='reading'>How to read any page here</h2>" "<div class='panel'><h2 style='margin-top:0' id='reading'>How to read any page here</h2>"
"<ol>" "<ol>"
@ -619,8 +691,10 @@ def render_guide() -> str:
# --- tour ---------------------------------------------------------- # --- tour ----------------------------------------------------------
"<div class='panel'><h2 style='margin-top:0' id='tour'>A five-minute tour</h2>" "<div class='panel'><h2 style='margin-top:0' id='tour'>A five-minute tour</h2>"
"<ol>" "<ol>"
"<li>Open <a href='/'>Posture</a> — read the verdict banner, then the panels top " "<li>Open the <a href='/'>Bridge</a> — the verdict strip is the whole system in "
"to bottom: latch, ledger, quorum, signing rules, recorded history.</li>" "one line; the crew cards are who does what.</li>"
"<li>Open <a href='/posture'>Posture</a> — read the verdict banner, then the "
"panels top to bottom: latch, ledger, quorum, signing rules, recorded history.</li>"
"<li>Open <a href='/incidents'>Incidents</a> — on a healthy wallet both lists are " "<li>Open <a href='/incidents'>Incidents</a> — on a healthy wallet both lists are "
"empty, and the page says why that is the good state.</li>" "empty, and the page says why that is the good state.</li>"
"<li>Open <a href='/queue'>Queue</a> — empty unless a signature is waiting for " "<li>Open <a href='/queue'>Queue</a> — empty unless a signature is waiting for "
@ -629,6 +703,8 @@ def render_guide() -> str:
"<code>examples/wallet-evidence/</code> and watch the deployed verifier run.</li>" "<code>examples/wallet-evidence/</code> and watch the deployed verifier run.</li>"
"<li>Open the <a href='/estate'>Estate map</a> — where this wallet sits in the " "<li>Open the <a href='/estate'>Estate map</a> — where this wallet sits in the "
"wider verified-crypto estate, and what is actually running where.</li>" "wider verified-crypto estate, and what is actually running where.</li>"
"<li>Then take the <a href='/station/newcomer'>Newcomer station</a> — your "
"first hour, mapped out.</li>"
"</ol></div>" "</ol></div>"
# --- glossary ------------------------------------------------------ # --- glossary ------------------------------------------------------
"<div class='panel'><h2 style='margin-top:0' id='glossary'>Glossary</h2>" "<div class='panel'><h2 style='margin-top:0' id='glossary'>Glossary</h2>"
@ -689,6 +765,11 @@ def render_guide() -> str:
"<dd>The dashed footer on every panel, naming the exact function that recomputed " "<dd>The dashed footer on every panel, naming the exact function that recomputed "
"the panel and when. It is the cockpit's signature move: evidence of freshness " "the panel and when. It is the cockpit's signature move: evidence of freshness "
"attached to every claim.</dd>" "attached to every claim.</dd>"
"<dt id='station'>station</dt>"
"<dd>One human role's console on the Bridge: mission, duties as runnable "
"commands, live instruments, the never-list (separation of duties), and "
"handoffs. Six stations: Proposer, Quorum bench, Operator, Cryptographer, "
"Architect, Newcomer.</dd>"
"<dt id='demo'>DEMO wallet</dt>" "<dt id='demo'>DEMO wallet</dt>"
"<dd>A throwaway wallet sealed by <code>pacta wallet cockpit --demo</code> so you " "<dd>A throwaway wallet sealed by <code>pacta wallet cockpit --demo</code> so you "
"can explore this cockpit before creating a real wallet. Its members are fake " "can explore this cockpit before creating a real wallet. Its members are fake "
@ -703,6 +784,8 @@ def render_guide() -> str:
"and the claim cards rather than hidden.</li>" "and the claim cards rather than hidden.</li>"
"<li>Whether the machine this cockpit runs on is itself clean — a compromised " "<li>Whether the machine this cockpit runs on is itself clean — a compromised "
"operating system can lie to any dashboard, including this one.</li>" "operating system can lie to any dashboard, including this one.</li>"
"<li>Whether a live service is <em>honest</em> — the liveness board checks "
"pulses, not truth; truth is the Cryptographer's replay work.</li>"
"<li>Anything it could not recompute just now — that renders as a red FAILED TO " "<li>Anything it could not recompute just now — that renders as a red FAILED TO "
"VERIFY panel, never as a guess and never as a stale green.</li>" "VERIFY panel, never as a guess and never as a stale green.</li>"
"</ul></div>" "</ul></div>"
@ -812,13 +895,45 @@ def make_handler(wallet_dir: Path):
def _wallet(self) -> Wallet: def _wallet(self) -> Wallet:
return Wallet(wallet_dir) return Wallet(wallet_dir)
def _station_embeds(self, station_id: str, probe: bool) -> list[str]:
wallet = self._wallet()
if station_id == "proposer":
return [render_queue(collect_airgap(wallet))]
if station_id == "quorum":
posture = collect("Wallet.posture()", wallet.posture)
return ([_panel_quorum(posture["data"])] if posture["ok"] else
[_failed_panel("Quorum bench roster", posture["via"],
posture["error"])])
if station_id == "operator":
live = render_liveness(collect_liveness() if probe else None)
posture = collect("Wallet.posture()", wallet.posture)
if posture["ok"]:
p = posture["data"]
return [live, _panel_latch(p), _panel_history(p)]
return [live, _failed_panel("Custody posture", posture["via"],
posture["error"])]
if station_id == "cryptographer":
return [render_inspect(None)]
if station_id == "architect":
return [_render_drift_panel()]
return []
def do_GET(self) -> None: # noqa: N802 - http.server API def do_GET(self) -> None: # noqa: N802 - http.server API
route = urllib.parse.urlparse(self.path).path parsed = urllib.parse.urlparse(self.path)
route = parsed.path
query = urllib.parse.parse_qs(parsed.query)
wd = str(wallet_dir) wd = str(wallet_dir)
if route == "/": if route == "/":
wallet = self._wallet()
posture = collect("Wallet.posture()", wallet.posture)
airgap = collect_airgap(wallet)
body = render_bridge(_bridge_strip(posture, airgap),
_bridge_live(posture, airgap))
self._send(_page("bridge", "/", body, wd))
elif route == "/posture":
wallet = self._wallet() wallet = self._wallet()
body = render_posture(collect("Wallet.posture()", wallet.posture)) body = render_posture(collect("Wallet.posture()", wallet.posture))
self._send(_page("posture", "/", body, wd)) self._send(_page("posture", "/posture", body, wd))
elif route == "/queue": elif route == "/queue":
body = render_queue(collect_airgap(self._wallet())) body = render_queue(collect_airgap(self._wallet()))
self._send(_page("signature queue", "/queue", body, wd)) self._send(_page("signature queue", "/queue", body, wd))
@ -833,6 +948,19 @@ def make_handler(wallet_dir: Path):
elif route == "/estate": elif route == "/estate":
from .estateview import ESTATE_HTML from .estateview import ESTATE_HTML
self._send(ESTATE_HTML + _ESTATE_BACK_CHIP) self._send(ESTATE_HTML + _ESTATE_BACK_CHIP)
elif route.startswith("/station/"):
station_id = route.removeprefix("/station/")
station = STATION_BY_ID.get(station_id)
if station is None:
self._send(_page("not found", "",
"<div class='panel bad'>No such station. The "
"STATIONS row above lists the whole crew.</div>",
wd), 404)
return
probe = query.get("probe", ["0"])[0] == "1"
body = render_station(station,
self._station_embeds(station_id, probe))
self._send(_page(f"{station['name']} station", route, body, wd))
else: else:
self._send(_page("not found", "", self._send(_page("not found", "",
"<div class='panel bad'>No such view. The tabs above list " "<div class='panel bad'>No such view. The tabs above list "

102
tests/test_walletui.py
View file

@ -178,7 +178,11 @@ def test_server_routes_and_read_only_guarantee(tmp_path):
thread = threading.Thread(target=server.serve_forever, daemon=True) thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start() thread.start()
try: try:
for route in ("/", "/queue", "/incidents", "/inspect", "/guide"): routes = ("/", "/posture", "/queue", "/incidents", "/inspect", "/guide",
"/station/proposer", "/station/quorum", "/station/operator",
"/station/operator?probe=1", "/station/cryptographer",
"/station/architect", "/station/newcomer")
for route in routes:
with urllib.request.urlopen(f"http://127.0.0.1:{port}{route}") as resp: with urllib.request.urlopen(f"http://127.0.0.1:{port}{route}") as resp:
body = resp.read().decode() body = resp.read().decode()
assert resp.status == 200 assert resp.status == 200
@ -293,7 +297,7 @@ def test_every_view_carries_lead_nav_and_explainers(tmp_path):
wallet = _seal_wallet(tmp_path) wallet = _seal_wallet(tmp_path)
server, thread, port = _serve(wallet.dir) server, thread, port = _serve(wallet.dir)
try: try:
for route in ("/", "/queue", "/incidents", "/inspect"): for route in ("/", "/posture", "/queue", "/incidents", "/inspect"):
status, body = _get(port, route) status, body = _get(port, route)
assert status == 200 assert status == 200
assert 'href="/guide"' in body, f"{route}: no path to the guide" assert 'href="/guide"' in body, f"{route}: no path to the guide"
@ -326,3 +330,97 @@ def test_estate_page_links_back_to_cockpit(tmp_path):
finally: finally:
server.shutdown() server.shutdown()
thread.join(timeout=5) thread.join(timeout=5)
# ---------------------------------------------------------------------------
# Crew law: the bridge presents six DISTINCT role stations that cooperate
# through explicit handoffs — everything a human crew would need with no AI
# around, as runnable commands. Roles never melt into each other.
# ---------------------------------------------------------------------------
STATION_IDS = ("proposer", "quorum", "operator", "cryptographer",
"architect", "newcomer")
def test_bridge_shows_crew_and_dispatch(tmp_path):
wallet = _seal_wallet(tmp_path)
server, thread, port = _serve(wallet.dir)
try:
status, body = _get(port, "/")
assert status == 200
assert "CUSTODY HEALTHY" in body # whole-system verdict, in words, on top
for name in ("Proposer", "Quorum bench", "Operator", "Cryptographer",
"Architect", "Newcomer"):
assert name in body, f"bridge is missing the {name} card"
assert "Take this station" in body
assert "If this happens, who acts" in body # the dispatch (andon) board
finally:
server.shutdown()
thread.join(timeout=5)
def test_every_station_defines_role_contract(tmp_path):
"""Each station page must state: mission, duties as runnable commands,
the never-list (separation of duties), and explicit handoffs."""
wallet = _seal_wallet(tmp_path)
server, thread, port = _serve(wallet.dir)
try:
for sid in STATION_IDS:
status, body = _get(port, f"/station/{sid}")
assert status == 200, f"{sid}: not served"
assert "Mission" in body, f"{sid}: no mission"
assert "Duties" in body, f"{sid}: no duties"
assert 'class="cmd"' in body, f"{sid}: no runnable command (the no-AI drill)"
assert "This station never" in body, f"{sid}: no separation-of-duties list"
assert "Handoffs" in body, f"{sid}: no handoffs"
assert "RECEIVES" in body and "DELIVERS" in body, f"{sid}: handoffs empty"
assert 'class="lead"' in body, f"{sid}: no lead"
finally:
server.shutdown()
thread.join(timeout=5)
def test_stations_are_distinct_roles(tmp_path):
"""No melting: each role's signature phrase appears on its own station
and on no other station."""
markers = {
"proposer": "front door",
"quorum": "unanimity or latch",
"operator": "Probe now",
"cryptographer": "Replay the whole log offline",
"architect": "Drift tripwire",
"newcomer": "Your first hour",
}
wallet = _seal_wallet(tmp_path)
server, thread, port = _serve(wallet.dir)
try:
bodies = {sid: _get(port, f"/station/{sid}")[1] for sid in STATION_IDS}
for sid, marker in markers.items():
assert marker in bodies[sid], f"{sid}: lost its signature duty ({marker})"
for other, body in bodies.items():
if other != sid:
assert marker not in body, (
f"{other} bleeds into {sid}'s role ({marker})")
finally:
server.shutdown()
thread.join(timeout=5)
def test_operator_probe_is_explicit_and_live(tmp_path):
"""The liveness board never phones home on an ordinary page load; probes
run only on the operator's explicit demand, then show per-target rows."""
wallet = _seal_wallet(tmp_path)
server, thread, port = _serve(wallet.dir)
try:
status, body = _get(port, "/station/operator")
assert status == 200
assert "has not probed" in body and "Probe now" in body
assert "Public services" not in body # no results without demand
status, body = _get(port, "/station/operator?probe=1")
assert status == 200
assert "Public services" in body and "Local working copies" in body
for target in ("log head", "paper", "blog", "lean-transparency-log"):
assert target in body, f"probe results missing target: {target}"
finally:
server.shutdown()
thread.join(timeout=5)