test: reconcile P4.2 PostgreSQL integration split

.github/workflows/ci.yml

@@ -118,7 +118,7 @@ jobs:
           SQL
 
       - name: Run PostgreSQL integration suite
-        run: uv run pytest -q -m pg tests/test_pg_integration.py
+        run: uv run pytest -q -m pg tests/pg/
 
       - name: Upload cleanup evidence
         if: always()

docs/guides/postgres-integration-tests.md

@@ -1,8 +1,10 @@
 # Disposable PostgreSQL integration tests
 
-`tests/test_pg_integration.py` is destructive by design: it creates and deletes
-repository-scoped sprint data. It must never run against the shared sprintctl
-authority or any other persistent database.
+`tests/pg/` (split by domain from the former `tests/test_pg_integration.py` in
+P4.2 -- `tests/pg/_shared.py` holds the shared fixtures, skip machinery, and
+helpers every file in the package imports) is destructive by design: it
+creates and deletes repository-scoped sprint data. It must never run against
+the shared sprintctl authority or any other persistent database.
 
 ## Safety contract
 
@@ -45,7 +47,7 @@ Then run only against that disposable instance:
 ```bash
 export SPRINTCTL_TEST_PG_URL='postgresql://sprintctl_test_local:<temporary-password>@127.0.0.1/sprintctl_test_local'
 export SPRINTCTL_TEST_PG_CLEANUP_REPORT="$PWD/pg-cleanup-report.json"
-uv run --extra remote pytest -m pg tests/test_pg_integration.py -v
+uv run --extra remote pytest -m pg tests/pg/ -v
 ```
 
 The password is temporary test infrastructure state. Never commit it or reuse

docs/ops-upgrade-plan.md

@@ -49,7 +49,7 @@ Do not trust the existing plan documents without checking them against current c
 Verify and reconcile at least these known findings:
 
 - The capability-receipts implementation has landed in sprintctl `main` in commit `a5aef844a8ea9c9776b31ebd7028c0d161d3d0fa`. Treat it as shipped, not pending.
-- `sprintctl/tests/test_pg_integration.py` accepts an arbitrary `SPRINTCTL_TEST_PG_URL`, creates an `itest-<uuid>` repo scope, and depends on normal fixture teardown for cleanup. This allowed test data to remain in production PostgreSQL.
+- `sprintctl/tests/pg/` accepts an arbitrary `SPRINTCTL_TEST_PG_URL`, creates an `itest-<uuid>` repo scope, and depends on normal fixture teardown for cleanup. This allowed test data to remain in production PostgreSQL.
 - The workspace backup gap described by `agentops/docs/plans/agentops/substrate-resilience-plan.md` is stale. `appservice/clusters/main/kubernetes/apps/vscode/app/workspace-backup.yaml` contains a daily Restic backup and monthly restore drill. Distinguish:
   - implemented in GitOps;
   - deployed/runtime state;

docs/plans/1219-recovery-export-plan.md

@@ -78,7 +78,7 @@ Decisions folded in (do not re-litigate during build):
   the repo).
 - Tests: unit for envelope/insert mapping; pg integration test exercising
   export-then-verify against a seeded disposable Postgres
-  (`tests/test_pg_integration.py` pattern).
+  (`tests/pg/` pattern).
 - Docs: update `docs/guides/remote-mode.md` "Reverting to local mode" to make this
   command the primary recovery path; note pg_dump remains the pg→pg option.
 
