build(agent): c3po#b883b4 iteration

.gitignore

@@ -0,0 +1,21 @@
+node_modules/
+.npmrc
+.env
+.env.*
+__tests__/
+coverage/
+.nyc_output/
+dist/
+build/
+.cache/
+*.log
+.DS_Store
+tmp/
+.tmp/
+__pycache__/
+*.pyc
+.venv/
+venv/
+*.egg-info/
+.pytest_cache/
+READY_TO_PUBLISH

AGENTS.md

@@ -0,0 +1,32 @@
+# AGENTS.md
+
+## Purpose
+CivicPulse is an offline-first coalition coordination core for organizers who need auditability, privacy controls, and deterministic sync across unreliable networks.
+
+## Stack
+- Python 3.11+
+- `pydantic` for schema validation
+- SQLite via the standard library for local event storage
+- `pytest` for tests
+- `build` for packaging verification
+
+## Architecture
+- `models.py`: domain objects for organizations, proposals, endorsements, briefs, shifts, tasks, and provenance records.
+- `ledger.py`: SQLite event ledger with immutable provenance records and audit-chain retrieval.
+- `sync.py`: deterministic merge and replay helpers for offline reconciliation.
+- `privacy.py`: role-based redaction and minimal-signal sharing.
+- `registry.py`: graph-of-contracts schema and adapter registry.
+- `service.py`: high-level orchestration API used by tests and CLI.
+- `cli.py`: small command-line entry point.
+
+## Rules
+- Keep changes deterministic and auditable.
+- Prefer small, explicit data transformations over hidden magic.
+- Do not remove provenance records; add new records instead.
+- Preserve privacy defaults when adding new shareable fields.
+- Update tests whenever behavior changes.
+
+## Verification
+- `bash test.sh`
+- `python3 -m pytest -q`
+- `python3 -m build`

README.md

@@ -1,3 +1,56 @@
-# idea192-civicpulse-offline-first
+# CivicPulse
 
-Source logic for Idea #192
+CivicPulse is an offline-first coalition intelligence and policy provenance core for community organizers, unions, tenant groups, and campaign coalitions.
+
+It focuses on three hard problems:
+- Deterministic coordination when teams work disconnected.
+- Auditable provenance for policy drafts, endorsements, and evidence.
+- Privacy-preserving sharing between organizations.
+
+## What is implemented
+
+- SQLite-backed immutable provenance ledger.
+- Coalition graph objects for organizations, contracts, proposals, endorsements, briefs, tasks, and shifts.
+- Deterministic sync helpers for replica reconciliation.
+- Role-based privacy redaction for shared policy briefs.
+- Graph-of-contracts registry for schemas and adapters.
+- CSV ingestion adapter for volunteer data.
+- CLI for audit and brief-sharing inspection.
+
+## Repository layout
+
+- `src/idea192_civicpulse_offline_first/` core package.
+- `tests/` pytest coverage for provenance, privacy, sync, and replay behavior.
+- `test.sh` clean verification entrypoint.
+- `AGENTS.md` contributor guidance for future agents.
+
+## Running
+
+```bash
+bash test.sh
+```
+
+Or individually:
+
+```bash
+python3 -m pytest -q
+python3 -m build
+```
+
+## CLI
+
+```bash
+civicpulse audit <entity_id>
+civicpulse share-brief <brief_id> <role>
+```
+
+## Design notes
+
+- Events are append-only and hashed for provenance.
+- Sync uses content-addressed event exchange, so local replicas can converge without a central admin.
+- Privacy defaults favor redaction, with minimal-signal sharing for public roles.
+- All stateful actions are recorded in the ledger before being surfaced through the service layer.
+
+## Packaging
+
+The package name is `idea192_civicpulse_offline_first`, and this README is wired into `pyproject.toml` as the project long description.

pyproject.toml

