diff --git a/IMPLEMENTATION_PLAN.md b/IMPLEMENTATION_PLAN.md
index 067b785..9d9c6d8 100644
--- a/IMPLEMENTATION_PLAN.md
+++ b/IMPLEMENTATION_PLAN.md
@@ -1,6 +1,6 @@
# Veritras implementation plan
-Originally prepared October 17, 2026 from [PROJECT_REVIEW.md](PROJECT_REVIEW.md). **Status updated October 19, 2026 for increment 7, following analytics release PR #22 (`ff88bcc`).** Integration recovery checks, typed contracts, protected-main merge controls and production admission gates are implemented. Final CI/deployment evidence is recorded in the increment 7 pull request; see [release controls](RELEASE_CONTROLS.md) and the [release/recovery runbook](RELEASE_RUNBOOK.md).
+Originally prepared October 17, 2026 from [PROJECT_REVIEW.md](PROJECT_REVIEW.md). **Status updated October 19, 2026 for increment 9.** Integration recovery checks, typed contracts, protected-main merge controls and production admission gates are implemented. Final CI/deployment evidence is recorded in the increment 7 pull request; see [release controls](RELEASE_CONTROLS.md) and the [release/recovery runbook](RELEASE_RUNBOOK.md).
## Status at a glance
@@ -21,7 +21,7 @@ Priority meanings: **P0 — release blocker** must be resolved before enabling p
| Analytics and exports (increment 6) | **Done** | **P1 addressed** | Canonical shipment cohorts/lifecycle, optional locked SLA, real route/sample metrics, UTC filters and safe matching exports; see [runbook](ANALYTICS.md) |
| Integration and release controls (increment 7) | **Done** | **P1 addressed** | Real write-failure/restart, process-loss, reorg and backup/restore rehearsals; typed detail/verification; protected main and exact-commit production admission; see [runbook](RELEASE_RUNBOOK.md) |
| QR verification, shipment detail and guided demo (increment 8) | **Done** | **P2 addressed** | Direct public links and SVG QR labels, canonical paginated timeline, shipment detail with contract-checked transitions, isolated guided demo, mobile wallet/keyboard access and route metadata; see [guide](PRODUCT_JOURNEY.md) |
-| Shipment exceptions (increment 9) | **Pending** | P2 | Depends on trustworthy lifecycle timestamps, SLA definitions and synchronization |
+| Shipment exceptions (increment 9) | **Done** | **P2 addressed** | Configurable owner policy, canonical arrival/checkpoint/custody rules, durable deduplicated occurrences, acknowledgement/history, synchronized evaluation and public mobile list; see [guide](SHIPMENT_EXCEPTIONS.md) |
| Performance and operations (increment 10) | **Pending** | P2 | Existing lazy routes/Lighthouse checks provide a baseline; measured optimization and operational completion remain |
| Business pilot integrations | **Deferred** | Later | Organizations, evidence files, imports, standards integrations and repeated journeys |
@@ -43,7 +43,7 @@ Priority meanings: **P0 — release blocker** must be resolved before enabling p
Historical increment 4 validation: **236 backend tests, 192 frontend tests, 23 Playwright regressions and 40 contract tests passed**, plus builds, generated-type checks, Lighthouse and real local integration/browser journeys. Hosted Sepolia registration → checkpoints → custody → delivery passed; a same-code testing redeploy preserved event counts. The backend matrix covers SQLite and PostgreSQL, with two intentional skips for PostgreSQL-only checks in the SQLite variant. A real local chain fork/recovery and hosted same-code redeploy also passed.
-Current released schema is `c62a904df318`; increments 7–8 add no migration. Historical durable-indexing rollout is tracked by [PR #18](https://github.com/Oionos/Veritras/pull/18); analytics was released in [PR #22](https://github.com/Oionos/Veritras/pull/22). Production indexing remains explicitly disabled. Increment 7 adds real PostgreSQL failure/process-loss and restart checks, local reorganization recovery, a locked backup/restore and read-only reconciliation rehearsal, and production gate rejection regressions. Test counts and final commit/deployment evidence are recorded in its pull request. Testing uses a dedicated Neon project/database (`Veritras Testing` / `veritras_preview`), a separate Render API and distinct Sepolia contracts. Setup and resource details: [TESTING.md](TESTING.md), [HOSTED_TESTING.md](HOSTED_TESTING.md), [AUTHENTICATION.md](AUTHENTICATION.md).
+Current schema is `e93a107bd624`; increments 7–8 add no migration. Historical durable-indexing rollout is tracked by [PR #18](https://github.com/Oionos/Veritras/pull/18); analytics was released in [PR #22](https://github.com/Oionos/Veritras/pull/22). Production indexing remains explicitly disabled. Increment 7 adds real PostgreSQL failure/process-loss and restart checks, local reorganization recovery, a locked backup/restore and read-only reconciliation rehearsal, and production gate rejection regressions. Test counts and final commit/deployment evidence are recorded in its pull request. Testing uses a dedicated Neon project/database (`Veritras Testing` / `veritras_preview`), a separate Render API and distinct Sepolia contracts. Setup and resource details: [TESTING.md](TESTING.md), [HOSTED_TESTING.md](HOSTED_TESTING.md), [AUTHENTICATION.md](AUTHENTICATION.md).
## Goal and completion criteria
@@ -285,7 +285,7 @@ Completed in increment 7: actual PostgreSQL checkpoint failure after a metadata
### PR 9: Shipment exception list
-**Status: pending. Priority: P2, after trustworthy metrics and synchronization.** Existing anomaly calculations do not satisfy the planned deduplicated, acknowledged exception lifecycle.
+**Status: implemented (increment 9). Priority: P2 addressed.** Configurable rules, canonical/synchronization gates, durable episodes and acknowledgement/history are delivered. See [SHIPMENT_EXCEPTIONS.md](SHIPMENT_EXCEPTIONS.md) for definitions, defaults and recovery behavior.
- Add configurable expected arrival and checkpoint-interval/dwell thresholds with explicit meanings.
- Show overdue shipments, stale checkpoint activity, and unresolved custody handoffs in an in-app exception list.
diff --git a/README.md b/README.md
index 05175e5..e045524 100644
--- a/README.md
+++ b/README.md
@@ -91,3 +91,7 @@ Veritras was built as a **portfolio project** to demonstrate full-stack blockcha
Try the Live Demo
+
+## Shipment exceptions
+
+See [the exception guide](SHIPMENT_EXCEPTIONS.md) for configurable rules, synchronization gates, acknowledgement/history, owner permissions and additive schema revision `e93a107bd624`. Open `/exceptions` or a shipment detail page. Production evaluation remains paused while indexing is disabled.
diff --git a/RELEASE_RUNBOOK.md b/RELEASE_RUNBOOK.md
index c06523b..ab522ff 100644
--- a/RELEASE_RUNBOOK.md
+++ b/RELEASE_RUNBOOK.md
@@ -32,7 +32,7 @@ Use `TESTING_COMPOSE_PROJECT=` for an existing named
The full journey uses actual wallet signatures, transactions, PostgreSQL and API calls: draft registration, custody A to B, checkpoints, delivery, verification and matching analytics/exports. Unauthorized contract writes and duplicate registration must revert. The reorganization drill replaces a mined branch and checks restored metadata, history, custody and shipment projections. The write-failure drill makes PostgreSQL reject a checkpoint in a batch that also updates metadata; journal, cursor and projections must all remain at the preceding commit across an API restart. Removing the fault replays the batch once; a second restart cannot duplicate it. A separate PostgreSQL regression kills a real writer process with uncommitted SQL and verifies lock release and complete rollback.
-The backup rehearsal validates the local chain/database before DDL, holds the indexer writer lock, uses `pg_dump`, creates a fresh clone and restores with `pg_restore --exit-on-error`. It checks schema head and runs the identity/history reconciliation functions read-only against the actual chain. Before/after fingerprints cover products, shipments, checkpoints, transfers, canonical/noncanonical journal, anchors, cursor, baselines, aliases, registration attempts and schema version. Source, backup and restored fingerprints must match exactly. The source is never reset. An error releases the writer lock on connection close. Private backups and named clones are retained for inspection; remove only a specifically identified disposable rehearsal artifact when finished, after checking it is not needed.
+The backup rehearsal validates the local chain/database before DDL, holds the indexer writer lock, uses `pg_dump`, creates a fresh clone and restores with `pg_restore --exit-on-error`. It checks schema head and runs the identity/history reconciliation functions read-only against the actual chain. Before/after fingerprints cover products, shipments, checkpoints, transfers, canonical/noncanonical journal, anchors, cursor, baselines, aliases, registration attempts, exception policies/occurrences/evaluation state and schema version. Source, backup and restored fingerprints must match exactly. The source is never reset. An error releases the writer lock on connection close. Private backups and named clones are retained for inspection; remove only a specifically identified disposable rehearsal artifact when finished, after checking it is not needed.
## Separate production indexing/backfill gate
@@ -58,3 +58,7 @@ This section is a future operator procedure, not authorization to enable product
| Database restore required | Pause writers and writes; preserve a fresh snapshot of the failing state. Restore the rehearsed backup into a new isolated target first, verify schema and fingerprints, reconcile confirmed chain events since the backup, and review counts/identity before a separately approved connection switch. Restoring a database does not undo mined transactions. |
Application rollback retains the additive schema, aliases, registration attempts and event journal. Never run a destructive schema downgrade or delete canonical history as a deployment rollback. Recheck API readiness and frontend configuration/attestation after recovery. A healthy old frontend can remain on a compatible API while a corrective backend release is prepared.
+
+## Increment 9 schema compatibility
+
+Shipment exceptions add revision `e93a107bd624`; see [SHIPMENT_EXCEPTIONS.md](SHIPMENT_EXCEPTIONS.md). Rehearse the generated additive SQL on an isolated copy and retain a pre-migration checkpoint before explicitly applying it. Old running handlers remain compatible, but an older backend artifact cannot restart against a schema head it does not recognize. Use a compatible corrective release and retain all exception history. Production indexing stays disabled, so no production exception inference or backfill is enabled by this release.
diff --git a/SHIPMENT_EXCEPTIONS.md b/SHIPMENT_EXCEPTIONS.md
new file mode 100644
index 0000000..ef7b53b
--- /dev/null
+++ b/SHIPMENT_EXCEPTIONS.md
@@ -0,0 +1,46 @@
+# Shipment exceptions (increment 9)
+
+`/exceptions` lists durable operational observations and resolved history. Each entry links to the shipment's canonical evidence. Shipment detail shows open observations and the owner's exception policy. These observations describe missing **confirmed indexed records**; they do not prove a physical shipment delay or authenticate an item.
+
+## Rules and defaults
+
+| Rule | Trigger | Default | Clears when |
+| --- | --- | --- | --- |
+| Planned arrival overdue | A planned arrival has passed after canonical departure, without canonical delivery. The plan must have been recorded on or before that departure. | Enabled; uses the shipment's optional UTC arrival plan | Delivery, valid plan/eligibility change, disabled rule or withdrawn departure evidence |
+| Checkpoint activity overdue | No checkpoint follows the latest checkpoint during a journey with an indexed departure | 24 hours | A new checkpoint resets the interval, delivery, disabled rule or evidence withdrawal |
+| Checkpoint follow-up overdue | No checkpoint follows the latest **At Checkpoint** record | 24 hours | A new checkpoint, delivery, disabled rule or evidence withdrawal |
+| Receiving handler follow-up overdue | The latest custody transfer has no subsequent checkpoint by its receiving handler | 6 hours | A checkpoint by that recipient, a newer transfer, delivery, disabled rule or evidence withdrawal |
+
+A threshold is reached at `now >= anchor timestamp + threshold`. At Checkpoint uses its own interval instead of also raising the general inactivity rule. It measures time since a **record**, not physical dwell. Custody transfers already take effect on-chain; follow-up does not represent a pending acceptance transaction. A checkpoint by a different wallet does not satisfy receiving-handler follow-up. Transaction/log order decides which event follows another, including equal timestamps.
+
+Owners can set intervals from 0.25 to 8760 hours or leave an interval blank to disable its rule. Arrival planning remains on the shipment record and is locked before indexed departure. Unknown historical plans are not invented. Drafts, unreconciled scope, superseded products and products without a canonical registration cannot establish exceptions.
+
+## Lifecycle and permissions
+
+- An occurrence persists with its threshold-crossing time, first/last observation, anchor event and occurrence identities, policy revision and evidence summary.
+- A partial unique index permits one open occurrence per scope, shipment and rule. Repeated evaluations update that occurrence; they do not duplicate it.
+- Material changes to a still-overdue rule/anchor resolve the prior occurrence as `condition_replaced` and open a new one. An unrelated policy change preserves acknowledgements. Newly cleared conditions use `condition_cleared`; removed canonical anchors use `evidence_withdrawn`.
+- A previously resolved condition that recurs opens a new occurrence. Its acknowledgement starts empty. Closed history and earlier acknowledgements remain intact across restarts and reorg recovery.
+- Only an authenticated, established product owner in the current contract scope may change policy or acknowledge. Current chain custody alone does not grant these API permissions. Legacy manufacturer text grants no rights.
+- Policy PUTs require `expected_revision`; stale saves return 409. Identical saves are idempotent. Acknowledgement POSTs are idempotent and preserve the first actor, timestamp and note. Resolved occurrences return 409.
+- Notes are public, plain text and limited to 500 characters. Acknowledgement records review; it does not resolve or suppress a condition. No email, webhook, external delivery or retry queue is configured.
+
+## Synchronization and operation
+
+The API process runs a lightweight evaluator approximately every 60 seconds when indexing is enabled. No additional paid service is required. Disabled indexing starts no evaluator. Render sleep naturally pauses assessment; the page reports the stored assessment time.
+
+Evaluation requires cached indexer status `current`, readiness, zero lag and zero finality gap. Inside the indexer's shared transaction writer lock, the evaluator checks the configured scope's durable cursor and block anchor against cached progress. It rechecks indexer health before committing. Journal reorgs, concurrent evaluators, policy writes and acknowledgements cannot interleave their write transactions. Failure rolls back the entire assessment. No RPC is performed by exception rules or public reads.
+
+Disabled, stale, catching-up, changed or unavailable indexing opens/resolves nothing. Invalid/future event timestamps and missing recipient evidence retain that shipment's previous observations; the response reports an uncertain-shipment count. A regressed clock cannot overwrite a later assessment. The UI labels paused evaluation and retains prior observations without treating missing data as proof of delay. Assessments older than three minutes are labelled stale even if indexer health is current. Public GET requests never evaluate or mutate lifecycle state.
+
+The list polls every 30 seconds, supports open/unacknowledged/acknowledged/resolved/all filters, shipment filtering and bounded pagination (maximum 100 via API). No count is inferred from client time. Existing anomaly analytics remain a separate statistical view.
+
+**Production indexing remains disabled.** Its exception list therefore shows paused evaluation. Production reconciliation/backfill and enablement retain their separate reviewed gate in [RELEASE_RUNBOOK.md](RELEASE_RUNBOOK.md).
+
+## Schema and release
+
+Additive Alembic revision `e93a107bd624` follows `c62a904df318`. It adds `exception_policies`, `shipment_exceptions` and `exception_evaluations` with constraints and indexes. There is no business-data backfill or canonical-journal modification. Rehearse on isolated copies, retain a pre-migration checkpoint, compare existing table fingerprints, and apply the same generated SQL transaction to the explicitly selected target before starting the new backend.
+
+October 19 rehearsals preserved fingerprints for products, shipments, checkpoints, transfers, canonical/orphan journal, block anchors, cursor, bindings, aliases and registration attempts on copies of both hosted databases. Retained `before-shipment-exceptions-20261019` checkpoint branches expire October 26; disposable rehearsal branches expire October 20. No database credentials were changed. Keep the additive schema on rollback. An old API artifact with an earlier schema head cannot pass its strict startup admission against this revision; use a corrective compatible release instead of a destructive downgrade.
+
+Recovery/backup fingerprints now include all three exception tables. Local PostgreSQL regressions use disposable schemas, preserving the shared regression database. Controlled-clock tests cover threshold crossings, custody source order, deduplication, material policy changes, acknowledgements, delivery, orphan recovery, uncertain evidence, paused/changed indexing, interrupted commits, restart persistence and concurrent writers. Browser coverage includes paused states, public mobile history, shipment links, retry and owner policy persistence/permission denial in the actual local wallet journey.
diff --git a/backend/README.md b/backend/README.md
index 9d06ee9..1434049 100644
--- a/backend/README.md
+++ b/backend/README.md
@@ -147,3 +147,7 @@ MIT
## Indexer startup and health
Use `/health/live` for hosting HTTP checks and `/health/ready` for dependency/indexer readiness. Initialization, confirmed-target lag, cursor progress and head freshness are reported separately. See [INDEXER_HEALTH.md](../INDEXER_HEALTH.md) for configuration, capped retries, safe diagnostics and bounded release probes. Production indexing remains disabled pending its separate enablement gate.
+
+## Shipment exceptions
+
+See [the exception guide](../SHIPMENT_EXCEPTIONS.md) for configurable rules, synchronization gates, acknowledgement/history, owner permissions and additive schema revision `e93a107bd624`. Open `/exceptions` or a shipment detail page. Production evaluation remains paused while indexing is disabled.
diff --git a/backend/alembic/versions/e93a107bd624_shipment_exceptions.py b/backend/alembic/versions/e93a107bd624_shipment_exceptions.py
new file mode 100644
index 0000000..1d68ad2
--- /dev/null
+++ b/backend/alembic/versions/e93a107bd624_shipment_exceptions.py
@@ -0,0 +1,98 @@
+"""Durable exception episodes and owner policy; no historical backfill.
+
+Revision ID: e93a107bd624
+Revises: c62a904df318
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "e93a107bd624"
+down_revision = "c62a904df318"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+ if op.get_bind().dialect.name == "postgresql":
+ op.execute("SET LOCAL lock_timeout = '5s'")
+ op.execute("SELECT pg_advisory_xact_lock(864382024)")
+ op.execute(
+ "DO $$ BEGIN IF (SELECT version_num FROM alembic_version) IS DISTINCT FROM 'c62a904df318' THEN RAISE EXCEPTION 'Exception migration requires c62a904df318'; END IF; END $$"
+ )
+ op.create_table(
+ "exception_policies",
+ sa.Column(
+ "shipment_id",
+ sa.String(),
+ sa.ForeignKey("shipments.shipment_id"),
+ primary_key=True,
+ ),
+ sa.Column("revision", sa.Integer(), nullable=False),
+ sa.Column("arrival_enabled", sa.Boolean(), nullable=False),
+ sa.Column("inactivity_hours", sa.Float()),
+ sa.Column("checkpoint_hours", sa.Float()),
+ sa.Column("custody_hours", sa.Float()),
+ sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+ sa.Column("updated_by", sa.String(42), nullable=False),
+ *[
+ sa.CheckConstraint(
+ f"{name} IS NULL OR ({name} >= 0.25 AND {name} <= 8760)",
+ name=f"ck_exception_policy_{name}",
+ )
+ for name in ("inactivity_hours", "checkpoint_hours", "custody_hours")
+ ],
+ )
+ op.create_table(
+ "exception_evaluations",
+ sa.Column("scope", sa.String(64), primary_key=True),
+ sa.Column("evaluated_at", sa.DateTime(timezone=True), nullable=False),
+ sa.Column("block_number", sa.BigInteger(), nullable=False),
+ sa.Column("block_hash", sa.String(66), nullable=False),
+ sa.Column("uncertain_shipments", sa.Integer(), nullable=False),
+ )
+ op.create_table(
+ "shipment_exceptions",
+ sa.Column("id", sa.String(36), primary_key=True),
+ sa.Column("scope", sa.String(64), nullable=False),
+ sa.Column(
+ "shipment_id",
+ sa.String(),
+ sa.ForeignKey("shipments.shipment_id"),
+ nullable=False,
+ ),
+ sa.Column("kind", sa.String(32), nullable=False),
+ sa.Column("fingerprint", sa.String(64), nullable=False),
+ sa.Column("triggered_at", sa.DateTime(timezone=True), nullable=False),
+ sa.Column("first_observed_at", sa.DateTime(timezone=True), nullable=False),
+ sa.Column("last_observed_at", sa.DateTime(timezone=True), nullable=False),
+ sa.Column("resolved_at", sa.DateTime(timezone=True)),
+ sa.Column("resolution", sa.String(40)),
+ sa.Column("acknowledged_at", sa.DateTime(timezone=True)),
+ sa.Column("acknowledged_by", sa.String(42)),
+ sa.Column("acknowledgement_note", sa.String(500)),
+ sa.Column("evidence", sa.JSON(), nullable=False),
+ sa.CheckConstraint(
+ "kind IN ('arrival_overdue','checkpoint_inactivity','checkpoint_wait','custody_followup')",
+ name="ck_shipment_exception_kind",
+ ),
+ )
+ op.create_index(
+ "uq_shipment_exception_open",
+ "shipment_exceptions",
+ ["scope", "shipment_id", "kind"],
+ unique=True,
+ postgresql_where=sa.text("resolved_at IS NULL"),
+ sqlite_where=sa.text("resolved_at IS NULL"),
+ )
+ op.create_index(
+ "ix_shipment_exception_list",
+ "shipment_exceptions",
+ ["scope", "resolved_at", "first_observed_at"],
+ )
+
+
+def downgrade():
+ op.drop_table("shipment_exceptions")
+ op.drop_table("exception_evaluations")
+ op.drop_table("exception_policies")
diff --git a/backend/app/api/routes/exceptions.py b/backend/app/api/routes/exceptions.py
new file mode 100644
index 0000000..31a4e1f
--- /dev/null
+++ b/backend/app/api/routes/exceptions.py
@@ -0,0 +1,176 @@
+from datetime import datetime, timezone
+from typing import Literal
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from sqlalchemy.orm import Session, joinedload
+from app.core.auth import get_current_user
+from app.core.database import get_db
+from app.core.limiter import limiter
+from app.core.permissions import require_product_owner
+from app.models.shipment import Shipment
+from app.models.exception import ExceptionPolicy, ShipmentException, ExceptionEvaluation
+from app.schemas.exception import (
+ PolicyResponse,
+ PolicyUpdate,
+ Acknowledgement,
+ ExceptionResponse,
+ ExceptionList,
+)
+from app.services.identity import require_current_scope
+from app.services.exceptions import current_scope, policy_for, pause_reason, writer_lock
+from app.services.indexer import IndexerBusy
+from app.services.analytics import utc
+
+router = APIRouter()
+
+
+def health():
+ from app.main import indexer_state
+
+ return indexer_state().model_dump()
+
+
+def lock(db):
+ try:
+ writer_lock(db)
+ except IndexerBusy:
+ raise HTTPException(
+ 409, "Canonical evidence is being updated; retry shortly"
+ ) from None
+
+
+def owned_shipment(db, shipment_id, user):
+ shipment = db.query(Shipment).filter_by(shipment_id=shipment_id).first()
+ if not shipment:
+ raise HTTPException(404, "Shipment not found")
+ require_product_owner(shipment.product, user)
+ require_current_scope(shipment.product)
+ return shipment
+
+
+@router.get("/", response_model=ExceptionList)
+def list_exceptions(
+ state: Literal[
+ "open", "unacknowledged", "acknowledged", "resolved", "all"
+ ] = "open",
+ shipment_id: str | None = None,
+ skip: int = Query(0, ge=0),
+ limit: int = Query(20, ge=1, le=100),
+ db: Session = Depends(get_db),
+):
+ scope = current_scope()
+ query = (
+ db.query(ShipmentException)
+ .options(joinedload(ShipmentException.shipment).joinedload(Shipment.product))
+ .filter_by(scope=scope)
+ )
+ if state in {"open", "unacknowledged", "acknowledged"}:
+ query = query.filter(ShipmentException.resolved_at.is_(None))
+ if state == "resolved":
+ query = query.filter(ShipmentException.resolved_at.is_not(None))
+ if state == "unacknowledged":
+ query = query.filter(ShipmentException.acknowledged_at.is_(None))
+ if state == "acknowledged":
+ query = query.filter(ShipmentException.acknowledged_at.is_not(None))
+ if shipment_id:
+ query = query.filter_by(shipment_id=shipment_id)
+ previous = db.get(ExceptionEvaluation, scope)
+ reason = pause_reason(health())
+ if (
+ not reason
+ and previous
+ and utc(previous.evaluated_at) > datetime.now(timezone.utc)
+ ):
+ reason = "clock_regressed"
+ if not reason and (
+ not previous
+ or (datetime.now(timezone.utc) - utc(previous.evaluated_at)).total_seconds()
+ > 180
+ ):
+ reason = "evaluation_pending" if not previous else "evaluation_stale"
+ return dict(
+ items=query.order_by(
+ ShipmentException.first_observed_at.desc(), ShipmentException.id
+ )
+ .offset(skip)
+ .limit(limit)
+ .all(),
+ total=query.count(),
+ evaluation=dict(
+ paused=bool(reason),
+ reason=reason,
+ evaluated_at=previous.evaluated_at if previous else None,
+ block_number=previous.block_number if previous else None,
+ uncertain_shipments=previous.uncertain_shipments if previous else 0,
+ ),
+ )
+
+
+@router.get("/policies/{shipment_id}", response_model=PolicyResponse)
+def get_policy(shipment_id: str, db: Session = Depends(get_db)):
+ if not db.query(Shipment).filter_by(shipment_id=shipment_id).first():
+ raise HTTPException(404, "Shipment not found")
+ return policy_for(db, shipment_id)
+
+
+@router.put("/policies/{shipment_id}", response_model=PolicyResponse)
+@limiter.limit("30/minute")
+def update_policy(
+ request: Request,
+ shipment_id: str,
+ body: PolicyUpdate,
+ db: Session = Depends(get_db),
+ user: str = Depends(get_current_user),
+):
+ lock(db)
+ owned_shipment(db, shipment_id, user)
+ previous = policy_for(db, shipment_id)
+ if body.expected_revision != previous.revision:
+ raise HTTPException(409, "Policy changed; reload before saving")
+ values = body.model_dump(exclude={"expected_revision"})
+ # Identical saves must not invalidate acknowledgements or create new episodes.
+ if values == previous.model_dump(include=set(values)):
+ return previous
+ row = db.get(ExceptionPolicy, shipment_id)
+ if row is None:
+ row = ExceptionPolicy(shipment_id=shipment_id)
+ db.add(row)
+ for key, value in values.items():
+ setattr(row, key, value)
+ row.revision = previous.revision + 1
+ row.updated_at, row.updated_by = datetime.now(timezone.utc), user.lower()
+ db.commit()
+ db.refresh(row)
+ return row
+
+
+@router.post("/{exception_id}/acknowledgement", response_model=ExceptionResponse)
+@limiter.limit("30/minute")
+def acknowledge(
+ request: Request,
+ exception_id: str,
+ body: Acknowledgement,
+ db: Session = Depends(get_db),
+ user: str = Depends(get_current_user),
+):
+ lock(db)
+ row = (
+ db.query(ShipmentException)
+ .filter_by(id=exception_id, scope=current_scope())
+ .first()
+ )
+ if row is None:
+ raise HTTPException(404, "Exception not found")
+ owned_shipment(db, row.shipment_id, user)
+ if row.resolved_at:
+ raise HTTPException(
+ 409, "This occurrence has resolved; reload the exception list"
+ )
+ if row.acknowledged_at is None:
+ row.acknowledged_at, row.acknowledged_by = (
+ datetime.now(timezone.utc),
+ user.lower(),
+ )
+ row.acknowledgement_note = body.note.strip() if body.note else None
+ db.commit()
+ db.refresh(row)
+ return row
diff --git a/backend/app/main.py b/backend/app/main.py
index 3ed5f20..6834b5b 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -20,7 +20,8 @@
from app.core.limiter import limiter
from app.core.logging import LOGGING_CONFIG
from app.core.security import SecurityHeadersMiddleware
-from app.api.routes import auth, products, shipments, analytics
+from app.api.routes import auth, products, shipments, analytics, exceptions
+from app.services.exceptions import ExceptionWorker
from app.services.indexer import EventIndexer
from app.services.health import DatabaseProbe
from app.schemas.health import (
@@ -73,9 +74,15 @@ async def lifespan(app: FastAPI):
logger.error("Indexer thread startup failed: %s", _indexer_startup_error)
else:
logger.info("Blockchain indexing explicitly disabled")
+ exception_worker = None
+ if settings.indexer_enabled:
+ exception_worker = ExceptionWorker(lambda: indexer_state().model_dump())
+ exception_worker.start()
try:
yield
finally:
+ if exception_worker is not None:
+ exception_worker.stop()
if owned_indexer is not None:
owned_indexer.stop()
logger.info("Blockchain indexer shutdown requested")
@@ -158,6 +165,7 @@ async def log_requests(request: Request, call_next):
app.include_router(products.router, prefix="/products", tags=["products"])
app.include_router(shipments.router, prefix="/shipments", tags=["shipments"])
app.include_router(analytics.router, prefix="/analytics", tags=["analytics"])
+app.include_router(exceptions.router, prefix="/exceptions", tags=["exceptions"])
def indexer_state():
diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py
index 2a48210..ee7be78 100644
--- a/backend/app/models/__init__.py
+++ b/backend/app/models/__init__.py
@@ -6,3 +6,4 @@
from .chain import IndexerCursor, ChainBlock, ChainEvent, IndexedProduct
from .registration import RegistrationAttempt
+from .exception import ExceptionPolicy, ShipmentException, ExceptionEvaluation
diff --git a/backend/app/models/exception.py b/backend/app/models/exception.py
new file mode 100644
index 0000000..7350e9c
--- /dev/null
+++ b/backend/app/models/exception.py
@@ -0,0 +1,91 @@
+"""Off-chain operational observations, separate from canonical chain evidence."""
+
+from sqlalchemy import (
+ Column,
+ String,
+ Integer,
+ Float,
+ Boolean,
+ DateTime,
+ ForeignKey,
+ Index,
+ JSON,
+ CheckConstraint,
+ BigInteger,
+)
+from app.core.database import Base
+from sqlalchemy.orm import relationship
+
+
+class ExceptionPolicy(Base):
+ __tablename__ = "exception_policies"
+ shipment_id = Column(String, ForeignKey("shipments.shipment_id"), primary_key=True)
+ revision = Column(Integer, nullable=False)
+ arrival_enabled = Column(Boolean, nullable=False)
+ inactivity_hours = Column(Float, nullable=True)
+ checkpoint_hours = Column(Float, nullable=True)
+ custody_hours = Column(Float, nullable=True)
+ updated_at = Column(DateTime(timezone=True), nullable=False)
+ updated_by = Column(String(42), nullable=False)
+ __table_args__ = tuple(
+ CheckConstraint(
+ f"{name} IS NULL OR ({name} >= 0.25 AND {name} <= 8760)",
+ name=f"ck_exception_policy_{name}",
+ )
+ for name in ("inactivity_hours", "checkpoint_hours", "custody_hours")
+ )
+
+
+class ShipmentException(Base):
+ __tablename__ = "shipment_exceptions"
+ id = Column(String(36), primary_key=True)
+ scope = Column(String(64), nullable=False)
+ shipment_id = Column(String, ForeignKey("shipments.shipment_id"), nullable=False)
+ kind = Column(String(32), nullable=False)
+ fingerprint = Column(String(64), nullable=False)
+ triggered_at = Column(DateTime(timezone=True), nullable=False)
+ first_observed_at = Column(DateTime(timezone=True), nullable=False)
+ last_observed_at = Column(DateTime(timezone=True), nullable=False)
+ resolved_at = Column(DateTime(timezone=True), nullable=True)
+ resolution = Column(String(40), nullable=True)
+ acknowledged_at = Column(DateTime(timezone=True), nullable=True)
+ acknowledged_by = Column(String(42), nullable=True)
+ acknowledgement_note = Column(String(500), nullable=True)
+ evidence = Column(JSON, nullable=False)
+ shipment = relationship("Shipment")
+
+ @property
+ def owner_address(self):
+ return self.shipment.product.owner_address
+
+ @property
+ def ownership_source(self):
+ return self.shipment.product.ownership_source
+
+ __table_args__ = (
+ Index(
+ "uq_shipment_exception_open",
+ "scope",
+ "shipment_id",
+ "kind",
+ unique=True,
+ postgresql_where=resolved_at.is_(None),
+ sqlite_where=resolved_at.is_(None),
+ ),
+ Index(
+ "ix_shipment_exception_list", "scope", "resolved_at", "first_observed_at"
+ ),
+ CheckConstraint(
+ "kind IN ('arrival_overdue','checkpoint_inactivity','checkpoint_wait','custody_followup')",
+ name="ck_shipment_exception_kind",
+ ),
+ )
+
+
+class ExceptionEvaluation(Base):
+ __tablename__ = "exception_evaluations"
+ scope = Column(String(64), primary_key=True)
+ evaluated_at = Column(DateTime(timezone=True), nullable=False)
+ block_number = Column(BigInteger, nullable=False)
+ block_hash = Column(String(66), nullable=False)
+ uncertain_shipments = Column(Integer, nullable=False, default=0)
diff --git a/backend/app/schemas/exception.py b/backend/app/schemas/exception.py
new file mode 100644
index 0000000..9ffc224
--- /dev/null
+++ b/backend/app/schemas/exception.py
@@ -0,0 +1,79 @@
+from datetime import datetime
+from typing import Literal
+from pydantic import BaseModel, ConfigDict, Field
+
+ExceptionKind = Literal[
+ "arrival_overdue", "checkpoint_inactivity", "checkpoint_wait", "custody_followup"
+]
+
+
+class PolicyValues(BaseModel):
+ model_config = ConfigDict(extra="forbid")
+ arrival_enabled: bool = True
+ inactivity_hours: float | None = Field(
+ default=24.0, ge=0.25, le=8760, allow_inf_nan=False
+ )
+ checkpoint_hours: float | None = Field(
+ default=24.0, ge=0.25, le=8760, allow_inf_nan=False
+ )
+ custody_hours: float | None = Field(
+ default=6.0, ge=0.25, le=8760, allow_inf_nan=False
+ )
+
+
+class PolicyUpdate(PolicyValues):
+ expected_revision: int = Field(ge=0)
+
+
+class PolicyResponse(PolicyValues):
+ model_config = ConfigDict(from_attributes=True)
+ revision: int = 0
+ updated_at: datetime | None = None
+ updated_by: str | None = None
+
+
+class Acknowledgement(BaseModel):
+ model_config = ConfigDict(extra="forbid")
+ note: str | None = Field(default=None, max_length=500)
+
+
+class ExceptionEvidence(BaseModel):
+ product_id: str
+ anchor_event_id: str
+ anchor_occurrence_id: str
+ threshold_hours: float | None
+ location: str | None = None
+ receiving_handler: str | None = None
+ policy_revision: int
+
+
+class ExceptionResponse(BaseModel):
+ model_config = ConfigDict(from_attributes=True)
+ id: str
+ shipment_id: str
+ kind: ExceptionKind
+ owner_address: str | None
+ ownership_source: str | None
+ triggered_at: datetime
+ first_observed_at: datetime
+ last_observed_at: datetime
+ resolved_at: datetime | None
+ resolution: str | None
+ acknowledged_at: datetime | None
+ acknowledged_by: str | None
+ acknowledgement_note: str | None
+ evidence: ExceptionEvidence
+
+
+class EvaluationStatus(BaseModel):
+ paused: bool
+ reason: str | None
+ evaluated_at: datetime | None
+ block_number: int | None
+ uncertain_shipments: int = 0
+
+
+class ExceptionList(BaseModel):
+ items: list[ExceptionResponse]
+ total: int
+ evaluation: EvaluationStatus
diff --git a/backend/app/services/exceptions.py b/backend/app/services/exceptions.py
new file mode 100644
index 0000000..62610d3
--- /dev/null
+++ b/backend/app/services/exceptions.py
@@ -0,0 +1,294 @@
+"""Reproducible, clock-injected rules and atomic off-chain exception episodes.
+
+No RPC, writes on GET, or notification delivery. All writers use the indexer's
+transaction lock so reorgs and concurrent evaluators cannot mix observations.
+"""
+
+import hashlib
+import logging
+import threading
+import uuid
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import text
+
+from app.core.config import settings
+from app.core.database import SessionLocal
+from app.models.chain import IndexerCursor, ChainBlock
+from app.models.exception import ExceptionPolicy, ShipmentException, ExceptionEvaluation
+from app.schemas.exception import PolicyResponse
+from app.services.analytics import shipment_dataset, utc
+from app.services.identity import require_current_scope
+from app.services.indexer import WRITER_LOCK, IndexerBusy
+from fastapi import HTTPException
+
+logger = logging.getLogger("veritras.exceptions")
+
+
+def current_scope():
+ value = f"{settings.chain_id}:{settings.product_registry_contract.lower()}:{settings.shipment_tracker_contract.lower()}"
+ return hashlib.sha256(value.encode()).hexdigest()
+
+
+def writer_lock(db):
+ if db.get_bind().dialect.name == "postgresql":
+ if not db.execute(
+ text("SELECT pg_try_advisory_xact_lock(:key)"), {"key": WRITER_LOCK}
+ ).scalar():
+ raise IndexerBusy("Projection writer is busy")
+ else:
+ db.execute(text("BEGIN IMMEDIATE"))
+
+
+def pause_reason(health):
+ if not health.get("enabled"):
+ return "indexing_disabled"
+ if not health.get("ready") or health.get("status") != "current":
+ return "indexer_not_current"
+ if health.get("lag_blocks") != 0 or health.get("finality_gap_blocks") != 0:
+ return "indexer_not_current"
+ if health.get("last_processed_block") is None:
+ return "cursor_unavailable"
+ return None
+
+
+def policy_for(db, shipment_id):
+ row = db.get(ExceptionPolicy, shipment_id)
+ return PolicyResponse.model_validate(row) if row else PolicyResponse()
+
+
+def event_time(event):
+ try:
+ return datetime.fromtimestamp(int(event.payload["timestamp"]), timezone.utc)
+ except (KeyError, ValueError, TypeError, OverflowError, OSError):
+ return None
+
+
+def conditions(record, policy, now):
+ """None means uncertain evidence: retain existing observations for review."""
+ shipment = record["shipment"]
+ try:
+ require_current_scope(shipment.product)
+ except HTTPException:
+ return {}
+ events = record["events"]
+ if not any(
+ e.event_type == "ProductRegistered" and e.scope == current_scope()
+ for e in events
+ ):
+ return {}
+ timed = [
+ e
+ for e in events
+ if e.event_type
+ in {"CheckpointRecorded", "CustodyTransferred", "DeliveryCompleted"}
+ ]
+ if any(event_time(e) is None or event_time(e) > now for e in timed):
+ return None
+ if record["delivered_at"]:
+ return {}
+ result = {}
+
+ def add(kind, anchor, due, threshold=None, **details):
+ # Exact threshold is overdue; no alerts before it.
+ if now < due:
+ return
+ evidence = dict(
+ product_id=shipment.product_id,
+ anchor_event_id=anchor.event_id,
+ anchor_occurrence_id=anchor.occurrence_id,
+ threshold_hours=threshold,
+ policy_revision=policy.revision,
+ **details,
+ )
+ identity = f"{kind}:{anchor.occurrence_id}:{threshold}:{due.isoformat()}"
+ result[kind] = dict(
+ fingerprint=hashlib.sha256(identity.encode()).hexdigest(),
+ triggered_at=due,
+ evidence=evidence,
+ )
+
+ departure = record["departed_at"]
+ if (
+ policy.arrival_enabled
+ and departure
+ and shipment.expected_arrival_at
+ and shipment.expected_arrival_recorded_at
+ and utc(shipment.expected_arrival_recorded_at) <= departure
+ ):
+ anchor = next(
+ e
+ for e in events
+ if e.event_type == "CheckpointRecorded"
+ and str(e.payload.get("status")) == "1"
+ )
+ add("arrival_overdue", anchor, utc(shipment.expected_arrival_at))
+ checkpoints = [e for e in events if e.event_type == "CheckpointRecorded"]
+ if checkpoints:
+ latest = checkpoints[-1]
+ at_checkpoint = str(latest.payload.get("status")) == "2"
+ threshold = (
+ policy.checkpoint_hours if at_checkpoint else policy.inactivity_hours
+ )
+ # Created-only checkpoints do not establish an active shipment journey.
+ if threshold is not None and (departure or at_checkpoint):
+ add(
+ "checkpoint_wait" if at_checkpoint else "checkpoint_inactivity",
+ latest,
+ event_time(latest) + timedelta(hours=threshold),
+ threshold,
+ location=str(latest.payload.get("location", "")),
+ )
+ transfers = [
+ i for i, e in enumerate(events) if e.event_type == "CustodyTransferred"
+ ]
+ if transfers and policy.custody_hours is not None:
+ position = transfers[-1]
+ transfer = events[position]
+ recipient = str(transfer.payload.get("to", "")).lower()
+ if not recipient or recipient == "0x" + "0" * 40:
+ return None
+ followed_up = any(
+ e.event_type == "CheckpointRecorded"
+ and str(e.payload.get("handler", "")).lower() == recipient
+ for e in events[position + 1 :]
+ )
+ if not followed_up:
+ add(
+ "custody_followup",
+ transfer,
+ event_time(transfer) + timedelta(hours=policy.custody_hours),
+ policy.custody_hours,
+ receiving_handler=recipient,
+ )
+ return result
+
+
+def evaluate(db, health_provider, *, now=None):
+ """Commit the entire evaluation or nothing. Caller supplies a fresh session."""
+ now = utc(now or datetime.now(timezone.utc))
+ before = health_provider()
+ reason = pause_reason(before)
+ if reason:
+ return reason
+ try:
+ writer_lock(db)
+ scope = current_scope()
+ cursor = db.get(IndexerCursor, scope)
+ if not cursor or cursor.last_processed_block != before["last_processed_block"]:
+ db.rollback()
+ return "cursor_changed"
+ anchor = db.get(ChainBlock, (scope, cursor.last_processed_block))
+ if not anchor or anchor.block_hash != cursor.block_hash:
+ db.rollback()
+ return "cursor_unavailable"
+ previous = db.get(ExceptionEvaluation, scope)
+ if previous and utc(previous.evaluated_at) > now:
+ db.rollback()
+ return "clock_regressed"
+ records = shipment_dataset(db)
+ policies = {
+ p.shipment_id: PolicyResponse.model_validate(p)
+ for p in db.query(ExceptionPolicy).all()
+ }
+ open_rows = (
+ db.query(ShipmentException).filter_by(scope=scope, resolved_at=None).all()
+ )
+ active = {(r.shipment_id, r.kind): r for r in open_rows}
+ seen = set()
+ uncertain = set()
+ canonical_anchors = {}
+ for record in records:
+ shipment_id = record["shipment"].shipment_id
+ canonical_anchors[shipment_id] = {e.occurrence_id for e in record["events"]}
+ expected = conditions(
+ record, policies.get(shipment_id, PolicyResponse()), now
+ )
+ if expected is None:
+ uncertain.add(shipment_id)
+ continue
+ for kind, values in expected.items():
+ key = (shipment_id, kind)
+ seen.add(key)
+ row = active.get(key)
+ if row and row.fingerprint != values["fingerprint"]:
+ row.resolved_at, row.resolution = now, "condition_replaced"
+ db.flush() # Release partial unique index before replacement.
+ row = None
+ if row is None:
+ db.add(
+ ShipmentException(
+ id=str(uuid.uuid4()),
+ scope=scope,
+ shipment_id=shipment_id,
+ kind=kind,
+ first_observed_at=now,
+ last_observed_at=now,
+ **values,
+ )
+ )
+ else:
+ row.last_observed_at = now
+ row.evidence = values["evidence"]
+ for key, row in active.items():
+ if key not in seen and row.shipment_id not in uncertain:
+ row.resolved_at = now
+ row.resolution = (
+ "evidence_withdrawn"
+ if row.evidence["anchor_occurrence_id"]
+ not in canonical_anchors.get(row.shipment_id, set())
+ else "condition_cleared"
+ )
+ after = health_provider()
+ if (
+ pause_reason(after)
+ or after["last_processed_block"] != before["last_processed_block"]
+ ):
+ db.rollback()
+ return "indexer_changed"
+ if previous is None:
+ previous = ExceptionEvaluation(scope=scope)
+ db.add(previous)
+ previous.evaluated_at, previous.block_number, previous.block_hash = (
+ now,
+ cursor.last_processed_block,
+ cursor.block_hash,
+ )
+ previous.uncertain_shipments = len(uncertain)
+ db.commit()
+ return None
+ except Exception:
+ db.rollback()
+ raise
+
+
+class ExceptionWorker:
+ def __init__(self, health_provider, session_factory=SessionLocal, interval=60):
+ self.health_provider, self.session_factory, self.interval = (
+ health_provider,
+ session_factory,
+ interval,
+ )
+ self.stopping = threading.Event()
+ self.thread = threading.Thread(
+ target=self.run, name="shipment-exceptions", daemon=True
+ )
+
+ def start(self):
+ self.thread.start()
+
+ def stop(self):
+ self.stopping.set()
+ self.thread.join(timeout=5)
+
+ def run(self):
+ while not self.stopping.is_set():
+ try:
+ if not pause_reason(self.health_provider()):
+ with self.session_factory() as db:
+ evaluate(db, self.health_provider)
+ except IndexerBusy:
+ pass # Next tick retries; canonical ingestion always has priority.
+ except Exception as error:
+ logger.error("Exception evaluation failed: %s", type(error).__name__)
+ self.stopping.wait(self.interval)
diff --git a/backend/rehearse_local_release.py b/backend/rehearse_local_release.py
index 50b9705..750b92b 100644
--- a/backend/rehearse_local_release.py
+++ b/backend/rehearse_local_release.py
@@ -22,6 +22,9 @@
"indexed_products",
"product_aliases",
"registration_attempts",
+ "exception_policies",
+ "shipment_exceptions",
+ "exception_evaluations",
"alembic_version",
)
diff --git a/backend/tests/test_exception_storage.py b/backend/tests/test_exception_storage.py
new file mode 100644
index 0000000..06ce811
--- /dev/null
+++ b/backend/tests/test_exception_storage.py
@@ -0,0 +1,169 @@
+"""Storage regression uses a new, explicitly owned schema per PG test."""
+
+import os
+import uuid
+import threading
+from datetime import timedelta
+from concurrent.futures import ThreadPoolExecutor
+import pytest
+import sqlalchemy as sa
+from sqlalchemy.engine import make_url
+from sqlalchemy.orm import sessionmaker
+from alembic import command
+from alembic.config import Config
+from app.core.database import Base
+from app.models.exception import ShipmentException, ExceptionEvaluation
+from app.services.exceptions import evaluate, writer_lock
+from app.services.indexer import IndexerBusy
+from tests.test_exceptions import seed, HEALTH, START
+
+
+@pytest.fixture(params=["sqlite", "postgres"])
+def storage(request, tmp_path):
+ admin = None
+ if request.param == "postgres":
+ url = os.environ.get("INDEXER_TEST_DATABASE_URL")
+ if not url:
+ pytest.skip("Dedicated PostgreSQL regression database not configured")
+ parsed = make_url(url)
+ assert parsed.database == "veritras_indexer_regression" and parsed.host in (
+ "localhost",
+ "127.0.0.1",
+ "postgres",
+ )
+ # Never reflect or remove tables from a pre-existing schema.
+ schema = "exception_test_" + uuid.uuid4().hex
+ admin = sa.create_engine(url)
+ with admin.begin() as conn:
+ conn.execute(sa.text(f'CREATE SCHEMA "{schema}"'))
+ url = parsed.update_query_dict({"options": f"-csearch_path={schema}"})
+ else:
+ url = f'sqlite:///{tmp_path / "exceptions.db"}'
+ engine = sa.create_engine(url)
+ try:
+ yield engine, sessionmaker(engine, autoflush=False)
+ finally:
+ engine.dispose()
+ if admin:
+ # This random schema was created above by this fixture only.
+ with admin.begin() as conn:
+ conn.execute(sa.text(f'DROP SCHEMA "{schema}" CASCADE'))
+ admin.dispose()
+
+
+def test_additive_migration_preserves_data_and_matches_metadata(storage):
+ engine, sessions = storage
+ config = Config("alembic.ini")
+ config.attributes["configure_logger"] = False
+ with engine.begin() as conn:
+ config.attributes["connection"] = conn
+ command.upgrade(config, "c62a904df318")
+ with sessions() as db:
+ seed(db)
+ with engine.begin() as conn:
+ config.attributes["connection"] = conn
+ command.upgrade(config, "head")
+ with sessions() as db:
+ assert db.execute(sa.text("SELECT count(*) FROM shipments")).scalar() == 1
+ assert db.execute(sa.text("SELECT count(*) FROM chain_events")).scalar() == 2
+ assert db.query(ShipmentException).count() == 0
+ assert (
+ db.execute(sa.text("SELECT version_num FROM alembic_version")).scalar()
+ == "e93a107bd624"
+ )
+ db.rollback()
+ assert evaluate(db, lambda: HEALTH, now=START + timedelta(hours=25)) is None
+ inspector = sa.inspect(engine)
+ for table in ("exception_policies", "exception_evaluations", "shipment_exceptions"):
+ assert {c["name"] for c in inspector.get_columns(table)} == set(
+ Base.metadata.tables[table].columns.keys()
+ )
+ assert any(
+ i["name"] == "uq_shipment_exception_open" and i["unique"]
+ for i in inspector.get_indexes("shipment_exceptions")
+ )
+
+
+def test_restart_reuses_episode_and_acknowledgement(storage):
+ engine, sessions = storage
+ Base.metadata.create_all(engine)
+ with sessions() as db:
+ seed(db)
+ evaluate(db, lambda: HEALTH, now=START + timedelta(hours=25))
+ row = db.query(ShipmentException).first()
+ identity = row.id
+ row.acknowledged_at = START
+ row.acknowledged_by = "0x" + "1" * 40
+ db.commit()
+ with sessions() as db:
+ evaluate(db, lambda: HEALTH, now=START + timedelta(hours=26))
+ assert db.query(ShipmentException).count() == 2
+ assert db.get(ShipmentException, identity).acknowledged_at is not None
+
+
+def test_interrupted_commit_rolls_back_entire_assessment(storage):
+ engine, sessions = storage
+ Base.metadata.create_all(engine)
+ with sessions() as db:
+ seed(db)
+ with sessions() as db:
+
+ def fail(session):
+ raise RuntimeError("simulated interruption")
+
+ sa.event.listen(db, "before_commit", fail)
+ with pytest.raises(RuntimeError):
+ evaluate(db, lambda: HEALTH, now=START + timedelta(hours=25))
+ with sessions() as db:
+ assert db.query(ShipmentException).count() == 0
+ assert db.query(ExceptionEvaluation).count() == 0
+ db.rollback()
+ evaluate(db, lambda: HEALTH, now=START + timedelta(hours=25))
+ assert db.query(ShipmentException).count() == 2
+
+
+def test_concurrent_evaluators_deduplicate(storage):
+ engine, sessions = storage
+ Base.metadata.create_all(engine)
+ with sessions() as db:
+ seed(db)
+ barrier = threading.Barrier(2)
+
+ def task(_):
+ first = True
+
+ def health():
+ nonlocal first
+ if first:
+ first = False
+ barrier.wait(timeout=5)
+ return HEALTH
+
+ with sessions() as db:
+ try:
+ return evaluate(db, health, now=START + timedelta(hours=25))
+ except IndexerBusy:
+ return "busy"
+
+ with ThreadPoolExecutor(max_workers=2) as executor:
+ results = list(executor.map(task, range(2)))
+ assert None in results and set(results) <= {None, "busy"}
+ with sessions() as db:
+ assert db.query(ShipmentException).count() == 2
+
+
+def test_indexer_lock_blocks_exception_writer(storage):
+ engine, sessions = storage
+ if engine.dialect.name != "postgresql":
+ pytest.skip("PostgreSQL advisory lock behavior")
+ Base.metadata.create_all(engine)
+ with sessions() as db:
+ seed(db)
+ with sessions() as owner, sessions() as contender:
+ writer_lock(owner)
+ with pytest.raises(IndexerBusy):
+ evaluate(contender, lambda: HEALTH, now=START + timedelta(hours=25))
+ owner.rollback()
+ assert (
+ evaluate(contender, lambda: HEALTH, now=START + timedelta(hours=25)) is None
+ )
diff --git a/backend/tests/test_exceptions.py b/backend/tests/test_exceptions.py
new file mode 100644
index 0000000..bb6c57a
--- /dev/null
+++ b/backend/tests/test_exceptions.py
@@ -0,0 +1,466 @@
+"""Controlled clocks, canonical recovery and off-chain permission boundaries."""
+
+from copy import deepcopy
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+import pytest
+from sqlalchemy import event as sql_event
+from app.core.auth import create_access_token
+from app.models.chain import ChainBlock, ChainEvent, IndexedProduct, IndexerCursor
+from app.models.exception import ShipmentException, ExceptionEvaluation
+from app.models.product import Product
+from app.models.shipment import Shipment
+from app.schemas.exception import PolicyResponse
+from app.services.analytics import shipment_dataset, utc
+from app.services.exceptions import conditions, evaluate, current_scope, ExceptionWorker
+
+START = datetime(2026, 1, 1, tzinfo=timezone.utc)
+OWNER = "0x1234567890123456789012345678901234567890"
+HANDLER = "0x" + "5" * 40
+HEALTH = dict(
+ enabled=True,
+ ready=True,
+ status="current",
+ last_processed_block=100,
+ lag_blocks=0,
+ finality_gap_blocks=0,
+)
+
+
+def seed(db):
+ from app.services.identity import scope_values
+
+ p = Product(
+ product_id="PROD-EXCEPTION",
+ name="Coffee",
+ manufacturer_address=OWNER,
+ owner_address=OWNER,
+ ownership_source="chain_event",
+ registration_status="indexed",
+ registered_at=START,
+ block_number=1,
+ tx_hash="0x01",
+ **scope_values("PROD-EXCEPTION"),
+ )
+ db.add(p)
+ db.flush()
+ s = Shipment(
+ shipment_id="SHIP-EXCEPTION",
+ product_id=p.product_id,
+ origin="Manila",
+ destination="Cebu",
+ expected_arrival_at=START + timedelta(hours=8),
+ expected_arrival_recorded_at=START - timedelta(hours=1),
+ )
+ db.add_all(
+ [
+ s,
+ IndexedProduct(
+ scope=current_scope(),
+ product_key=p.on_chain_product_id,
+ product_id=p.product_id,
+ baseline={},
+ status="1",
+ ),
+ IndexerCursor(
+ scope=current_scope(),
+ chain_id=p.chain_id,
+ registry_address=p.registry_address,
+ tracker_address=p.tracker_address,
+ start_block=0,
+ last_processed_block=100,
+ block_hash="0xabc",
+ updated_at=START,
+ ),
+ ChainBlock(scope=current_scope(), number=100, block_hash="0xabc"),
+ ]
+ )
+ db.flush()
+ add_event(db, 1, "ProductRegistered", registry=True)
+ add_event(db, 2, "CheckpointRecorded", status=1, handler=OWNER, location="Manila")
+ db.commit()
+ return s
+
+
+def add_event(db, n, kind, *, at=START, registry=False, **payload):
+ p = db.query(Product).filter_by(product_id="PROD-EXCEPTION").one()
+ e = ChainEvent(
+ occurrence_id=str(n),
+ event_id=str(n),
+ scope=current_scope(),
+ chain_id=p.chain_id,
+ product_key=p.on_chain_product_id,
+ contract_address=p.registry_address if registry else p.tracker_address,
+ transaction_hash="0x" + f"{n:064x}",
+ block_hash="0x" + f"{n:064x}",
+ block_number=n,
+ transaction_index=0,
+ log_index=0,
+ event_type=kind,
+ canonical=True,
+ processed_at=START,
+ payload={"timestamp": int(at.timestamp()), **payload},
+ )
+ db.add(e)
+ return e
+
+
+def run(db, hours=25, health=None):
+ db.commit() # Production evaluator always starts with a fresh transaction.
+ return evaluate(
+ db,
+ lambda: deepcopy(HEALTH if health is None else health),
+ now=START + timedelta(hours=hours),
+ )
+
+
+def open_rows(db):
+ return (
+ db.query(ShipmentException)
+ .filter_by(resolved_at=None)
+ .order_by(ShipmentException.kind)
+ .all()
+ )
+
+
+@pytest.fixture
+def shipment(db_session):
+ seed(db_session)
+ return db_session
+
+
+def test_threshold_boundaries_repeat_evaluation_and_restart(shipment):
+ db = shipment
+ assert run(db, 7.999) is None
+ assert open_rows(db) == []
+ assert run(db, 8) is None
+ first = open_rows(db)[0]
+ assert first.kind == "arrival_overdue"
+ arrival_id = first.id
+ run(db, 24)
+ assert {e.kind for e in open_rows(db)} == {
+ "arrival_overdue",
+ "checkpoint_inactivity",
+ }
+ run(db, 25)
+ db.expire_all()
+ assert db.query(ShipmentException).count() == 2
+ first = db.get(ShipmentException, arrival_id)
+ assert utc(first.triggered_at) == START + timedelta(hours=8)
+ assert utc(first.first_observed_at) == START + timedelta(hours=8)
+ assert utc(first.last_observed_at) == START + timedelta(hours=25)
+
+
+@pytest.mark.parametrize(
+ "health",
+ [
+ dict(HEALTH, enabled=False),
+ dict(HEALTH, status="catching_up"),
+ dict(HEALTH, ready=False),
+ dict(HEALTH, lag_blocks=1),
+ dict(HEALTH, finality_gap_blocks=1),
+ dict(HEALTH, last_processed_block=None),
+ ],
+)
+def test_paused_indexer_neither_opens_nor_resolves(shipment, health):
+ db = shipment
+ assert run(db, 25, health)
+ assert db.query(ShipmentException).count() == 0
+ run(db)
+ add_event(db, 3, "DeliveryCompleted", at=START + timedelta(hours=26))
+ assert run(db, 27, health)
+ assert len(open_rows(db)) == 2
+ assert run(db, 27) is None
+ assert open_rows(db) == []
+
+
+def test_custody_followup_uses_source_order_and_receiving_handler(shipment):
+ db = shipment
+ add_event(db, 3, "CustodyTransferred", **{"from": OWNER, "to": HANDLER})
+ run(db, 6)
+ row = open_rows(db)[0]
+ assert row.kind == "custody_followup"
+ occurrence = row.id
+ add_event(db, 4, "CheckpointRecorded", status=2, handler=OWNER, location="Hub")
+ run(db, 6)
+ assert db.get(ShipmentException, occurrence).resolved_at is None
+ add_event(db, 5, "CheckpointRecorded", status=2, handler=HANDLER, location="Hub")
+ run(db, 6)
+ assert db.get(ShipmentException, occurrence).resolved_at is not None
+ run(db, 25)
+ assert {e.kind for e in open_rows(db)} == {"arrival_overdue", "checkpoint_wait"}
+
+
+def test_reorg_resolves_and_reopens_new_unacknowledged_episode(shipment):
+ db = shipment
+ run(db)
+ old = open_rows(db)[0]
+ old.acknowledged_at = START + timedelta(hours=25)
+ old.acknowledged_by = OWNER
+ add_event(db, 3, "DeliveryCompleted", at=START + timedelta(hours=26))
+ run(db, 27)
+ assert not open_rows(db)
+ db.get(ChainEvent, "3").canonical = False
+ run(db, 28)
+ reopened = open_rows(db)
+ assert len(reopened) == 2
+ assert all(e.id != old.id and e.acknowledged_at is None for e in reopened)
+ assert db.get(ShipmentException, old.id).acknowledged_by == OWNER
+
+
+@pytest.mark.parametrize(
+ "change",
+ [
+ {"scope": "other"},
+ {"canonical": False},
+ {"chain_id": 1},
+ {"contract_address": "0x" + "9" * 40},
+ {"product_key": "0x" + "f" * 64},
+ ],
+)
+def test_wrong_identity_or_orphan_cannot_clear_exceptions(shipment, change):
+ db = shipment
+ run(db)
+ e = add_event(db, 3, "DeliveryCompleted", at=START + timedelta(hours=26))
+ for key, value in change.items():
+ setattr(e, key, value)
+ run(db, 27)
+ assert len(open_rows(db)) == 2
+
+
+@pytest.mark.parametrize(
+ "timestamp",
+ [
+ None,
+ "invalid",
+ 999999999999999999999999,
+ int((START + timedelta(days=100)).timestamp()),
+ ],
+)
+def test_uncertain_timestamps_preserve_prior_observations(shipment, timestamp):
+ db = shipment
+ run(db)
+ add_event(
+ db, 3, "CheckpointRecorded", timestamp=timestamp, status=2, handler=HANDLER
+ )
+ run(db, 26)
+ assert len(open_rows(db)) == 2
+ assert all(
+ utc(e.last_observed_at) == START + timedelta(hours=25) for e in open_rows(db)
+ )
+
+
+def test_no_arrival_exception_for_plan_recorded_after_departure(shipment):
+ db = shipment
+ db.query(Shipment).one().expected_arrival_recorded_at = START + timedelta(seconds=1)
+ run(db)
+ assert [e.kind for e in open_rows(db)] == ["checkpoint_inactivity"]
+
+
+def test_atomic_failure_rolls_back_episodes_and_evaluation(shipment):
+ db = shipment
+
+ def fail(*args):
+ raise RuntimeError("interrupted write")
+
+ sql_event.listen(db, "before_commit", fail)
+ with pytest.raises(RuntimeError):
+ evaluate(db, lambda: HEALTH, now=START + timedelta(hours=25))
+ sql_event.remove(db, "before_commit", fail)
+ db.rollback()
+ assert db.query(ShipmentException).count() == 0
+ assert db.query(ExceptionEvaluation).count() == 0
+ run(db)
+ assert len(open_rows(db)) == 2
+
+
+def test_health_change_mid_evaluation_rolls_back(shipment):
+ db = shipment
+ db.commit()
+ values = iter([HEALTH, dict(HEALTH, status="degraded")])
+ assert (
+ evaluate(db, lambda: next(values), now=START + timedelta(hours=25))
+ == "indexer_changed"
+ )
+ assert not open_rows(db)
+
+
+def test_cursor_anchor_and_clock_guards(shipment):
+ db = shipment
+ run(db)
+ assert run(db, 24) == "clock_regressed"
+ db.get(ChainBlock, (current_scope(), 100)).block_hash = "changed"
+ assert run(db, 26) == "cursor_unavailable"
+ assert len(open_rows(db)) == 2
+ db.get(IndexerCursor, current_scope()).last_processed_block = 101
+ assert run(db, 26) == "cursor_changed"
+
+
+def test_owner_policy_revision_validation_and_episode_replacement(
+ client_fixture, db_session
+):
+ seed(db_session)
+ db = db_session
+ run(db)
+ old = [e.id for e in open_rows(db)]
+ path = "/exceptions/policies/SHIP-EXCEPTION"
+ assert client_fixture.get(path).json()["revision"] == 0
+ body = dict(
+ expected_revision=0,
+ arrival_enabled=True,
+ inactivity_hours=30,
+ checkpoint_hours=None,
+ custody_hours=6,
+ )
+ response = client_fixture.put(path, json=body)
+ assert response.status_code == 200, response.text
+ assert response.json()["revision"] == 1
+ assert client_fixture.put(path, json=body).status_code == 409
+ body["expected_revision"] = 1
+ assert client_fixture.put(path, json=body).json()["revision"] == 1
+ run(db, 26)
+ assert len(open_rows(db)) == 1
+ assert (
+ open_rows(db)[0].id in old
+ ) # An unrelated rule change preserves its acknowledgement.
+ assert sum(bool(db.get(ShipmentException, id).resolved_at) for id in old) == 1
+ for invalid in [-1, 0, 0.1, 8761]:
+ assert (
+ client_fixture.put(path, json=dict(body, custody_hours=invalid)).status_code
+ == 422
+ )
+ assert client_fixture.put(path, json=dict(body, unexpected="x")).status_code == 422
+
+
+def test_ack_permissions_idempotency_and_resolved_conflict(client_fixture, db_session):
+ seed(db_session)
+ run(db_session)
+ row = open_rows(db_session)[0]
+ path = f"/exceptions/{row.id}/acknowledgement"
+ client_fixture.headers.pop("Authorization")
+ assert client_fixture.post(path, json={}).status_code == 401
+ client_fixture.headers["Authorization"] = "Bearer " + create_access_token(HANDLER)
+ assert client_fixture.post(path, json={}).status_code == 403
+ assert (
+ client_fixture.put(
+ "/exceptions/policies/SHIP-EXCEPTION", json={"expected_revision": 0}
+ ).status_code
+ == 403
+ )
+ client_fixture.headers["Authorization"] = "Bearer " + create_access_token(OWNER)
+ first = client_fixture.post(path, json={"note": "Checked with carrier"})
+ assert first.status_code == 200, first.text
+ assert client_fixture.post(path, json={"note": "Overwrite"}).json() == first.json()
+ assert client_fixture.post(path, json={"note": "x" * 501}).status_code == 422
+ db_session.expire_all()
+ row = db_session.get(ShipmentException, row.id)
+ row.resolved_at = START + timedelta(hours=26)
+ db_session.commit()
+ assert client_fixture.post(path, json={}).status_code == 409
+ assert (
+ client_fixture.post("/exceptions/missing/acknowledgement", json={}).status_code
+ == 404
+ )
+
+
+def test_get_is_read_only_filters_history_pagination_and_scope(
+ client_fixture, db_session, monkeypatch
+):
+ from app.api.routes import exceptions
+
+ seed(db_session)
+ monkeypatch.setattr(exceptions, "health", lambda: dict(HEALTH, enabled=False))
+ assert client_fixture.get("/exceptions").json()["evaluation"]["paused"]
+ assert db_session.query(ExceptionEvaluation).count() == 0
+ run(db_session)
+ rows = open_rows(db_session)
+ rows[0].acknowledged_at = START
+ db_session.commit()
+ assert client_fixture.get("/exceptions?state=acknowledged").json()["total"] == 1
+ assert client_fixture.get("/exceptions?state=unacknowledged").json()["total"] == 1
+ assert len(client_fixture.get("/exceptions?limit=1").json()["items"]) == 1
+ assert client_fixture.get("/exceptions?skip=1&limit=1").json()["total"] == 2
+ assert client_fixture.get("/exceptions?shipment_id=other").json()["total"] == 0
+ assert client_fixture.get("/exceptions?state=bad").status_code == 422
+ assert client_fixture.get("/exceptions?limit=101").status_code == 422
+ add_event(db_session, 3, "DeliveryCompleted", at=START + timedelta(hours=26))
+ run(db_session, 27)
+ assert client_fixture.get("/exceptions?state=open").json()["total"] == 0
+ assert client_fixture.get("/exceptions?state=resolved").json()["total"] == 2
+ assert client_fixture.get("/exceptions/policies/missing").status_code == 404
+
+
+def test_disabled_worker_never_opens_database():
+ from unittest.mock import Mock
+
+ sessions = Mock(side_effect=AssertionError("No DB while disabled"))
+ worker = ExceptionWorker(
+ lambda: dict(HEALTH, enabled=False), session_factory=sessions, interval=0.01
+ )
+ worker.start()
+ worker.stop()
+ sessions.assert_not_called()
+ assert not worker.thread.is_alive()
+
+
+def test_changed_threshold_replaces_only_affected_occurrence(shipment):
+ from app.models.exception import ExceptionPolicy
+
+ db = shipment
+ run(db)
+ old = {e.kind: e.id for e in open_rows(db)}
+ for e in open_rows(db):
+ e.acknowledged_at, e.acknowledged_by = START, OWNER
+ db.add(
+ ExceptionPolicy(
+ shipment_id="SHIP-EXCEPTION",
+ revision=1,
+ arrival_enabled=True,
+ inactivity_hours=12.0,
+ checkpoint_hours=24.0,
+ custody_hours=6.0,
+ updated_at=START,
+ updated_by=OWNER,
+ )
+ )
+ run(db, 26)
+ rows = {e.kind: e for e in open_rows(db)}
+ assert rows["arrival_overdue"].id == old["arrival_overdue"]
+ assert rows["arrival_overdue"].acknowledged_at
+ assert rows["checkpoint_inactivity"].id != old["checkpoint_inactivity"]
+ assert rows["checkpoint_inactivity"].acknowledged_at is None
+ assert (
+ db.get(ShipmentException, old["checkpoint_inactivity"]).resolution
+ == "condition_replaced"
+ )
+
+
+def test_new_transfer_restarts_followup_clock(shipment):
+ db = shipment
+ add_event(db, 3, "CustodyTransferred", **{"from": OWNER, "to": HANDLER})
+ run(db, 6)
+ first = open_rows(db)[0].id
+ add_event(
+ db,
+ 4,
+ "CustodyTransferred",
+ at=START + timedelta(hours=7),
+ **{"from": HANDLER, "to": OWNER},
+ )
+ run(db, 8)
+ assert db.get(ShipmentException, first).resolved_at
+ assert not any(e.kind == "custody_followup" for e in open_rows(db))
+ run(db, 13)
+ followup = next(e for e in open_rows(db) if e.kind == "custody_followup")
+ assert followup.id != first and followup.evidence["receiving_handler"] == OWNER
+
+
+def test_orphaned_anchor_is_retained_as_withdrawn_history(shipment):
+ db = shipment
+ run(db)
+ db.get(ChainEvent, "2").canonical = False
+ run(db, 26)
+ assert not open_rows(db)
+ assert all(
+ e.resolution == "evidence_withdrawn" for e in db.query(ShipmentException).all()
+ )
diff --git a/frontend/README.md b/frontend/README.md
index 23f6bbb..9fa5491 100644
--- a/frontend/README.md
+++ b/frontend/README.md
@@ -30,3 +30,7 @@ CI=true VITE_WALLETCONNECT_PROJECT_ID=ci-test-project-id npm run test:e2e
Browser fixtures are separate from the real API/Hardhat browser acceptance in [TESTING.md](../TESTING.md). QR tests use an independent reader.
`npm run build` compiles the configured environment. Production Vercel builds use `build-vercel.mjs`, which requires successful exact-main-commit CI and a compatible ready API. Use feature PRs and stable testing before all protected-main checks; do not promote preview artifacts into production. See [RELEASE_CONTROLS.md](../RELEASE_CONTROLS.md).
+
+## Shipment exceptions
+
+See [the exception guide](../SHIPMENT_EXCEPTIONS.md) for configurable rules, synchronization gates, acknowledgement/history, owner permissions and additive schema revision `e93a107bd624`. Open `/exceptions` or a shipment detail page. Production evaluation remains paused while indexing is disabled.
diff --git a/frontend/e2e/exceptions.spec.ts b/frontend/e2e/exceptions.spec.ts
new file mode 100644
index 0000000..9fb36f0
--- /dev/null
+++ b/frontend/e2e/exceptions.spec.ts
@@ -0,0 +1,104 @@
+import { test, expect } from "@playwright/test";
+import { setupRoutes, isApiOrigin } from "./helpers/routes";
+
+test("exceptions explain disabled evaluation without declaring shipments on time", async ({
+ page,
+}) => {
+ await setupRoutes(page);
+ await page.goto("/exceptions");
+ await expect(
+ page.getByRole("heading", { name: "Shipment exceptions" }),
+ ).toBeVisible();
+ await expect(page.getByText(/Evaluation paused/)).toBeVisible();
+ await expect(page.getByText(/timeliness is not established/)).toBeVisible();
+ await expect(page).toHaveTitle("Veritras — Shipment Exceptions");
+ await page.getByLabel("Show observations").selectOption("resolved");
+ await expect(page.getByText(/No stored observations/)).toBeVisible();
+});
+
+test("mobile exception list links to evidence and exposes resolved history", async ({
+ page,
+}) => {
+ await page.setViewportSize({ width: 390, height: 844 });
+ await setupRoutes(page);
+ await page.route(
+ (url) =>
+ isApiOrigin(url) && url.pathname.replace(/\/$/, "") === "/exceptions",
+ (route) => {
+ const resolved =
+ new URL(route.request().url()).searchParams.get("state") === "resolved";
+ return route.fulfill({
+ json: {
+ total: 1,
+ evaluation: {
+ paused: false,
+ evaluated_at: "2026-10-19T01:00:00Z",
+ block_number: 100,
+ uncertain_shipments: 0,
+ },
+ items: [
+ {
+ id: resolved ? "old" : "current",
+ shipment_id: "SHIP-001",
+ kind: "custody_followup",
+ triggered_at: "2026-10-19T00:00:00Z",
+ last_observed_at: "2026-10-19T01:00:00Z",
+ acknowledged_at: null,
+ resolved_at: resolved ? "2026-10-19T02:00:00Z" : null,
+ resolution: resolved ? "condition_cleared" : null,
+ evidence: {
+ threshold_hours: 6,
+ receiving_handler: "0x5555555555555555555555555555555555555555",
+ },
+ },
+ ],
+ },
+ });
+ },
+ );
+ await page.goto("/exceptions");
+ await expect(page.getByText(/Custody already transferred/)).toBeVisible();
+ await expect(
+ page.getByRole("button", { name: "Acknowledge occurrence" }),
+ ).toHaveCount(0);
+ await expect(
+ page.getByRole("link", { name: /View shipment evidence/ }),
+ ).toHaveAttribute("href", "/shipments/SHIP-001");
+ await page.getByLabel("Show observations").selectOption("resolved");
+ await expect(page.getByText(/Resolved .*condition cleared/)).toBeVisible();
+ expect(
+ await page.evaluate(
+ () => document.documentElement.scrollWidth <= window.innerWidth,
+ ),
+ ).toBe(true);
+ await page.getByRole("link", { name: /View shipment evidence/ }).click();
+ await expect(
+ page.getByRole("heading", { name: "Shipment SHIP-001", exact: true }),
+ ).toBeVisible();
+});
+
+test("exception read failure supports retry", async ({ page }) => {
+ await setupRoutes(page);
+ let failed = true;
+ await page.route(
+ (url) =>
+ isApiOrigin(url) && url.pathname.replace(/\/$/, "") === "/exceptions",
+ (route) =>
+ failed
+ ? route.fulfill({ status: 503, json: { detail: "unavailable" } })
+ : route.fulfill({
+ json: {
+ items: [],
+ total: 0,
+ evaluation: { paused: true, reason: "indexing_disabled" },
+ },
+ }),
+ );
+ await page.goto("/exceptions");
+ await expect(
+ page.getByRole("button", { name: "Retry exceptions" }),
+ ).toBeVisible({ timeout: 20000 });
+ failed = false;
+ await page.getByRole("button", { name: "Retry exceptions" }).click();
+ await expect(page.getByText(/No stored observations/)).toBeVisible();
+});
diff --git a/frontend/e2e/helpers/routes.ts b/frontend/e2e/helpers/routes.ts
index eb1dbf2..05afa22 100644
--- a/frontend/e2e/helpers/routes.ts
+++ b/frontend/e2e/helpers/routes.ts
@@ -47,6 +47,7 @@ function apiPath(url: URL, pathname: string): boolean {
}
export async function setupRoutes(page: Page): Promise {
+ await page.route(url => isApiOrigin(url) && url.pathname.replace(/\/$/, '') === '/exceptions', route => route.fulfill({ json: { items: [], total: 0, evaluation: { paused: true, reason: 'indexing_disabled', evaluated_at: null, block_number: null, uncertain_shipments: 0 } } }))
await page.route(url => isApiOrigin(url) && /^\/shipments\/[^/]+$/.test(url.pathname), route => route.fulfill({ json: MOCK_SHIPMENTS.items[0] }))
await page.route(url => isApiOrigin(url) && /^\/products\/[^/]+\/timeline$/.test(url.pathname), route => route.fulfill({ json: { product_id: 'PROD-001', items: [], total: 0, skip: 0, limit: 100 } }))
// SSE — return an empty event stream so EventSource does not keep reconnecting
diff --git a/frontend/scripts/check-local-auth.mjs b/frontend/scripts/check-local-auth.mjs
index 2cbb72e..f7a6d40 100644
--- a/frontend/scripts/check-local-auth.mjs
+++ b/frontend/scripts/check-local-auth.mjs
@@ -165,6 +165,16 @@ try {
// Increment 8: real mobile detail actions, canonical history and public labels.
await page.setViewportSize({ width: 390, height: 844 })
await page.goto(`http://localhost:13000/shipments/${shipment.shipment_id}`, { waitUntil: 'domcontentloaded' })
+ // Increment 9: owner policy is a durable API write, separate from chain custody.
+ await page.getByLabel('Hours without a checkpoint during transit', { exact: true }).fill('30')
+ const policyResponse = page.waitForResponse(response => response.request().method() === 'PUT' && response.url().includes(`/exceptions/policies/${shipment.shipment_id}`))
+ await page.getByRole('button', { name: 'Save exception policy', exact: true }).click()
+ const policy = await (await policyResponse).json()
+ assert.equal(policy.inactivity_hours, 30)
+ assert.equal(policy.revision, 1)
+ await page.reload({ waitUntil: 'domcontentloaded' })
+ await page.getByLabel('Hours without a checkpoint during transit', { exact: true }).waitFor()
+ assert.equal(await page.getByLabel('Hours without a checkpoint during transit', { exact: true }).inputValue(), '30')
const recordButton = page.getByRole('button', { name: 'Record checkpoint', exact: true })
await recordButton.waitFor()
await recordButton.click()
@@ -227,11 +237,12 @@ try {
const edit = await fetch(`${api}/products/${productId}`, { method: 'PUT', headers: auth, body: JSON.stringify({ name: 'Unauthorized' }) })
const plan = await fetch(`${api}/shipments/`, { method: 'POST', headers: auth, body: JSON.stringify({ product_id: productId, origin: 'Other', destination: 'Other' }) })
const shipmentEdit = await fetch(`${api}/shipments/${shipmentId}`, { method: 'PUT', headers: auth, body: JSON.stringify({ notes: 'Unauthorized' }) })
+ const exceptionPolicy = await fetch(`${api}/exceptions/policies/${shipmentId}`, { method: 'PUT', headers: auth, body: JSON.stringify({ expected_revision: 1, inactivity_hours: 31 }) })
const anonymous = await fetch(`${api}/products/`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ name: 'Anonymous' }) })
const publicRead = await fetch(`${api}/products/${productId}`)
- return [edit.status, plan.status, shipmentEdit.status, anonymous.status, publicRead.status]
+ return [edit.status, plan.status, shipmentEdit.status, exceptionPolicy.status, anonymous.status, publicRead.status]
}, { api, productId: draft.product_id, shipmentId: shipment.shipment_id })
- assert.deepEqual(checks, [403, 403, 403, 401, 200])
+ assert.deepEqual(checks, [403, 403, 403, 403, 401, 200])
// Real browser analytics and downloads must agree with the same API cohort.
await page.goto('http://localhost:13000/analytics', { waitUntil: 'domcontentloaded' })
await page.getByRole('region', { name: 'Shipment sample counts' }).waitFor()
diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx
index 3f1536d..9d929b2 100644
--- a/frontend/src/App.tsx
+++ b/frontend/src/App.tsx
@@ -15,6 +15,7 @@ import Layout from './components/Layout'
import Dashboard from './pages/Dashboard'
const ShipmentDetail = lazy(() => import('./pages/ShipmentDetail'))
+const Exceptions = lazy(() => import('./pages/Exceptions'))
const Demo = lazy(() => import('./pages/Demo'))
const NotFound = lazy(() => import('./pages/NotFound'))
const Products = lazy(() => import('./pages/Products'))
@@ -63,6 +64,7 @@ function App() {
} />
} />
} />
+ } />
} />
} />
} />
diff --git a/frontend/src/components/ExceptionList.test.tsx b/frontend/src/components/ExceptionList.test.tsx
new file mode 100644
index 0000000..92fa964
--- /dev/null
+++ b/frontend/src/components/ExceptionList.test.tsx
@@ -0,0 +1,124 @@
+import { fireEvent, render, screen } from "@testing-library/react";
+import { MemoryRouter } from "react-router-dom";
+import { beforeEach, expect, it, vi } from "vitest";
+import ExceptionList from "./ExceptionList";
+import { useExceptions, useAcknowledgeException } from "@/hooks/useExceptions";
+import { useAuth } from "@/hooks/useAuth";
+vi.mock("@/hooks/useExceptions", () => ({
+ useExceptions: vi.fn(),
+ useAcknowledgeException: vi.fn(),
+}));
+vi.mock("@/hooks/useAuth", () => ({ useAuth: vi.fn() }));
+const owner = "0x1234";
+const mutate = vi.fn();
+const item = {
+ id: "episode",
+ shipment_id: "SHIP-1",
+ kind: "custody_followup",
+ owner_address: owner,
+ ownership_source: "chain_event",
+ evidence: { receiving_handler: "0x5678", threshold_hours: 6 },
+ triggered_at: "2026-01-01T00:00:00Z",
+ first_observed_at: "2026-01-01T00:00:00Z",
+ last_observed_at: "2026-01-01T01:00:00Z",
+ resolved_at: null,
+ acknowledged_at: null,
+};
+function query(items: unknown[] = [item], paused = false) {
+ vi.mocked(useExceptions).mockReturnValue({
+ data: {
+ items,
+ total: items.length,
+ evaluation: {
+ paused,
+ reason: paused ? "indexing_disabled" : null,
+ evaluated_at: null,
+ },
+ },
+ } as ReturnType);
+}
+function mount() {
+ return render(
+
+
+ ,
+ );
+}
+beforeEach(() => {
+ vi.mocked(useAuth).mockReturnValue({
+ isAuthenticated: true,
+ address: owner,
+ } as ReturnType);
+ vi.mocked(useAcknowledgeException).mockReturnValue({
+ mutate,
+ isPending: false,
+ } as unknown as ReturnType);
+ mutate.mockClear();
+ query();
+});
+it("explains custody is already transferred and links to shipment evidence", () => {
+ mount();
+ expect(
+ screen.getByText(/Custody already transferred on-chain/),
+ ).toBeInTheDocument();
+ expect(
+ screen.getByRole("link", { name: /View shipment evidence/ }),
+ ).toHaveAttribute("href", "/shipments/SHIP-1");
+ fireEvent.change(screen.getByRole("textbox"), {
+ target: { value: "Contacted carrier" },
+ });
+ fireEvent.click(
+ screen.getByRole("button", { name: "Acknowledge occurrence" }),
+ );
+ expect(mutate).toHaveBeenCalledWith({
+ id: "episode",
+ note: "Contacted carrier",
+ });
+});
+it.each([
+ { isAuthenticated: false, address: null },
+ { isAuthenticated: true, address: "other" },
+])("hides acknowledgement for nonowners: %o", (auth) => {
+ vi.mocked(useAuth).mockReturnValue(auth as ReturnType);
+ mount();
+ expect(
+ screen.queryByRole("button", { name: "Acknowledge occurrence" }),
+ ).not.toBeInTheDocument();
+});
+it("shows paused empty state without asserting shipment timeliness", () => {
+ query([], true);
+ mount();
+ expect(
+ screen.getByText(/missing updates do not establish/),
+ ).toBeInTheDocument();
+ expect(screen.getByText(/timeliness is not established/)).toBeInTheDocument();
+});
+it.each([
+ { acknowledged_at: "2026-01-01T02:00:00Z" },
+ { resolved_at: "2026-01-01T03:00:00Z", resolution: "condition_cleared" },
+])(
+ "retains history and prevents acknowledgement of changed episodes",
+ (change) => {
+ query([{ ...item, ...change }]);
+ mount();
+ expect(
+ screen.queryByRole("button", { name: "Acknowledge occurrence" }),
+ ).not.toBeInTheDocument();
+ },
+);
+it("renders notes as text", () => {
+ query([{ ...item, acknowledgement_note: "" }]);
+ const { container } = mount();
+ expect(container.querySelector("script")).toBeNull();
+ expect(screen.getByText(/