build(agent): c3po#b883b4 iteration

README.md

@@ -10,6 +10,7 @@ CivicSwarm stores proposals, comments, resident profiles, preference signals, an
 - summarizes discussion by detected language
 - routes proposal fragments to relevant residents by geography, interests, experience, and language
 - aggregates preferences with optional differential-privacy noise
+- ingests offline capture batches from mobile, SMS, and field canvassing with audit-friendly idempotency
 - exports a civic brief with provenance and audit trail
 
 ## Stack
@@ -50,10 +51,12 @@ Environment variables:
 - `POST /residents`
 - `POST /proposals/{proposal_id}/comments`
 - `POST /proposals/{proposal_id}/preferences`
+- `POST /capture-batches`
 - `GET /proposals/{proposal_id}/route`
 - `GET /proposals/{proposal_id}/dashboard`
 - `GET /proposals/{proposal_id}/brief`
 - `GET /proposals/{proposal_id}/ledger`
+- `GET /capture-batches/{batch_key}`
 
 ## Testing
 

civicswarm/api.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
+from datetime import datetime
 from pathlib import Path
+from typing import Annotated, Literal
 
 from fastapi import FastAPI, HTTPException
 from pydantic import BaseModel, Field
@@ -38,6 +40,61 @@ class PreferenceCreate(BaseModel):
     channel: str = "mobile"
 
 
+class BatchResident(BaseModel):
+    kind: Literal["resident"] = "resident"
+    resident_key: str
+    geography: str = ""
+    interests: list[str] = Field(default_factory=list)
+    lived_experience: list[str] = Field(default_factory=list)
+    languages: list[str] = Field(default_factory=list)
+
+
+class BatchComment(BaseModel):
+    kind: Literal["comment"] = "comment"
+    proposal_id: int
+    text: str
+    resident_key: str | None = None
+    channel: str = "mobile"
+    geography: str = ""
+    metadata: dict = Field(default_factory=dict)
+
+
+class BatchPreference(BaseModel):
+    kind: Literal["preference"] = "preference"
+    proposal_id: int
+    resident_key: str
+    score: float
+    channel: str = "mobile"
+
+
+class BatchLedger(BaseModel):
+    kind: Literal["ledger"] = "ledger"
+    proposal_id: int
+    entry_kind: str
+    payload: dict = Field(default_factory=dict)
+
+
+class BatchProposalVersion(BaseModel):
+    kind: Literal["proposal_version"] = "proposal_version"
+    proposal_id: int
+    rationale: str = ""
+    objections: list[str] = Field(default_factory=list)
+    resolution_status: str = "open"
+
+
+CaptureItem = Annotated[
+    BatchResident | BatchComment | BatchPreference | BatchLedger | BatchProposalVersion,
+    Field(discriminator="kind"),
+]
+
+
+class CaptureBatchCreate(BaseModel):
+    batch_key: str
+    source: str = "mobile"
+    captured_at: datetime | None = None
+    items: list[CaptureItem] = Field(default_factory=list)
+
+
 def create_app(db_path: str | Path = ":memory:") -> FastAPI:
     service = CivicSwarmService.from_path(db_path)
     app = FastAPI(title="CivicSwarm", version="0.1.0")
@@ -62,6 +119,17 @@ def create_app(db_path: str | Path = ":memory:") -> FastAPI:
     def create_preference(proposal_id: int, payload: PreferenceCreate):
         return service.submit_preference(proposal_id, payload.resident_key, payload.score, payload.channel)
 
+    @app.post("/capture-batches")
+    def ingest_capture_batch(payload: CaptureBatchCreate):
+        return service.ingest_capture_batch(payload.batch_key, payload.source, [item.model_dump() for item in payload.items], payload.captured_at)
+
+    @app.get("/capture-batches/{batch_key}")
+    def get_capture_batch(batch_key: str):
+        try:
+            return service.get_capture_batch(batch_key)
+        except KeyError as exc:
+            raise HTTPException(status_code=404, detail=str(exc)) from exc
+
     @app.get("/proposals/{proposal_id}/route")
     def route_proposal(proposal_id: int, top_n: int = 3):
         return service.route_proposal(proposal_id, top_n=top_n)

civicswarm/db.py

@@ -18,6 +18,7 @@ from sqlalchemy import (
     create_engine,
     inspect,
 )
+from sqlalchemy.engine import Engine
 
 
 def utcnow() -> datetime:
@@ -103,6 +104,21 @@ routes = Table(
     Column("created_at", DateTime(timezone=True), nullable=False, default=utcnow),
 )
 