@@ -88,7 +88,7 @@ Out of scope: deleting split-mode code (#1164), enabling remote for `export`/
 Verification (per sprintctl dispatch manifest — targeted first):
 
 ```
-uv run pytest tests/test_db_recover.py tests/test_pg_integration.py -x --tb=short
+uv run pytest tests/test_db_recover.py tests/pg/ -x --tb=short
 python /projects/dev/agentops/templates/dispatch/scripts/validate_verification_artifacts.py --root .
 ```
 

docs/protocols/projection-recovery.md

@@ -32,7 +32,7 @@ than a fabricated current projection.
 |---|---|---|
 | Records and watermark commit atomically, then retry converges | `sprintctl/projection.py:apply_ingested_records` and `sprintctl/sync.py:synchronize_outbox` | `tests/test_projection.py:test_failed_record_application_rolls_back_records_and_watermark` and `tests/test_sync.py:test_sync_recovers_after_lost_response_or_projection_apply_failure` |
 | Offline catch-up consumes the contiguous suffix | `sprintctl/projection.py:apply_ingested_records` | `tests/test_projection.py:test_offline_catch_up_and_full_log_rebuild_converge` |
-| Complete-log rebuild converges | `sprintctl/pg.py:list_ingested_records` and projection apply | SQLite history above plus `tests/test_pg_integration.py:test_disposable_remote_history_rebuilds_projection` |
+| Complete-log rebuild converges | `sprintctl/pg.py:list_ingested_records` and projection apply | SQLite history above plus `tests/pg/test_outbox.py::TestProducerOutboxIngestion::test_disposable_remote_history_rebuilds_projection` |
 | Retention cannot jump a cursor | `sprintctl/projection.py:ProjectionGapError` | `tests/test_projection.py:test_retained_suffix_cannot_advance_an_empty_projection` |
 
 The PostgreSQL expansion is guarded by disposable credentials and repository

sprintctl.dispatch.json

@@ -42,7 +42,7 @@
   "risk_surfaces": [
     {
       "id": "claim-ownership",
-      "paths": ["sprintctl/db.py", "sprintctl/pg.py", "tests/test_claims.py", "tests/test_pg_integration.py"],
+      "paths": ["sprintctl/db.py", "sprintctl/pg.py", "tests/test_claims.py", "tests/pg/"],
       "skills": ["verify-state-protocols", "reconcile-project-contracts"],
       "default_depth": 2,
       "required_on_change": true
@@ -62,7 +62,7 @@
         "sprintctl/pg.py",
         "sprintctl/cli.py",
         "tests/test_capability_receipts.py",
-        "tests/test_pg_integration.py",
+        "tests/pg/",
         "docs/reference/capability-receipts.md"
       ],
       "skills": ["verify-state-protocols", "reconcile-project-contracts"],

sprintctl/workitemcore.py

@@ -21,7 +21,7 @@
 writers on *different* work items already serialize against each other
 (same as before this extraction), while PostgreSQL's row lock lets them
 proceed in parallel. What both backends must guarantee identically — and
-what ``tests/test_core.py``'s and ``tests/test_pg_integration.py``'s
+what ``tests/test_core.py``'s and ``tests/pg/test_work_item.py``'s
 concurrency tests assert — is that two writers racing the *same* item
 produce exactly one accepted write and one ``EditConflict``.
 

tests/pg/_shared.py

@@ -0,0 +1,248 @@
+"""
+Shared fixtures, helpers, and marker for the tests/pg/ PostgreSQL integration test package.
+
+Requires a disposable PostgreSQL database owned by a dedicated, unprivileged
+test role. See docs/guides/postgres-integration-tests.md for the contract.
+Set SPRINTCTL_TEST_PG_URL to run:
+
+    SPRINTCTL_TEST_PG_URL=postgresql://localhost/testdb pytest tests/pg/ -v
+
+All tests are automatically skipped when the variable is unset or psycopg is unavailable.
+"""
+from __future__ import annotations
+
+from dataclasses import replace
+from datetime import datetime, timedelta, timezone
+import hashlib
+import io
+import json
+import os
+import threading
+import time
+import uuid
+
+import pytest
+
+_PG_URL: str | None = os.environ.get("SPRINTCTL_TEST_PG_URL")
+
+try:
+    import psycopg
+    from psycopg.rows import dict_row
+    _PSYCOPG_AVAILABLE = True
+except ImportError:
+    # Split modules import these names during collection even without the
+    # optional remote extra. PG_MARKS skips them before runtime use.
+    psycopg = None
+    dict_row = None
+    _PSYCOPG_AVAILABLE = False
+
+_SKIP = not _PG_URL or not _PSYCOPG_AVAILABLE
+_SKIP_REASON = (
+    "SPRINTCTL_TEST_PG_URL not set"
+    if not _PG_URL
+    else "psycopg not installed — run: pip install 'sprintctl[remote]'"
+)
+
+PG_MARKS = [
+    pytest.mark.pg,
+    pytest.mark.skipif(_SKIP, reason=_SKIP_REASON),
+]
+
+# Safe unconditional imports: pg.py handles missing psycopg gracefully.
+from sprintctl import authority, contracts, db, maintain, observations, pg, projection, sync
+from sprintctl import outbox
+from sprintctl import pg_migrations
+from sprintctl.cli import cli
+from sprintctl.db import ClaimConflict, InvalidTransition
+from sprintctl.pg_testing import (
+    assert_disposable_connection,
+    cleanup_test_repositories,
+    new_test_repo_id,
+    new_test_repo_uuid,
+    write_cleanup_report,
+)
+from sprintctl.maintenance_capability import (
+    MaintenanceCapabilityError,
+    PostgresMaintenanceCapabilityStore,
+)
+from tests.test_maintenance_capability import (
+    CAPABILITY_ID, AT, envelope, shifted_envelope, _stamp,
+)
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+@pytest.fixture(scope="module")
+def pg_test_scope():
+    """Register test scopes, then clean and report them from one finalizer."""
+    if _SKIP:
+        pytest.skip(_SKIP_REASON)
+    conn = psycopg.connect(_PG_URL, row_factory=dict_row)
+    assert_disposable_connection(conn)
+    repo_ids: set[str] = set()
+
+    def register(label: str = "scope", *, canonical_uuid: bool = False) -> str:
+        repo_id = new_test_repo_uuid() if canonical_uuid else new_test_repo_id(label)
+        repo_ids.add(repo_id)
+        return repo_id
+
+    try:
+        yield register
+    finally:
+        report_path = os.environ.get("SPRINTCTL_TEST_PG_CLEANUP_REPORT")
+        try:
+            report = cleanup_test_repositories(conn, repo_ids)
+        except Exception as exc:
+            if report_path:
+                write_cleanup_report(
+                    report_path,
+                    {
+                        "schema_version": "sprintctl-pg-cleanup/v1",
+                        "cleanup_completed": False,
+                        "error_type": type(exc).__name__,
+                        "repo_ids": sorted(repo_ids),
+                    },
+                )
+            raise
+        else:
+            if report_path:
+                write_cleanup_report(report_path, report)
+        finally:
+            conn.close()
+
+
+@pytest.fixture(scope="module")
+def store(pg_test_scope):
+    """Create the module store only after the disposable-target preflight."""
+    conn = psycopg.connect(_PG_URL, row_factory=dict_row)
+    assert_disposable_connection(conn)
+    repo_id = pg_test_scope("module")
+    s = pg.PgStore(
+        conn=conn,
+        repo_id=repo_id,
+        authority_repo_uuid=str(uuid.uuid5(uuid.NAMESPACE_URL, f"sprintctl-repo:{repo_id}")),
+    )
+    pg.init_db(s)
+    try:
+        yield s
+    finally:
+        conn.close()
+
+
+def _uid() -> str:
+    return uuid.uuid4().hex[:8]
+
+
+def _anchor_capability_window_to_db_clock(conn, repo_id, capability_id) -> None:
+    """Align a prepared capability's window with the database clock.
+
+    Claim admission filters live capabilities on the database clock
+    (``expires_at > now()``), while capability transitions evaluate the window
+    against the caller-supplied ``at``. In production those track each other;
+    the fixture envelope instead pins ``at`` to a fixed AT and its window to a
+    fixed instant, so once wall-clock time passes that instant the two
+    arbitration paths disagree and repo-wide mutual exclusion silently stops
+    being exercised. Anchoring the stored window to the database's own clock
+    keeps these races meaningful at any future date.
+    """
+    with conn.cursor() as cur:
+        cur.execute(
+            "UPDATE maintenance_capability SET expires_at = now() + interval '1 hour' "
+            "WHERE repo_id = %s AND capability_id = %s",
+            (repo_id, capability_id),
+        )
+    conn.commit()
+
+
+def _authority_repo_uuid(store) -> str:
+    return str(uuid.uuid5(uuid.NAMESPACE_URL, f"sprintctl-repo:{store.repo_id}"))
+
+
+def _append_authority_command(
+    conn,
+    store,
+    *,
+    record_type,
+    aggregate_type,
+    basis_revision,
+    payload,
+    aggregate_uuid=None,
+    claim_id=None,
+    actor="authority-test",
+):
+    refs = {
+        "repo_id": _authority_repo_uuid(store),
+        "aggregate_type": aggregate_type,
+    }
+    if aggregate_uuid is not None:
+        refs["aggregate_uuid"] = aggregate_uuid
+    if claim_id is not None:
+        refs["claim_id"] = claim_id
+    command = contracts.AuthorityCommand(
+        event_id=str(uuid.uuid4()),
+        record_type=record_type,
+        schema_version="1",
+        actor=actor,
+        authored_at="2026-07-14T18:00:00Z",
+        refs=refs,
+        payload=payload,
+        basis_revision=basis_revision,
+        correlation_id=str(uuid.uuid4()),
+    )
+    return outbox.append_authority_command(conn, command)
+
+
+def _receipt_bytes(store, sprint_id, boundary_event_id, **overrides):
+    receipt_id = f"{store.repo_id}.2026-07-13.boundary"
+    receipt = {
+        "schema_version": "capability-receipt/v1",
+        "id": receipt_id,
+        "project": store.repo_id,
+        "status": "draft",
+        "publication": "private",
+        "boundary": {
+            "kind": "sprint-close",
+            "ref": {
+                "kind": "sprint-event",
+                "source": f"sprintctl:{store.repo_id}:sprint:{sprint_id}",
+                "revision": f"event:{boundary_event_id}",
+            },
+        },
+    }
+    receipt.update(overrides)
+    return json.dumps(receipt, sort_keys=True).encode()
+
+
+def _receipt_payload(store, receipt_bytes):
+    receipt_id = f"{store.repo_id}.2026-07-13.boundary"
+    return {
+        "project": store.repo_id,
+        "receipt_id": receipt_id,
+        "receipt_path": (
+            f"/projects/dev/_artifacts/{store.repo_id}/capability/receipts/"
+            f"{receipt_id}.json"
+        ),
+        "receipt_sha256": hashlib.sha256(receipt_bytes).hexdigest(),
+    }
+
+
+@pytest.fixture
+def sprint_id(store):
+    return pg.create_sprint(store, f"S-{_uid()}", "Goal", "2026-01-01", "2026-12-31", "active")
+
+
+@pytest.fixture
+def track_id(store, sprint_id):
+    return pg.get_or_create_track(store, sprint_id, "eng")
+
+
+@pytest.fixture
+def work_item_id(store, sprint_id, track_id):
+    return pg.create_work_item(store, sprint_id, track_id, f"Item-{_uid()}")
+
+
+# ---------------------------------------------------------------------------
+# Schema
+# ---------------------------------------------------------------------------

tests/pg/conftest.py

@@ -0,0 +1,13 @@
+"""Registers tests/pg/_shared.py's module-scoped fixtures for this directory.
+
+Fixture functions must be visible as names in a conftest.py (or the test
+module itself) for pytest to discover them; importing them here re-exports
+tests/pg/_shared.py's fixtures to every test file in this package.
+"""
+from tests.pg._shared import (  # noqa: F401
+    pg_test_scope,
+    store,
+    sprint_id,
+    track_id,
+    work_item_id,
+)