@@ -0,0 +1,25 @@
+[build-system]
+requires = ["setuptools>=68", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "idea192_civicpulse_offline_first"
+version = "0.1.0"
+description = "Offline-first coalition intelligence and policy provenance mesh for community organizers"
+readme = "README.md"
+requires-python = ">=3.11"
+license = "MIT"
+authors = [{name = "Community"}]
+dependencies = ["pydantic>=2.7"]
+
+[project.optional-dependencies]
+dev = ["build>=1.2", "pytest>=8.0"]
+
+[project.scripts]
+civicpulse = "idea192_civicpulse_offline_first.cli:main"
+
+[tool.setuptools]
+package-dir = {"" = "src"}
+
+[tool.setuptools.packages.find]
+where = ["src"]

src/idea192_civicpulse_offline_first/__init__.py

@@ -0,0 +1,29 @@
+"""CivicPulse offline-first coalition coordination core."""
+
+from .models import (
+    Amendment,
+    CoalitionNode,
+    ConsentRule,
+    DeploymentTask,
+    Endorsement,
+    EvidenceAttachment,
+    PolicyBrief,
+    Proposal,
+    RelationshipContract,
+    ShiftAssignment,
+)
+from .service import CivicPulseService
+
+__all__ = [
+    "Amendment",
+    "CoalitionNode",
+    "ConsentRule",
+    "CivicPulseService",
+    "DeploymentTask",
+    "Endorsement",
+    "EvidenceAttachment",
+    "PolicyBrief",
+    "Proposal",
+    "RelationshipContract",
+    "ShiftAssignment",
+]

src/idea192_civicpulse_offline_first/__main__.py

@@ -0,0 +1,5 @@
+from .cli import main
+
+
+if __name__ == "__main__":
+    main()

src/idea192_civicpulse_offline_first/cli.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+from .service import CivicPulseService
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(prog="civicpulse")
+    parser.add_argument("--db", default="civicpulse.sqlite3")
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    audit = sub.add_parser("audit")
+    audit.add_argument("entity_id")
+
+    brief = sub.add_parser("share-brief")
+    brief.add_argument("brief_id")
+    brief.add_argument("role")
+
+    args = parser.parse_args()
+    service = CivicPulseService(Path(args.db))
+
+    if args.command == "audit":
+        for item in service.audit(args.entity_id):
+            print(item)
+    elif args.command == "share-brief":
+        print(service.share_brief(args.brief_id, args.role))

src/idea192_civicpulse_offline_first/ledger.py

@@ -0,0 +1,132 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import sqlite3
+from collections.abc import Iterable
+from pathlib import Path
+from typing import Any
+
+from .models import ProvenanceRecord, utc_now
+
+
+def canonical_json(value: Any) -> str:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
+
+
+def digest_record(record: dict[str, Any]) -> str:
+    return hashlib.sha256(canonical_json(record).encode("utf-8")).hexdigest()
+
+
+class EventLedger:
+    def __init__(self, path: str | Path) -> None:
+        self.path = Path(path)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self.conn = sqlite3.connect(self.path)
+        self.conn.row_factory = sqlite3.Row
+        self._init_schema()
+
+    def _init_schema(self) -> None:
+        self.conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS provenance (
+                hash TEXT PRIMARY KEY,
+                previous_hash TEXT,
+                timestamp TEXT NOT NULL,
+                actor TEXT NOT NULL,
+                entity_type TEXT NOT NULL,
+                entity_id TEXT NOT NULL,
+                action TEXT NOT NULL,
+                payload TEXT NOT NULL
+            )
+            """
+        )
+        self.conn.commit()
+
+    def append(
+        self,
+        actor: str,
+        entity_type: str,
+        entity_id: str,
+        action: str,
+        payload: dict[str, Any],
+        timestamp: str | None = None,
+    ) -> ProvenanceRecord:
+        record = {
+            "previous_hash": self.latest_hash(),
+            "timestamp": timestamp or utc_now(),
+            "actor": actor,
+            "entity_type": entity_type,
+            "entity_id": entity_id,
+            "action": action,
+            "payload": payload,
+        }
+        record_hash = digest_record(record)
+        self.conn.execute(
+            "INSERT OR IGNORE INTO provenance VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+            (
+                record_hash,
+                record["previous_hash"],
+                record["timestamp"],
+                actor,
+                entity_type,
+                entity_id,
+                action,
+                canonical_json(payload),
+            ),
+        )
+        self.conn.commit()
+        return ProvenanceRecord(hash=record_hash, **record)
+
+    def latest_hash(self) -> str | None:
+        row = self.conn.execute("SELECT hash FROM provenance ORDER BY timestamp DESC, hash DESC LIMIT 1").fetchone()
+        return None if row is None else str(row["hash"])
+
+    def count(self) -> int:
+        row = self.conn.execute("SELECT COUNT(*) AS c FROM provenance").fetchone()
+        return int(row["c"] if row else 0)
+
+    def all(self) -> list[ProvenanceRecord]:
+        rows = self.conn.execute(
+            "SELECT * FROM provenance ORDER BY timestamp ASC, hash ASC"
+        ).fetchall()
+        return [
+            ProvenanceRecord(
+                hash=row["hash"],
+                previous_hash=row["previous_hash"],
+                timestamp=row["timestamp"],
+                actor=row["actor"],
+                entity_type=row["entity_type"],
+                entity_id=row["entity_id"],
+                action=row["action"],
+                payload=json.loads(row["payload"]),
+            )
+            for row in rows
+        ]
+
+    def missing_hashes(self, known_hashes: set[str]) -> list[ProvenanceRecord]:
+        return [record for record in self.all() if record.hash not in known_hashes]
+
+    def import_records(self, records: Iterable[ProvenanceRecord]) -> int:
+        inserted = 0
+        for record in records:
+            before = self.count()
+            self.conn.execute(
+                "INSERT OR IGNORE INTO provenance VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+                (
+                    record.hash,
+                    record.previous_hash,
+                    record.timestamp,
+                    record.actor,
+                    record.entity_type,
+                    record.entity_id,
+                    record.action,
+                    canonical_json(record.payload),
+                ),
+            )
+            self.conn.commit()
+            inserted += self.count() - before
+        return inserted
+
+    def audit_chain(self, entity_id: str) -> list[ProvenanceRecord]:
+        return [record for record in self.all() if record.entity_id == entity_id]

src/idea192_civicpulse_offline_first/models.py

@@ -0,0 +1,130 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from enum import StrEnum
+from typing import Any
+from uuid import uuid4
+
+from pydantic import BaseModel, Field
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def new_id(prefix: str) -> str:
+    return f"{prefix}_{uuid4().hex}"
+
+
+class CoalitionNode(BaseModel):
+    id: str = Field(default_factory=lambda: new_id("node"))
+    kind: str
+    name: str
+    jurisdiction: str | None = None
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+
+class RelationshipContract(BaseModel):
+    id: str = Field(default_factory=lambda: new_id("contract"))
+    left_node: str
+    right_node: str
+    relationship: str
+    version: int = 1
+    terms: dict[str, Any] = Field(default_factory=dict)
+    active: bool = True
+
+
+class ConsentRule(BaseModel):
+    role: str
+    allowed_fields: set[str] = Field(default_factory=set)
+    share_minimal_signal: bool = True
+
+
+class ProposalStatus(StrEnum):
+    pending = "pending"
+    approved = "approved"
+    rejected = "rejected"
+    withdrawn = "withdrawn"
+    contested = "contested"
+
+
+class Proposal(BaseModel):
+    id: str = Field(default_factory=lambda: new_id("proposal"))
+    title: str
+    body: str
+    sponsor_org: str
+    status: ProposalStatus = ProposalStatus.pending
+    revisions: list[str] = Field(default_factory=list)
+
+
+class Amendment(BaseModel):
+    id: str = Field(default_factory=lambda: new_id("amendment"))
+    proposal_id: str
+    org_id: str
+    body: str
+    accepted: bool | None = None
+
+
+class EndorsementStatus(StrEnum):
+    proposed = "proposed"
+    approved = "approved"
+    declined = "declined"
+    withdrawn = "withdrawn"
+    contested = "contested"
+
+
+class Endorsement(BaseModel):
+    id: str = Field(default_factory=lambda: new_id("endorsement"))
+    proposal_id: str
+    org_id: str
+    status: EndorsementStatus = EndorsementStatus.proposed
+    rationale: str | None = None
+    signature: str | None = None
+
+
+class ShiftAssignment(BaseModel):
+    id: str = Field(default_factory=lambda: new_id("shift"))
+    campaign_id: str
+    region: str
+    volunteers: list[str] = Field(default_factory=list)
+    required_count: int = 0
+    status: str = "planned"
+
+
+class DeploymentTask(BaseModel):
+    id: str = Field(default_factory=lambda: new_id("task"))
+    campaign_id: str
+    title: str
+    owner_org: str
+    due_at: str | None = None
+    status: str = "open"
+    fields: dict[str, Any] = Field(default_factory=dict)
+
+
+class EvidenceAttachment(BaseModel):
+    id: str = Field(default_factory=lambda: new_id("evidence"))
+    brief_id: str
+    uri: str
+    digest: str
+    note: str | None = None
+
+
+class PolicyBrief(BaseModel):
+    id: str = Field(default_factory=lambda: new_id("brief"))
+    title: str
+    body: str
+    source_refs: list[str] = Field(default_factory=list)
+    revision: int = 1
+    supersedes: str | None = None
+    status: str = "draft"
+
+
+class ProvenanceRecord(BaseModel):
+    hash: str
+    previous_hash: str | None = None
+    timestamp: str
+    actor: str
+    entity_type: str
+    entity_id: str
+    action: str
+    payload: dict[str, Any] = Field(default_factory=dict)

src/idea192_civicpulse_offline_first/privacy.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from .models import ConsentRule
+
+
+DEFAULT_REDACTIONS = {"address", "email", "phone", "notes", "member_name", "member_id"}
+
+
+def redact_payload(payload: Mapping[str, Any], rule: ConsentRule | None) -> dict[str, Any]:
+    if rule is None:
+        rule = ConsentRule(role="public")
+    visible = set(rule.allowed_fields)
+    output: dict[str, Any] = {}
+    for key, value in payload.items():
+        if key in visible:
+            output[key] = value
+        elif key in DEFAULT_REDACTIONS:
+            output[key] = "[redacted]"
+        elif rule.share_minimal_signal:
+            if isinstance(value, (int, float, bool)):
+                output[key] = value
+            elif isinstance(value, list):
+                output[key] = len(value)
+            else:
+                output[key] = "[redacted]"
+        else:
+            output[key] = "[redacted]"
+    return output
+
+
+def minimal_signal(payload: Mapping[str, Any]) -> dict[str, Any]:
+    return {
+        key: value
+        for key, value in payload.items()
+        if isinstance(value, (int, float, bool)) or key.endswith("_count")
+    }

src/idea192_civicpulse_offline_first/registry.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass(frozen=True)
+class SchemaDefinition:
+    schema_id: str
+    version: str
+    fields: tuple[str, ...]
+    required: tuple[str, ...] = ()
+
+
+@dataclass
+class AdapterDefinition:
+    adapter_id: str
+    kind: str
+    config: dict[str, Any] = field(default_factory=dict)
+
+
+class GraphRegistry:
+    def __init__(self) -> None:
+        self._schemas: dict[str, SchemaDefinition] = {}
+        self._adapters: dict[str, AdapterDefinition] = {}
+
+    def register_schema(self, schema: SchemaDefinition) -> None:
+        self._schemas[schema.schema_id] = schema
+
+    def register_adapter(self, adapter: AdapterDefinition) -> None:
+        self._adapters[adapter.adapter_id] = adapter
+
+    def describe(self) -> dict[str, Any]:
+        return {
+            "schemas": [schema.__dict__ for schema in self._schemas.values()],
+            "adapters": [adapter.__dict__ for adapter in self._adapters.values()],
+        }
+
+
+def csv_rows(text: str) -> list[dict[str, str]]:
+    import csv
+    from io import StringIO
+
+    reader = csv.DictReader(StringIO(text.strip()))
+    return [dict(row) for row in reader]

src/idea192_civicpulse_offline_first/service.py

@@ -0,0 +1,143 @@
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+from typing import Any
+
+from .ledger import EventLedger
+from .models import (
+    Amendment,
+    CoalitionNode,
+    ConsentRule,
+    DeploymentTask,
+    Endorsement,
+    EndorsementStatus,
+    PolicyBrief,
+    Proposal,
+    ProposalStatus,
+    RelationshipContract,
+    ShiftAssignment,
+)
+from .privacy import redact_payload
+from .registry import AdapterDefinition, GraphRegistry, SchemaDefinition, csv_rows
+from .sync import SyncResult, sync_ledgers
+
+
+class CivicPulseService:
+    def __init__(self, db_path: str | Path) -> None:
+        self.ledger = EventLedger(db_path)
+        self.registry = GraphRegistry()
+        self.consent_rules: dict[str, ConsentRule] = {
+            "public": ConsentRule(role="public", allowed_fields={"id", "title", "status"}),
+            "org": ConsentRule(role="org", allowed_fields={"id", "title", "status", "revision", "source_refs"}),
+            "admin": ConsentRule(role="admin", allowed_fields={"id", "title", "body", "status", "revision", "source_refs"}),
+        }
+        self.nodes: dict[str, CoalitionNode] = {}
+        self.contracts: dict[str, RelationshipContract] = {}
+        self.proposals: dict[str, Proposal] = {}
+        self.amendments: dict[str, Amendment] = {}
+        self.endorsements: dict[str, Endorsement] = {}
+        self.briefs: dict[str, PolicyBrief] = {}
+        self.tasks: dict[str, DeploymentTask] = {}
+        self.shifts: dict[str, ShiftAssignment] = {}
+
+    def register_schema_defaults(self) -> None:
+        self.registry.register_schema(SchemaDefinition("coalition-node", "1.0", ("id", "kind", "name", "jurisdiction")))
+        self.registry.register_schema(SchemaDefinition("policy-brief", "1.0", ("id", "title", "body", "revision")))
+        self.registry.register_adapter(AdapterDefinition("csv-volunteer-import", "import", {"format": "csv"}))
+
+    def create_org(self, name: str, jurisdiction: str | None = None) -> CoalitionNode:
+        node = CoalitionNode(kind="organization", name=name, jurisdiction=jurisdiction)
+        self.nodes[node.id] = node
+        self.ledger.append("system", "organization", node.id, "create", node.model_dump())
+        return node
+
+    def link_contract(self, left_node: str, right_node: str, relationship: str, terms: dict[str, Any] | None = None) -> RelationshipContract:
+        contract = RelationshipContract(left_node=left_node, right_node=right_node, relationship=relationship, terms=terms or {})
+        self.contracts[contract.id] = contract
+        self.ledger.append("system", "contract", contract.id, "link", contract.model_dump())
+        return contract
+
+    def create_proposal(self, sponsor_org: str, title: str, body: str) -> Proposal:
+        proposal = Proposal(sponsor_org=sponsor_org, title=title, body=body)
+        self.proposals[proposal.id] = proposal
+        self.ledger.append(sponsor_org, "proposal", proposal.id, "create", proposal.model_dump())
+        return proposal
+
+    def add_amendment(self, proposal_id: str, org_id: str, body: str, accepted: bool | None = None) -> Amendment:
+        amendment = Amendment(proposal_id=proposal_id, org_id=org_id, body=body, accepted=accepted)
+        self.amendments[amendment.id] = amendment
+        self.ledger.append(org_id, "amendment", amendment.id, "propose", amendment.model_dump())
+        proposal = self.proposals[proposal_id]
+        proposal.revisions.append(amendment.id)
+        if accepted is True:
+            proposal.status = ProposalStatus.contested if proposal.status == ProposalStatus.rejected else ProposalStatus.approved
+        return amendment
+
+    def decide_endorsement(self, proposal_id: str, org_id: str, approved: bool, rationale: str | None = None) -> Endorsement:
+        endorsement = Endorsement(
+            proposal_id=proposal_id,
+            org_id=org_id,
+            status=EndorsementStatus.approved if approved else EndorsementStatus.declined,
+            rationale=rationale,
+            signature=self._signature({"proposal_id": proposal_id, "org_id": org_id, "approved": approved, "rationale": rationale}),
+        )
+        self.endorsements[endorsement.id] = endorsement
+        self.ledger.append(org_id, "endorsement", endorsement.id, "decide", endorsement.model_dump())
+        self._recompute_proposal_status(proposal_id)
+        return endorsement
+
+    def create_policy_brief(self, title: str, body: str, source_refs: list[str] | None = None, supersedes: str | None = None) -> PolicyBrief:
+        brief = PolicyBrief(title=title, body=body, source_refs=source_refs or [], supersedes=supersedes)
+        if supersedes and supersedes in self.briefs:
+            brief.revision = self.briefs[supersedes].revision + 1
+        self.briefs[brief.id] = brief
+        self.ledger.append("policy-office", "policy_brief", brief.id, "publish", brief.model_dump())
+        return brief
+
+    def attach_evidence(self, brief_id: str, uri: str, digest: str, note: str | None = None) -> dict[str, Any]:
+        payload = {"brief_id": brief_id, "uri": uri, "digest": digest, "note": note}
+        self.ledger.append("policy-office", "evidence", brief_id, "attach", payload)
+        return payload
+
+    def allocate_shift(self, campaign_id: str, region: str, volunteers: list[str], required_count: int) -> ShiftAssignment:
+        shift = ShiftAssignment(campaign_id=campaign_id, region=region, volunteers=volunteers, required_count=required_count)
+        self.shifts[shift.id] = shift
+        self.ledger.append("field-team", "shift", shift.id, "allocate", shift.model_dump())
+        return shift
+
+    def create_task(self, campaign_id: str, title: str, owner_org: str, due_at: str | None = None, fields: dict[str, Any] | None = None) -> DeploymentTask:
+        task = DeploymentTask(campaign_id=campaign_id, title=title, owner_org=owner_org, due_at=due_at, fields=fields or {})
+        self.tasks[task.id] = task
+        self.ledger.append(owner_org, "task", task.id, "create", task.model_dump())
+        return task
+
+    def share_brief(self, brief_id: str, role: str) -> dict[str, Any]:
+        brief = self.briefs[brief_id]
+        rule = self.consent_rules.get(role, self.consent_rules["public"])
+        return redact_payload(brief.model_dump(), rule)
+
+    def audit(self, entity_id: str) -> list[dict[str, Any]]:
+        return [record.model_dump() for record in self.ledger.audit_chain(entity_id)]
+
+    def sync_with(self, other: "CivicPulseService") -> SyncResult:
+        return sync_ledgers(self.ledger, other.ledger)
+
+    def ingest_volunteers_csv(self, text: str) -> list[dict[str, str]]:
+        rows = csv_rows(text)
+        self.ledger.append("importer", "adapter", "csv-volunteer-import", "ingest", {"row_count": len(rows)})
+        return rows
+
+    def _signature(self, payload: dict[str, Any]) -> str:
+        return hashlib.sha256(str(sorted(payload.items())).encode("utf-8")).hexdigest()
+
+    def _recompute_proposal_status(self, proposal_id: str) -> None:
+        proposal = self.proposals[proposal_id]
+        decisions = [item for item in self.endorsements.values() if item.proposal_id == proposal_id]
+        if any(item.status == EndorsementStatus.contested for item in decisions):
+            proposal.status = ProposalStatus.contested
+            return
+        if decisions and all(item.status == EndorsementStatus.approved for item in decisions):
+            proposal.status = ProposalStatus.approved
+        elif any(item.status == EndorsementStatus.declined for item in decisions):
+            proposal.status = ProposalStatus.rejected

src/idea192_civicpulse_offline_first/sync.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .ledger import EventLedger
+from .models import ProvenanceRecord
+
+
+@dataclass(frozen=True)
+class SyncResult:
+    accepted: int
+    missing_local: int
+
+
+def deterministic_replay(records: list[ProvenanceRecord]) -> list[ProvenanceRecord]:
+    return sorted(records, key=lambda item: (item.timestamp, item.hash))
+
+
+def sync_ledgers(local: EventLedger, remote: EventLedger) -> SyncResult:
+    local_hashes = {record.hash for record in local.all()}
+    remote_hashes = {record.hash for record in remote.all()}
+    imported_local = local.import_records(remote.missing_hashes(local_hashes))
+    imported_remote = remote.import_records(local.missing_hashes(remote_hashes))
+    return SyncResult(accepted=imported_local + imported_remote, missing_local=len(local.missing_hashes(remote_hashes)))

test.sh

@@ -0,0 +1,6 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+python3 -m pip install -e ".[dev]"
+python3 -m pytest -q
+python3 -m build

tests/test_civicpulse.py

@@ -0,0 +1,53 @@
+from pathlib import Path
+
+from idea192_civicpulse_offline_first.service import CivicPulseService
+from idea192_civicpulse_offline_first.sync import deterministic_replay
+
+
+def test_endorsement_and_audit(tmp_path: Path) -> None:
+    service = CivicPulseService(tmp_path / "ledger.sqlite3")
+    org = service.create_org("Tenant Action Network")
+    proposal = service.create_proposal(org.id, "Endorse rent stabilization", "Coalition policy brief")
+    endorsement = service.decide_endorsement(proposal.id, org.id, True, "Aligned with tenant protections")
+
+    audit = service.audit(proposal.id)
+
+    assert proposal.status == "approved"
+    assert endorsement.signature
+    assert any(item["action"] == "create" for item in audit)
+    assert any(item["action"] == "decide" for item in service.audit(endorsement.id))
+
+
+def test_privacy_redacts_public_briefs(tmp_path: Path) -> None:
+    service = CivicPulseService(tmp_path / "ledger.sqlite3")
+    brief = service.create_policy_brief("Field brief", "Sensitive analysis", ["source-1"])
+
+    public_view = service.share_brief(brief.id, "public")
+    admin_view = service.share_brief(brief.id, "admin")
+
+    assert public_view["body"] == "[redacted]"
+    assert admin_view["body"] == "Sensitive analysis"
+    assert public_view["revision"] == 1
+
+
+def test_sync_merges_ledger_events(tmp_path: Path) -> None:
+    left = CivicPulseService(tmp_path / "left.sqlite3")
+    right = CivicPulseService(tmp_path / "right.sqlite3")
+
+    org = left.create_org("Coalition A")
+    left.create_task("campaign-1", "Call supporters", org.id, fields={"region": "North"})
+    result = left.sync_with(right)
+
+    assert result.accepted > 0
+    assert right.ledger.count() > 0
+
+
+def test_deterministic_replay_orders_by_timestamp_and_hash(tmp_path: Path) -> None:
+    service = CivicPulseService(tmp_path / "ledger.sqlite3")
+    org = service.create_org("Organization")
+    service.create_task("campaign", "Door knock", org.id)
+
+    records = service.ledger.all()
+    replayed = deterministic_replay(records)
+
+    assert replayed == sorted(records, key=lambda item: (item.timestamp, item.hash))