+capture_batches = Table(
+    "capture_batches",
+    metadata,
+    Column("id", Integer, primary_key=True),
+    Column("batch_key", String, nullable=False, unique=True),
+    Column("source", String, nullable=False),
+    Column("captured_at", DateTime(timezone=True), nullable=False, default=utcnow),
+    Column("item_count", Integer, nullable=False, default=0),
+    Column("status", String, nullable=False, default="received"),
+    Column("payload_json", Text, nullable=False, default="{}"),
+    Column("results_json", Text, nullable=False, default="[]"),
+    Column("created_at", DateTime(timezone=True), nullable=False, default=utcnow),
+    Column("ingested_at", DateTime(timezone=True), nullable=True),
+)
+
 ledger_entries = Table(
     "ledger_entries",
     metadata,
@@ -116,7 +132,7 @@ ledger_entries = Table(
 
 @dataclass(frozen=True)
 class Database:
-    engine: object
+    engine: Engine
 
 
 def create_database(db_url: str):

civicswarm/service.py

@@ -23,7 +23,9 @@ from .analysis import (
     tokenize,
 )
 from .db import (
+    capture_batches,
     comments,
+    Database,
     create_database,
     ledger_entries,
     preferences,
@@ -60,6 +62,7 @@ def _normalize_terms(values: Iterable[str]) -> set[str]:
 class CivicSwarmService:
     db_url: str
     secret: str = "civicswarm"
+    db: Database | None = None
 
     def __post_init__(self) -> None:
         self.db = create_database(self.db_url)
@@ -167,29 +170,8 @@ class CivicSwarmService:
         metadata: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
         metadata = metadata or {}
-        language = detect_language(text)
         with self.db.engine.begin() as conn:
-            result = conn.execute(
+            comment_id = self._insert_comment_row(conn, proposal_id, text, channel, resident_key, geography, metadata)
-                insert(comments).values(
-                    proposal_id=proposal_id,
-                    resident_key=self._token(resident_key),
-                    channel=channel,
-                    geography=geography,
-                    language=language,
-                    text=text,
-                    metadata_json=_json(metadata),
-                    created_at=_now(),
-                )
-            )
-            comment_id = result.inserted_primary_key[0]
-            conn.execute(
-                insert(ledger_entries).values(
-                    proposal_id=proposal_id,
-                    kind="comment_received",
-                    payload_json=_json({"comment_id": comment_id, "channel": channel, "language": language}),
-                    created_at=_now(),
-                )
-            )
         return self.get_comment(comment_id)
 
     def get_comment(self, comment_id: int) -> dict[str, Any]:
@@ -201,26 +183,101 @@ class CivicSwarmService:
 
     def submit_preference(self, proposal_id: int, resident_key: str, score: float, channel: str = "mobile") -> dict[str, Any]:
         with self.db.engine.begin() as conn:
-            result = conn.execute(
+            preference_id = self._insert_preference_row(conn, proposal_id, resident_key, score, channel)
-                insert(preferences).values(
-                    proposal_id=proposal_id,
-                    resident_key=self._token(resident_key) or resident_key,
-                    score=float(score),
-                    channel=channel,
-                    created_at=_now(),
-                )
-            )
-            preference_id = result.inserted_primary_key[0]
-            conn.execute(
-                insert(ledger_entries).values(
-                    proposal_id=proposal_id,
-                    kind="preference_submitted",
-                    payload_json=_json({"preference_id": preference_id, "channel": channel}),
-                    created_at=_now(),
-                )
-            )
         return {"id": preference_id, "proposal_id": proposal_id, "resident_key": resident_key, "score": float(score), "channel": channel}
 
+    def ingest_capture_batch(self, batch_key: str, source: str, items: list[dict[str, Any]], captured_at: datetime | None = None) -> dict[str, Any]:
+        captured_at = captured_at or _now()
+        payload = {"batch_key": batch_key, "source": source, "captured_at": captured_at.isoformat(), "items": items}
+
+        with self.db.engine.begin() as conn:
+            existing = conn.execute(select(capture_batches).where(capture_batches.c.batch_key == batch_key)).mappings().first()
+            if existing is not None:
+                return self._capture_batch_dict(existing)
+
+            result = conn.execute(
+                insert(capture_batches).values(
+                    batch_key=batch_key,
+                    source=source,
+                    captured_at=captured_at,
+                    item_count=len(items),
+                    status="received",
+                    payload_json=_json(payload),
+                    results_json=_json([]),
+                    created_at=_now(),
+                    ingested_at=None,
+                )
+            )
+            batch_id = result.inserted_primary_key[0]
+
+            processed: list[dict[str, Any]] = []
+            for item in items:
+                kind = item.get("kind")
+                if kind == "resident":
+                    self._insert_resident_row(
+                        conn,
+                        resident_key=item["resident_key"],
+                        geography=item.get("geography", ""),
+                        interests=item.get("interests", []),
+                        lived_experience=item.get("lived_experience", []),
+                        languages=item.get("languages", []),
+                    )
+                    processed.append({"kind": "resident", "resident_key": item["resident_key"]})
+                elif kind == "comment":
+                    comment_id = self._insert_comment_row(
+                        conn,
+                        proposal_id=int(item["proposal_id"]),
+                        text=item["text"],
+                        channel=item.get("channel", source),
+                        resident_key=item.get("resident_key"),
+                        geography=item.get("geography", ""),
+                        metadata={**item.get("metadata", {}), "batch_key": batch_key, "source": source},
+                    )
+                    processed.append({"kind": "comment", "comment_id": comment_id, "proposal_id": int(item["proposal_id"])})
+                elif kind == "preference":
+                    preference_id = self._insert_preference_row(
+                        conn,
+                        proposal_id=int(item["proposal_id"]),
+                        resident_key=item["resident_key"],
+                        score=float(item["score"]),
+                        channel=item.get("channel", source),
+                    )
+                    processed.append({"kind": "preference", "preference_id": preference_id, "proposal_id": int(item["proposal_id"])})
+                elif kind == "ledger":
+                    ledger_id = self._insert_ledger_row(
+                        conn,
+                        proposal_id=int(item["proposal_id"]),
+                        kind=item["entry_kind"],
+                        payload={**item.get("payload", {}), "batch_key": batch_key, "source": source},
+                    )
+                    processed.append({"kind": "ledger", "ledger_id": ledger_id, "proposal_id": int(item["proposal_id"])})
+                elif kind == "proposal_version":
+                    version_info = self._record_proposal_version_row(
+                        conn,
+                        proposal_id=int(item["proposal_id"]),
+                        rationale=item.get("rationale", ""),
+                        objections=item.get("objections", []),
+                        resolution_status=item.get("resolution_status", "open"),
+                    )
+                    processed.append({"kind": "proposal_version", **version_info})
+                else:
+                    raise ValueError(f"unsupported capture kind: {kind!r}")
+
+            conn.execute(
+                update(capture_batches)
+                .where(capture_batches.c.id == batch_id)
+                .values(status="processed", results_json=_json(processed), ingested_at=_now())
+            )
+
+        return self.get_capture_batch(batch_key)
+
+    def get_capture_batch(self, batch_key: str) -> dict[str, Any]:
+        with self.db.engine.begin() as conn:
+            row = conn.execute(select(capture_batches).where(capture_batches.c.batch_key == batch_key)).mappings().first()
+        if row is None:
+            raise KeyError(f"capture batch {batch_key} not found")
+        return self._capture_batch_dict(row)
+
     def aggregate_preferences(self, proposal_id: int, epsilon: float | None = None, seed: int = 0) -> dict[str, Any]:
         with self.db.engine.begin() as conn:
             rows = conn.execute(select(preferences.c.score).where(preferences.c.proposal_id == proposal_id)).all()
@@ -302,46 +359,13 @@ class CivicSwarmService:
 
     def record_ledger_entry(self, proposal_id: int, kind: str, payload: dict[str, Any]) -> dict[str, Any]:
         with self.db.engine.begin() as conn:
-            result = conn.execute(
+            result_id = self._insert_ledger_row(conn, proposal_id, kind, payload)
-                insert(ledger_entries).values(
+        return {"id": result_id, "proposal_id": proposal_id, "kind": kind, "payload": payload}
-                    proposal_id=proposal_id,
-                    kind=kind,
-                    payload_json=_json(payload),
-                    created_at=_now(),
-                )
-            )
-        return {"id": result.inserted_primary_key[0], "proposal_id": proposal_id, "kind": kind, "payload": payload}
 
     def record_proposal_version(self, proposal_id: int, rationale: str, objections: list[str], resolution_status: str) -> dict[str, Any]:
         with self.db.engine.begin() as conn:
-            current = conn.execute(
+            version_info = self._record_proposal_version_row(conn, proposal_id, rationale, objections, resolution_status)
-                select(func.max(proposal_versions.c.version)).where(proposal_versions.c.proposal_id == proposal_id)
+        return version_info
-            ).scalar_one_or_none()
-            version = int(current or 0) + 1
-            result = conn.execute(
-                insert(proposal_versions).values(
-                    proposal_id=proposal_id,
-                    version=version,
-                    rationale=rationale,
-                    objections_json=_json(objections),
-                    resolution_status=resolution_status,
-                    created_at=_now(),
-                )
-            )
-            conn.execute(
-                update(proposals)
-                .where(proposals.c.id == proposal_id)
-                .values(version=version, updated_at=_now(), status=resolution_status)
-            )
-            conn.execute(
-                insert(ledger_entries).values(
-                    proposal_id=proposal_id,
-                    kind="proposal_version_recorded",
-                    payload_json=_json({"version": version, "resolution_status": resolution_status}),
-                    created_at=_now(),
-                )
-            )
-        return {"id": result.inserted_primary_key[0], "proposal_id": proposal_id, "version": version, "resolution_status": resolution_status}
 
     def list_proposal_versions(self, proposal_id: int) -> list[dict[str, Any]]:
         with self.db.engine.begin() as conn:
@@ -466,6 +490,139 @@ class CivicSwarmService:
         overlap = len(left_tokens & right_tokens)
         return overlap / math.sqrt(len(left_tokens) * len(right_tokens))
 
+    def _insert_resident_row(
+        self,
+        conn,
+        resident_key: str,
+        geography: str = "",
+        interests: list[str] | None = None,
+        lived_experience: list[str] | None = None,
+        languages: list[str] | None = None,
+    ) -> str:
+        interests = interests or []
+        lived_experience = lived_experience or []
+        languages = languages or []
+        existing = conn.execute(select(residents).where(residents.c.resident_key == resident_key)).mappings().first()
+        values = dict(
+            resident_key=resident_key,
+            geography=geography,
+            interests_json=_json(interests),
+            lived_experience_json=_json(lived_experience),
+            languages_json=_json(languages),
+            updated_at=_now(),
+        )
+        if existing is None:
+            values["created_at"] = _now()
+            conn.execute(insert(residents).values(**values))
+        else:
+            conn.execute(update(residents).where(residents.c.resident_key == resident_key).values(**values))
+        return resident_key
+
+    def _insert_comment_row(
+        self,
+        conn,
+        proposal_id: int,
+        text: str,
+        channel: str,
+        resident_key: str | None,
+        geography: str,
+        metadata: dict[str, Any],
+    ) -> int:
+        language = detect_language(text)
+        result = conn.execute(
+            insert(comments).values(
+                proposal_id=proposal_id,
+                resident_key=self._token(resident_key),
+                channel=channel,
+                geography=geography,
+                language=language,
+                text=text,
+                metadata_json=_json(metadata),
+                created_at=_now(),
+            )
+        )
+        comment_id = result.inserted_primary_key[0]
+        self._insert_ledger_row(
+            conn,
+            proposal_id,
+            "comment_received",
+            {"comment_id": comment_id, "channel": channel, "language": language},
+        )
+        return int(comment_id)
+
+    def _insert_preference_row(self, conn, proposal_id: int, resident_key: str, score: float, channel: str) -> int:
+        result = conn.execute(
+            insert(preferences).values(
+                proposal_id=proposal_id,
+                resident_key=self._token(resident_key) or resident_key,
+                score=float(score),
+                channel=channel,
+                created_at=_now(),
+            )
+        )
+        preference_id = int(result.inserted_primary_key[0])
+        self._insert_ledger_row(
+            conn,
+            proposal_id,
+            "preference_submitted",
+            {"preference_id": preference_id, "channel": channel},
+        )
+        return preference_id
+
+    def _insert_ledger_row(self, conn, proposal_id: int, kind: str, payload: dict[str, Any]) -> int:
+        result = conn.execute(
+            insert(ledger_entries).values(
+                proposal_id=proposal_id,
+                kind=kind,
+                payload_json=_json(payload),
+                created_at=_now(),
+            )
+        )
+        return int(result.inserted_primary_key[0])
+
+    def _record_proposal_version_row(
+        self,
+        conn,
+        proposal_id: int,
+        rationale: str,
+        objections: list[str],
+        resolution_status: str,
+    ) -> dict[str, Any]:
+        current = conn.execute(select(func.max(proposal_versions.c.version)).where(proposal_versions.c.proposal_id == proposal_id)).scalar_one_or_none()
+        version = int(current or 0) + 1
+        result = conn.execute(
+            insert(proposal_versions).values(
+                proposal_id=proposal_id,
+                version=version,
+                rationale=rationale,
+                objections_json=_json(objections),
+                resolution_status=resolution_status,
+                created_at=_now(),
+            )
+        )
+        conn.execute(update(proposals).where(proposals.c.id == proposal_id).values(version=version, updated_at=_now(), status=resolution_status))
+        self._insert_ledger_row(
+            conn,
+            proposal_id,
+            "proposal_version_recorded",
+            {"version": version, "resolution_status": resolution_status},
+        )
+        return {"id": int(result.inserted_primary_key[0]), "proposal_id": proposal_id, "version": version, "resolution_status": resolution_status}
+
+    def _capture_batch_dict(self, row: dict[str, Any]) -> dict[str, Any]:
+        return {
+            "id": row["id"],
+            "batch_key": row["batch_key"],
+            "source": row["source"],
+            "captured_at": row["captured_at"].isoformat(),
+            "item_count": row["item_count"],
+            "status": row["status"],
+            "payload": _parse_json(row["payload_json"], {}),
+            "results": _parse_json(row["results_json"], []),
+            "created_at": row["created_at"].isoformat(),
+            "ingested_at": row["ingested_at"].isoformat() if row["ingested_at"] else None,
+        }
+
     def _proposal_dict(self, row: dict[str, Any]) -> dict[str, Any]:
         return {
             "id": row["id"],

tests/test_service.py

@@ -2,6 +2,8 @@ from __future__ import annotations
 
 from pathlib import Path
 
+from fastapi.testclient import TestClient
+
 from civicswarm.api import create_app
 from civicswarm.service import CivicSwarmService
 
@@ -89,3 +91,70 @@ def test_private_preference_aggregation_supports_noise(tmp_path: Path):
     assert clean["count"] == 3
     assert noisy["count"] >= 1
     assert noisy["mean_score"] != clean["mean_score"] or noisy["count"] != clean["count"]
+
+
+def test_capture_batch_ingest_is_atomic_and_idempotent(tmp_path: Path):
+    service = build_service(tmp_path)
+    proposal_id = seed_service(service)
+
+    batch = service.ingest_capture_batch(
+        "ward-3-mobile-001",
+        "sms",
+        [
+            {"kind": "resident", "resident_key": "resident-d", "geography": "ward-3", "interests": ["schools"], "languages": ["en"]},
+            {"kind": "comment", "proposal_id": proposal_id, "text": "Please keep crossings safe for elders.", "resident_key": "resident-d", "channel": "sms"},
+            {"kind": "preference", "proposal_id": proposal_id, "resident_key": "resident-d", "score": 0.8, "channel": "sms"},
+            {"kind": "ledger", "proposal_id": proposal_id, "entry_kind": "facilitator_note", "payload": {"note": "captured at block party"}},
+        ],
+    )
+
+    assert batch["batch_key"] == "ward-3-mobile-001"
+    assert batch["status"] == "processed"
+    assert len(batch["results"]) == 4
+    assert service.get_resident("resident-d")["geography"] == "ward-3"
+    comments = service.list_comments(proposal_id)
+    assert any(comment["metadata"].get("batch_key") == "ward-3-mobile-001" for comment in comments)
+    assert service.aggregate_preferences(proposal_id)["count"] == 4
+
+    duplicate = service.ingest_capture_batch(
+        "ward-3-mobile-001",
+        "sms",
+        [],
+    )
+    assert duplicate["id"] == batch["id"]
+    assert duplicate["results"] == batch["results"]
+    assert len(service.list_comments(proposal_id)) == len(comments)
+
+
+def test_capture_batch_endpoint_surfaces_audit_payload(tmp_path: Path):
+    app = create_app(tmp_path / "batch-api.sqlite")
+    client = TestClient(app)
+
+    proposal = client.post(
+        "/proposals",
+        json={
+            "city": "Springfield",
+            "title": "Crosswalk renewal",
+            "body": "Renew the busiest crosswalks near the library.",
+            "geography": "ward-4",
+            "tags": ["safety", "mobility"],
+        },
+    ).json()
+
+    response = client.post(
+        "/capture-batches",
+        json={
+            "batch_key": "ward-4-townhall-01",
+            "source": "field",
+            "items": [
+                {"kind": "comment", "proposal_id": proposal["id"], "text": "I support the renewal.", "channel": "field"},
+            ],
+        },
+    )
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["status"] == "processed"
+    assert payload["results"][0]["kind"] == "comment"
+    fetched = client.get("/capture-batches/ward-4-townhall-01").json()
+    assert fetched["batch_key"] == "ward-4-townhall-01"