diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e277284c2..534a5a5aa 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -31,6 +31,9 @@ jobs: uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 with: persist-credentials: false + ref: ${{ github.event.pull_request.head.sha || github.sha }} + - name: Verify exact source head + run: test "$(git rev-parse HEAD)" = "${{ github.event.pull_request.head.sha || github.sha }}" - name: Set up Python uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 with: @@ -58,6 +61,9 @@ jobs: uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 with: persist-credentials: false + ref: ${{ github.event.pull_request.head.sha || github.sha }} + - name: Verify exact source head + run: test "$(git rev-parse HEAD)" = "${{ github.event.pull_request.head.sha || github.sha }}" - name: Set up Python uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 with: @@ -111,6 +117,9 @@ jobs: uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 with: persist-credentials: false + ref: ${{ github.event.pull_request.head.sha || github.sha }} + - name: Verify exact source head + run: test "$(git rev-parse HEAD)" = "${{ github.event.pull_request.head.sha || github.sha }}" - name: Set up Python uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 with: @@ -122,8 +131,8 @@ jobs: prune-cache: true - name: Install locked dependencies run: uv sync --locked - - name: Run live checkpoint audit integration - run: uv run pytest -q tests/test_checkpoint_audit_integration.py -m integration + - name: Run live checkpoint storage integration + run: uv run pytest -q tests/test_checkpoint_audit_integration.py tests/test_checkpoint_migration_operator_integration.py -m integration container-builds: name: Container builds @@ -138,9 +147,12 @@ jobs: uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 with: persist-credentials: false + ref: ${{ github.event.pull_request.head.sha || github.sha }} + - name: Verify exact source head + run: test "$(git rev-parse HEAD)" = "${{ github.event.pull_request.head.sha || github.sha }}" - name: Validate Compose configuration run: docker compose config >/dev/null - name: Build component image run: docker build --tag pg-llm-batch:ci . - name: Build PostgreSQL image - run: docker build --tag pg-llm-batch-postgres:ci docker/postgres \ No newline at end of file + run: docker build --tag pg-llm-batch-postgres:ci docker/postgres diff --git a/AGENTS.md b/AGENTS.md index f1fd6a1a5..4cd97f2c9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -237,3 +237,36 @@ add CODEOWNERS-based merge gates until multiple independent maintainers exist. deterministic transaction, event-time, tenant, validation, bounded-read, migration, rollback, immutability-trigger, compatibility, and documentation tests. + +## Checkpoint migration operator contract + +- Keep the **Checkpoint migration operator** opt-in through + `init-checkpoint-storage`; `init-db` must remain the backward-compatible core + schema command. +- Load, bounded-read, strict UTF-8 decode, and SHA-256 identify both + `0007_result_stream_checkpoints` and + `0008_result_checkpoint_audit_events` before database access. Each file is + limited to 1 MiB plus one detection byte and the SQL body remains private. +- Obtain the fixed two-key transaction-level advisory lock with + `pg_advisory_xact_lock`, then execute migration 0007 before migration 0008 in + one PostgreSQL transaction and issue one commit only after both succeed. +- A second-migration failure must roll back the first migration from the same + invocation and release the transaction-level advisory lock automatically. +- Emit only immutable migration identifiers, bounded byte counts, SHA-256, and + schema version after commit. SHA-256 is change-identification evidence and + **not a signature**, provenance statement, remote attestation, or release + authority. +- Preserve the independent `apply_result_checkpoint_schema()` and + `apply_result_checkpoint_audit_schema()` helpers for hosts that intentionally + own separate transactions. Do not add a migration ledger, downgrade, or + destructive retained-evidence rollback to this bounded operator. +- Keep standalone operation and modular MSA use independent of `naruon` and + `contextual-orchestrator`. Advisory locking coordinates cooperating package + operators; it is not authorization and does not constrain an administrator or + unrelated SQL client. +- Maintain 100% production statement, branch, and public-docstring coverage with + deterministic input-bound, order, transaction, rollback, live PostgreSQL, + concurrency, compatibility, CLI, documentation, and body-free evidence tests. +- Update README, architecture, ADR, operator documentation, doctoring, and + CHANGELOG whenever migration ordering, locking, evidence, failure, or + compatibility semantics change. diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index d9afe42b5..8b007a768 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -303,6 +303,49 @@ the surrounding transaction later committed. The host owns telemetry retention, access control, alerting, collector availability, database-client instrumentation, and any correlation outside this confidential package boundary. +## Checkpoint schema migration operator boundary + +`init-checkpoint-storage` is an explicit opt-in operator command for existing +PostgreSQL volumes. It preserves `init-db` as the core-schema command and keeps +the independent `apply_result_checkpoint_schema()` and +`apply_result_checkpoint_audit_schema()` helpers compatible for hosts that +intentionally own separate transactions. + +Before database access, the operator bounded-reads, strict UTF-8 decodes, counts, +and SHA-256 identifies `0007_result_stream_checkpoints` and +`0008_result_checkpoint_audit_events` in that exact order. Each file is limited +to 1 MiB plus one oversize-detection byte. The loaded SQL remains private; public +migration descriptors contain only configured identifiers, positive bounded byte +counts, and lowercase SHA-256. + +After both inputs are valid, one package-owned PostgreSQL transaction obtains the +fixed two-key `pg_advisory_xact_lock`, executes migration 0007, executes migration +0008, and issues one commit. A migration 0008 failure rolls back migration 0007 +from the same invocation and transaction end releases the lock automatically. +The advisory lock serializes cooperating package operators; it is not an +authorization mechanism and does not constrain an administrator, owner, +superuser, or unrelated SQL client. + +The CLI emits one canonical JSON report only after commit. It excludes DSNs, +credentials, SQL bodies, tenants, checkpoint values, provider payloads, audit +rows, and raw database exception text. SHA-256 is deterministic +change-identification evidence and is not a signature, provenance claim, remote +attestation, publication authority, or integrated-release approval. + +No migration ledger, downgrade path, destructive retained-evidence rollback, +provider credential, LLM key, `naruon`, or `contextual-orchestrator` dependency is +introduced. Fresh Docker data directories retain their ordered initialization +scripts; existing PostgreSQL volumes use the explicit operator command. + +```text +init-checkpoint-storage + ├─ bounded load: 0007_result_stream_checkpoints + ├─ bounded load: 0008_result_checkpoint_audit_events + ├─ pg_advisory_xact_lock(PGLM, BATH) + ├─ execute 0007 → 0008 + └─ one commit → bounded migration identity JSON +``` + ## Modular interoperability CWL hosts such as `contextual-orchestrator` and `naruon` supply tenant context @@ -327,7 +370,10 @@ place local PostgreSQL effects and `save_in_transaction()` on the same caller cursor; cross-system effects remain host-owned recovery boundaries. Hosts that choose the audited store gain transaction-coupled accepted-save evidence but still own identity authorization, retention, export, and stronger tamper-proof -controls where required. +controls where required. Hosts may use the checkpoint migration operator as a +standalone deployment primitive and retain its bounded descriptors in a +change-management record, but must not reinterpret them as tenant authorization +or release provenance. ## Verification boundary @@ -373,5 +419,10 @@ attributes, storage-agnostic operation spans, seconds-based nonnegative duration confidential failure classification, explicit Error status without descriptions, Unset success status, disabled exception recording, and preservation of application results and exception identity during ordinary tracer, meter, span, -status, and clock failures. Final merge evidence must be regenerated against the -integrated base; successful stacked-base runs are not reusable release evidence. +status, and clock failures. Checkpoint migration operator tests prove strict +bounded input, canonical identities, load-before-connect, one transaction-level +advisory lock, exact 0007→0008 order, one commit, second-migration rollback, +concurrent lock waiting, body-free JSON, unchanged `init-db`, and 100% production +statement, branch, and public-docstring coverage. Final merge evidence must be +regenerated against the integrated base; successful stacked-base runs are not +reusable release evidence. diff --git a/CHANGELOG.md b/CHANGELOG.md index 9a03e0c64..5825217f2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Explicit opt-in `init-checkpoint-storage` operator for existing PostgreSQL + volumes. It bounded-reads, validates, and SHA-256 identifies + `0007_result_stream_checkpoints` and + `0008_result_checkpoint_audit_events` before database access, obtains one + fixed two-key `pg_advisory_xact_lock`, executes both migrations in canonical + order inside one transaction, and issues one commit only after both succeed. + Failure in migration 0008 rolls back migration 0007 from the same invocation. + Success output contains only stable migration identifiers, byte counts, + SHA-256, and schema version; the digest is change-identification evidence, not + a signature, attestation, provenance statement, or release authority. + `init-db` and the two independent migration helpers remain compatible. Live + PostgreSQL tests prove rollback and advisory-lock serialization. No migration + ledger, downgrade, destructive retained-evidence rollback, version bump, or + release is included. - Optional append-only checkpoint accepted-save audit evidence through `AuditedPostgresBatchResultCheckpointStore` and `llm_result_checkpoint_audit_events`. Successful save calls append a fixed diff --git a/CLAUDE.md b/CLAUDE.md index 23e3959b8..ee08b9d72 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -214,3 +214,35 @@ - Maintain test-first transaction, event-time, validation, bounded-read, migration, rollback, trigger, compatibility, documentation, and 100% production statement/branch/public-docstring coverage. + +## Checkpoint migration operator invariants + +- Keep the **Checkpoint migration operator** explicit and opt-in through + `init-checkpoint-storage`; never expand `init-db` to install optional checkpoint + objects silently. +- Bounded-read, strict UTF-8 decode, and SHA-256 identify + `0007_result_stream_checkpoints` and + `0008_result_checkpoint_audit_events` before database access. The maximum + package read is 1 MiB plus one oversize-detection byte per file. +- Acquire the fixed two-key transaction-level advisory lock with + `pg_advisory_xact_lock`, execute 0007 before 0008 in one transaction, and issue + one commit only after both succeed. +- A second-migration failure must roll back the first migration from the same + invocation. Transaction exit must release the transaction-level advisory lock; + do not add a persistent lock table or migration ledger for this slice. +- Public and CLI evidence contains only configured migration IDs, bounded byte + counts, SHA-256, and schema version after commit. SHA-256 is + change-identification evidence and **not a signature**, provenance, remote + attestation, or release approval. +- Preserve source compatibility for `apply_result_checkpoint_schema()` and + `apply_result_checkpoint_audit_schema()`. They remain available to hosts that + intentionally coordinate independent transactions. +- Treat advisory locking as cooperative serialization, not authorization. An + administrator, superuser, owner, or unrelated SQL client remains outside the + lock assurance boundary. +- Preserve standalone and modular MSA use without requiring `naruon`, + `contextual-orchestrator`, provider credentials, or an LLM key. +- Maintain 100% production statement, branch, and public-docstring coverage with + strict red-green-refactor tests for bounded input, exact order, one lock, one + transaction, one commit, rollback, concurrency, compatibility, CLI output, + documentation, and body-free diagnostics. diff --git a/README.md b/README.md index e68518559..0370d7996 100644 --- a/README.md +++ b/README.md @@ -19,6 +19,9 @@ core and relicensed under Apache-2.0. See [`NOTICE`](NOTICE) for provenance. bootstrap transport. - **Durable provider reconciliation.** Opt-in lifecycle clients persist ordered remote state across restarts and worker failover. +- **Atomic checkpoint upgrades.** Existing PostgreSQL volumes can install the + durable checkpoint and accepted-save audit schemas with one explicit, + serialized, all-or-nothing operator command. - **Standalone or tenant-scoped operation.** Existing single-tenant consumers retain a compatible facade, while shared-table MSA deployments can use a trusted tenant-qualified identity and forced PostgreSQL row-level security. @@ -64,6 +67,12 @@ Optional durable projection: TenantDurableBatchAPIClient └─ (tenant_scope, endpoint_alias, remote_batch_id) + +Optional checkpoint schema operator: + init-checkpoint-storage + ├─ 0007_result_stream_checkpoints + ├─ 0008_result_checkpoint_audit_events + └─ pg_advisory_xact_lock → one transaction → one commit ``` | Capability | Module | @@ -74,6 +83,8 @@ Optional durable projection: | Incremental bounded result and error records | `pg_llm_batch/result_streaming.py` | | Durable standalone and tenant lifecycle clients | `pg_llm_batch/durable_client.py` | | Tenant-qualified lifecycle persistence and reads | `pg_llm_batch/db.py` | +| Durable checkpoint persistence and audit | `pg_llm_batch/checkpoint_store.py`, `pg_llm_batch/checkpoint_audit.py` | +| Atomic checkpoint schema planning and application | `pg_llm_batch/checkpoint_migrations.py` | | Opt-in OpenTelemetry operations | `pg_llm_batch/observability.py` | | KV configuration and encrypted secrets | `pg_llm_batch/config.py` | | Canonical PostgreSQL DDL | `pg_llm_batch/schema.sql` | @@ -107,6 +118,23 @@ python -m pg_llm_batch config set-secret \ gateway_api_key.default sk-your-key ``` +`init-db` applies only the backward-compatible core batch schema. To upgrade an +existing PostgreSQL volume for durable result checkpoints and accepted-save audit +evidence, run the explicit opt-in command: + +```bash +python -m pg_llm_batch init-checkpoint-storage +``` + +The command bounded-reads and SHA-256 identifies +`0007_result_stream_checkpoints` and +`0008_result_checkpoint_audit_events` before database access, obtains one +transaction-level `pg_advisory_xact_lock`, applies both in one transaction, and +issues one commit. A failure in the second migration rolls back the first. The +success JSON contains migration identifiers, byte counts, and SHA-256 only; +SHA-256 is change-identification evidence and not a signature or attestation. +See the [checkpoint storage migration operator guide](docs/checkpoint-storage-migrations.md). + Production gateway destinations require HTTPS. Plain HTTP is accepted only for explicit loopback development endpoints. URLs containing user information, queries, fragments, whitespace, controls, backslashes, or invalid ports fail @@ -381,10 +409,18 @@ PG_LLM_BATCH_TEST_DSN=postgresql://pgllm:pgllm@localhost:5432/pgllm \ Protected CI verifies Python 3.10, 3.12, and 3.14; compilation; Ruff; 100% production statement and branch coverage; 100% production docstrings; lockfile freshness; source and wheel packaging; Compose validation; component and -PostgreSQL container builds; SAST; and security scanning. +PostgreSQL container builds; live checkpoint migration rollback and advisory +serialization; SAST; and security scanning. ## Documentation +- [`docs/checkpoint-storage-migrations.md`](docs/checkpoint-storage-migrations.md) + — atomic existing-volume checkpoint/audit upgrade, rollback, evidence, and + concurrency contract. +- [`docs/adr/0010-atomic-checkpoint-schema-operator.md`](docs/adr/0010-atomic-checkpoint-schema-operator.md) + — decision record for ordered transaction-level advisory locking. +- [`docs/doctoring/checkpoint-migration-operator.md`](docs/doctoring/checkpoint-migration-operator.md) + — standards mapping, exact verification boundary, and APA 7 references. - [`docs/remote-batch-lifecycle.md`](docs/remote-batch-lifecycle.md) — standalone and tenant lifecycle operations, RLS trust boundary, migration, rollback, and recovery. diff --git a/docs/adr/0010-atomic-checkpoint-schema-operator.md b/docs/adr/0010-atomic-checkpoint-schema-operator.md new file mode 100644 index 000000000..42e93589f --- /dev/null +++ b/docs/adr/0010-atomic-checkpoint-schema-operator.md @@ -0,0 +1,115 @@ +# ADR 0010: Atomic checkpoint schema operator + +- **Status:** Accepted +- **Date:** 2026-08-07 +- **Decision owners:** ContextualWisdomLab maintainers + +## Context + +The durable checkpoint migration `0007_result_stream_checkpoints` and the +accepted-save audit migration `0008_result_checkpoint_audit_events` are ordered +and individually idempotent. Existing public helpers apply each migration through +a separate connection and commit. That separation is intentionally compatible +with hosts that own upgrade orchestration, but it leaves a buyer-visible operator +gap: a routine existing-volume upgrade can partially install checkpoint schema +when the audit migration is missing, unreadable, malformed, or fails after the +first helper commits. + +Docker entrypoint initialization is suitable only for a fresh data directory and +must not be represented as an upgrade mechanism for existing PostgreSQL volumes. +The core `init-db` command also cannot silently gain optional checkpoint objects +without breaking the established standalone compatibility boundary. + +## Decision + +Add the explicit opt-in command `init-checkpoint-storage` and public coordinator +`apply_checkpoint_schema_migrations()`. + +The coordinator performs these steps in order: + +1. bounded-read, validate as strict UTF-8, count, and SHA-256 identify + `0007_result_stream_checkpoints`; +2. bounded-read, validate as strict UTF-8, count, and SHA-256 identify + `0008_result_checkpoint_audit_events`; +3. connect only after both canonical files pass the 1 MiB bound; +4. obtain the fixed two-key transaction-level advisory lock through + `pg_advisory_xact_lock`; +5. execute migration 0007 and then migration 0008 in one transaction; +6. issue one commit; and +7. return only immutable migration identifiers, byte counts, and SHA-256 values. + +The JSON emitted by the CLI appears only after successful commit. It excludes +DSNs, credentials, SQL bodies, tenants, checkpoint values, provider payloads, +audit rows, and raw exception text. SHA-256 is change-identification evidence, +not a signature, provenance statement, attestation, or release authorization. + +`init-db`, `apply_result_checkpoint_schema()`, and +`apply_result_checkpoint_audit_schema()` remain source compatible. No migration +ledger table, downgrade operation, destructive rollback, version bump, or release +publication is introduced. + +## Rationale + +PostgreSQL transaction-level advisory locking coordinates cooperating operator +invocations without creating another database object or lifecycle table. The lock +waits rather than treating routine concurrent deployment as a fatal race and is +released automatically on commit, rollback, process failure, or connection +termination. One transaction makes the ordered pair all-or-nothing: a failure in +`0008_result_checkpoint_audit_events` rolls back changes made by +`0007_result_stream_checkpoints` during the same command. + +Loading both files before database access addresses a different failure class +than transaction rollback. A missing or oversized second file is a local package +input failure and should not open a database transaction or acquire a lock. +Bounded reads prevent a replaced package file from causing unbounded library-owned +memory use before the size decision. + +A migration ledger was rejected for this slice. The two canonical migrations are +already idempotent, and a ledger would create a second mutable state authority, +new recovery and tamper questions, and compatibility obligations without being +necessary for ordered atomic application. + +## Consequences + +### Positive + +- Existing PostgreSQL volumes receive a beginner-readable, deterministic upgrade + command. +- Concurrent package operators serialize on one reviewed lock namespace. +- The two optional schemas cannot be partially committed by this command. +- Change records can retain bounded migration identity evidence without secrets + or SQL bodies. +- Standalone and modular MSA hosts can adopt the coordinator without requiring + `naruon` or `contextual-orchestrator`. + +### Limitations + +- `pg_advisory_xact_lock` coordinates only clients that use the same lock. A + privileged administrator or unrelated SQL process can still alter schema + outside this boundary. +- SHA-256 identifies bytes but does not authenticate their origin. +- Database roles, backups, maintenance windows, replication, and organization + change approval remain operator responsibilities. +- The command upgrades forward only. It does not delete retained checkpoint or + audit evidence and does not execute packaged rollback SQL. + +## Verification + +Deterministic tests cover descriptor validation, canonical order, bounded reads, +pre-connection failure, one lock, one transaction, one commit, canonical body-free +JSON, and unchanged `init-db` behavior. Live PostgreSQL tests prove that invalid +migration 0008 rolls back migration 0007 and that a concurrent invocation waits +on the transaction-level advisory lock before completing. CI retains read-only +repository permissions and credential-free checkout. + +## Standards and references + +NIST SP 800-53 Rev. 5 CM-3 and CM-3(2) motivate controlled, tested, validated, +and documented configuration changes. PostgreSQL 18 documents transaction blocks +as all-or-nothing, `ROLLBACK` as discarding the transaction's updates, and +`pg_advisory_xact_lock` as an exclusive transaction-level lock released at +transaction end. + +`init-checkpoint-storage`, `0007_result_stream_checkpoints`, +`0008_result_checkpoint_audit_events`, `pg_advisory_xact_lock`, and SHA-256 are +therefore part of the authoritative migration contract. diff --git a/docs/checkpoint-storage-migrations.md b/docs/checkpoint-storage-migrations.md new file mode 100644 index 000000000..6ace998bf --- /dev/null +++ b/docs/checkpoint-storage-migrations.md @@ -0,0 +1,108 @@ +# Checkpoint storage migrations + +## Purpose + +`pg-llm-batch` keeps durable result checkpoints and accepted-save audit evidence +opt-in. Fresh bundled PostgreSQL data directories receive their schemas through +ordered container initialization, but **existing PostgreSQL volumes** need an +explicit upgrade action. The supported operator command is: + +```bash +python -m pg_llm_batch init-checkpoint-storage \ + --dsn 'postgresql://operator@database/pg_llm_batch' +``` + +`init-db` remains the backward-compatible core-schema command. It does not +silently install checkpoint or audit tables. Hosts that intentionally control +separate transactions may continue to call `apply_result_checkpoint_schema()` +and `apply_result_checkpoint_audit_schema()` independently. + +## What the command does + +The operator performs one deterministic sequence: + +```text +init-checkpoint-storage + ├─ bounded read + UTF-8 validation + SHA-256 of + │ 0007_result_stream_checkpoints + ├─ bounded read + UTF-8 validation + SHA-256 of + │ 0008_result_checkpoint_audit_events + ├─ connect only after both files are valid + ├─ SELECT pg_advisory_xact_lock(PGLM, BATH) + ├─ execute 0007_result_stream_checkpoints + ├─ execute 0008_result_checkpoint_audit_events + └─ one commit, then bounded JSON evidence +``` + +Each canonical SQL file is read with a maximum request of 1 MiB plus one byte. +An empty file, a file larger than 1 MiB, invalid UTF-8, or an unreadable second +file fails **before database access**. This prevents migration 0007 from being +committed merely because migration 0008 could not be loaded. + +Inside PostgreSQL, the command obtains the fixed two-key transaction-level +advisory lock with `pg_advisory_xact_lock`. Cooperating package operators using +the same command serialize. The lock is released automatically when the +transaction ends. It does not prevent a privileged administrator or unrelated +SQL client from changing schemas outside this package boundary. + +Both migrations run in one transaction and the package issues one commit only +after migration 0008 succeeds. A lock failure or SQL failure in migration 0008 +rolls back migration 0007 from the same invocation. The command never executes +rollback scripts, downgrades schemas, deletes checkpoints, or erases retained +audit evidence. + +## Success output + +After commit, the command prints one canonical JSON object: + +```json +{ + "applied_migrations": [ + { + "byte_count": 123, + "migration_id": "0007_result_stream_checkpoints", + "sha256": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + }, + { + "byte_count": 456, + "migration_id": "0008_result_checkpoint_audit_events", + "sha256": "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789" + } + ], + "schema_version": 1 +} +``` + +The exact values depend on the reviewed package bytes. Output never contains the +DSN, credentials, SQL text, tenant identity, checkpoint values, provider bodies, +or audit rows. SHA-256 is deterministic change-identification evidence for the +loaded migration bytes; it is **not a signature**, authenticated provenance, +publication authority, or remote attestation. + +## Deployment procedure + +1. Back up the target database under the host's normal recovery policy. +2. Confirm the package artifact and version approved for the environment. +3. Run `plan_checkpoint_schema_migrations()` in a read-only preparation step + when a change record needs byte counts and digests before database access. +4. Run `init-checkpoint-storage` once with an operator identity authorized to + create and alter the two package tables, policies, indexes, functions, and + triggers. +5. Retain the canonical JSON with the deployment change record. +6. Run the command again in non-production rehearsal and upgrade tests to prove + idempotency before rollout to additional environments. +7. Verify application roles remain `NOSUPERUSER NOBYPASSRLS` and receive only + the least privileges documented for checkpoint and audit operations. + +A failed command emits no success JSON. Preserve the original exception and +PostgreSQL logs under the host's restricted operator process; do not paste DSNs, +credentials, or unbounded SQL errors into public tickets. + +## Modular use + +The operator is standalone and has no dependency on `naruon` or +`contextual-orchestrator`. CWL hosts may call +`apply_checkpoint_schema_migrations()` during their own deployment workflow and +store the returned descriptors in a tenant-neutral change record. They must not +reinterpret the descriptors as tenant authorization or integrated-release +provenance. diff --git a/docs/doctoring/checkpoint-migration-operator.md b/docs/doctoring/checkpoint-migration-operator.md new file mode 100644 index 000000000..fb948ea8a --- /dev/null +++ b/docs/doctoring/checkpoint-migration-operator.md @@ -0,0 +1,178 @@ +# Checkpoint migration operator assurance record + +## Assurance objective + +Provide one bounded operator workflow for existing PostgreSQL volumes that loads, +identifies, serializes, atomically applies, and reports the package's durable +checkpoint and checkpoint accepted-save audit migrations without changing the +backward-compatible core `init-db` command. + +The command is `init-checkpoint-storage`. Its exact ordered inputs are +`0007_result_stream_checkpoints` and +`0008_result_checkpoint_audit_events`. + +## Control boundary + +The package controls: + +- a 1 MiB maximum for each canonical migration file, enforced with a read request + of at most the limit plus one detection byte; +- strict UTF-8 decoding and SHA-256 identity evidence before database access; +- exact 0007 then 0008 ordering; +- one fixed two-key transaction-level `pg_advisory_xact_lock`; +- one PostgreSQL transaction and one commit after both statements succeed; +- rollback of migration 0007 when migration 0008 fails in the same invocation; +- bounded success output containing only migration identifier, byte count, and + SHA-256; and +- preservation of the existing `init-db` and separate-helper interfaces. + +The package does not control administrator behavior, arbitrary SQL clients, +backup policy, database clocks, replication, credential issuance, operating +system access, or organizational change approval. A PostgreSQL owner, +superuser, administrator, or client that does not obtain the same advisory lock +can execute schema changes outside this cooperative boundary. + +SHA-256 identifies the exact loaded bytes for deterministic change comparison. It +is not a signature, authenticated provenance, remote attestation, code-signing +claim, publication authority, or integrated-release approval. + +## Transaction semantics + +PostgreSQL transaction blocks group statements into one all-or-nothing unit. +`init-checkpoint-storage` loads both files before calling psycopg, then opens one +connection, obtains `pg_advisory_xact_lock`, executes +`0007_result_stream_checkpoints`, executes +`0008_result_checkpoint_audit_events`, and issues one commit. An exception before +that commit leaves the connection context through its failure path, so PostgreSQL +rolls back the transaction and releases the transaction-level lock. + +The lock keys are the reviewed signed 32-bit values for ASCII namespaces `PGLM` +and `BATH`. The two-key API reduces accidental collision with unrelated +application advisory locks while remaining constant across standalone and modular +MSA deployments. + +Transaction-level advisory locks wait for conflicting holders and release +automatically at transaction end. They avoid stale lock rows and a new migration +ledger table. They coordinate only cooperating invocations; they are not an +authorization mechanism. + +## Input and memory safety + +Each file is opened in binary mode and read once with +`MAX_CHECKPOINT_SCHEMA_MIGRATION_BYTES + 1`. This establishes a strict package- +owned memory ceiling while still detecting one byte beyond the 1 MiB contract. +Empty, oversized, unreadable, or invalid UTF-8 files fail before database access. + +The loaded SQL text remains private. The public immutable +`CheckpointSchemaMigration` exposes only: + +- `migration_id` from the configured ordered plan; +- positive `byte_count` within the 1 MiB limit; and +- lowercase 64-character hexadecimal `sha256`. + +The CLI serializes these fields with a fixed schema version only after successful +commit. It never emits the DSN, password, SQL body, tenant, checkpoint record, +provider response, audit row, or raw database exception. + +## Compatibility and modularity + +`init-db` continues to apply only the core batch schema. The existing +`apply_result_checkpoint_schema()` and +`apply_result_checkpoint_audit_schema()` functions retain their independent +transaction behavior for hosts that intentionally own orchestration. + +`apply_checkpoint_schema_migrations()` is independently usable and does not +require `naruon`, `contextual-orchestrator`, a model provider, or an LLM key. +CWL hosts may store returned descriptors in their deployment change record but +must not reinterpret them as tenant identity or release provenance. + +Fresh Docker data directories keep their ordered initialization scripts. Docker +entrypoint initialization is not an upgrade mechanism for existing PostgreSQL +volumes, which use the explicit operator command. + +## Deterministic verification matrix + +- Public RED head `9aedff7a50270a81cd245771dbe7f649a31fe66f`: + CI run `31149845743`, Python 3.10 job `92777048403`, failed collection because + `CheckpointSchemaMigration` and the migration coordinator did not exist. +- Unit GREEN head `c2de384dd5a42a026f1c49d9eeec92c5bff3217f`: + Python 3.14 unit tests passed; the quality job completed compilation, Ruff, + 100% public docstrings, 100% production line coverage, lock freshness, and + package build before the next test-first commit superseded the run. Superseded + workflow evidence is not final merge evidence. +- Bounded-read contract head `049e6aab513494b114fddfe2e0679dc1ff19e921` + introduced a test that rejects `Path.read_bytes()` and any negative-size read. + Its workflow was superseded before completion, so it is retained only as + test-first source history and is not counted as executed RED or GREEN evidence. +- Live-workflow contract head `6a7ac0125ea67ee60b88d41d3a6a9d56cf8cef75` + required the permanent PostgreSQL job to run both audit and migration operator + integration suites. Its workflow was likewise superseded before completion and + is not counted as a successful or failed gate. +- Exact-head checkout RED head `2cb3607a7b9c696c58fa0df72743cd3667822685`: + CI run `31152186361`, quality job `92783956312`, passed 423 deterministic tests + and then failed the newly added workflow contract because the required CI jobs + still checked out GitHub's pull-request merge ref rather than the exact source + head. That failure is not reused as success. +- Exact-head GREEN head `25e8e493695c227e4510647305574025c0ec58eb`: + CI run `31152516949` checked out and asserted this exact SHA independently in + every required unit, quality, live PostgreSQL, and container job. All jobs + completed successfully. The quality job `92784969375` reported 658 passed and + 8 deselected, 2,514 production statements and 628 production branches at 100%, + 100% public docstrings, Ruff success, a fresh lockfile, and successful wheel and + source-distribution builds. The live PostgreSQL job proved both accepted-save + audit behavior and migration rollback/advisory-lock serialization. Release + Acceptance run `31152516946`, job `92784969354`, independently checked out the + exact head and proved reproducible wheel and source-distribution identity. +- Unit tests prove immutable descriptors, strict non-coercive field validation, + exact order/digests, empty/oversized/non-UTF-8 rejection, pre-connection input + failure, one advisory lock, exact SQL order, one commit, and failure propagation + without success evidence. +- Live PostgreSQL tests prove an invalid second migration rolls back the first and + a concurrent worker has an ungranted advisory lock until the holder commits. +- CI contract tests bind the pinned PostgreSQL image, DSN, credential-free exact- + head checkout, and exact integration command to the reviewed job rather than a + repository-wide string search. +- Every later documentation or review-fix commit still requires fresh exact-head + CI and release acceptance. Final integration also requires reconciliation onto + the actual integrated base; successful stacked-base evidence is not reusable as + final release evidence. +- Final acceptance requires fresh exact-head/current-base CI, security, + dependency, packaging, migration, rollback, concurrency, container, + reproducibility, review, and branch-protection evidence. Queued, pending, + cancelled, skipped-required, absent, stale-head, stale-base, and synthetic- + merge-only results are not success. + +## Standards mapping + +NIST SP 800-53 Rev. 5 CM-3 requires configuration changes to be controlled, +documented, and reviewed. CM-3(2) requires changes to be tested, validated, and +documented before finalization. The ordered plan, immutable SHA-256 descriptors, +live rollback/concurrency tests, operator guide, ADR, and exact-head CI evidence +support those objectives. + +PostgreSQL 18 system administration documentation defines +`pg_advisory_xact_lock` as an exclusive transaction-level advisory lock that +waits when necessary and is released automatically at transaction end. The +PostgreSQL transaction tutorial explains that a transaction groups multiple +steps into one all-or-nothing operation, and the `ROLLBACK` reference states that +rollback discards all updates in the current transaction. + +These references support coordination and atomicity, not administrator-proof +schema control. Least-privilege roles, restricted direct SQL, backups, and +separately governed release provenance remain necessary controls. + +## APA 7 references + +Joint Task Force. (2020). *Security and privacy controls for information systems +and organizations* (NIST Special Publication 800-53, Revision 5). National +Institute of Standards and Technology. https://doi.org/10.6028/NIST.SP.800-53r5 + +PostgreSQL Global Development Group. (2026). *System administration functions* +(PostgreSQL 18 documentation). +https://www.postgresql.org/docs/18/functions-admin.html + +PostgreSQL Global Development Group. (2026). *Transactions* (PostgreSQL 18 +documentation). https://www.postgresql.org/docs/18/tutorial-transactions.html + +PostgreSQL Global Development Group. (2026). *ROLLBACK* (PostgreSQL 18 +documentation). https://www.postgresql.org/docs/18/sql-rollback.html diff --git a/docs/superpowers/plans/2026-08-07-checkpoint-migration-operator.md b/docs/superpowers/plans/2026-08-07-checkpoint-migration-operator.md new file mode 100644 index 000000000..a88981c00 --- /dev/null +++ b/docs/superpowers/plans/2026-08-07-checkpoint-migration-operator.md @@ -0,0 +1,423 @@ +# Checkpoint Migration Operator Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Add one opt-in operator command that validates, serializes, atomically applies, and reports the durable checkpoint plus checkpoint-audit PostgreSQL migrations. + +**Architecture:** A focused `checkpoint_migrations` module loads both canonical SQL files before database access, creates immutable bounded descriptors, obtains one fixed transaction-level PostgreSQL advisory lock, executes migration 0007 before 0008, and commits once. The existing separate helpers and `init-db` remain unchanged; the new CLI command emits bounded JSON evidence only after commit. + +**Tech Stack:** Python 3.10+, psycopg 3, PostgreSQL 18 advisory locks and transaction semantics, argparse, hashlib, pytest, GitHub Actions, existing uv/coverage/Ruff toolchain. + +## Global Constraints + +- Base exact head: `2820aa36d8dedf7d89d1b745e5728acf3b913d2b` from stacked PR #62. +- Branch: `agent/checkpoint-migration-operator`; do not add a competing writer or temporary repair workflow. +- Preserve `init-db`, `apply_result_checkpoint_schema()`, and `apply_result_checkpoint_audit_schema()` behavior. +- Load both canonical migration files before database connection. +- Accept only non-empty files of at most 1,048,576 bytes each. +- Execute `0007_result_stream_checkpoints` before `0008_result_checkpoint_audit_events` under one transaction-level advisory lock and one commit. +- Emit no DSN, credential, SQL body, provider body, tenant, checkpoint, audit-row, or raw database-error data in success evidence. +- Keep every database object name descriptive and snake_case. +- Maintain 100% production statement, branch, and public-docstring coverage. +- Update AGENTS.md, CLAUDE.md, ARCHITECTURE.md, CHANGELOG.md, README.md, ADR, operator documentation, and doctoring when the contract changes. +- No version bump or release publication until the integrated exact head passes every repository gate and independent approval. + +--- + +### Task 1: Establish the RED migration coordinator contract + +**Files:** +- Create: `tests/test_checkpoint_migration_operator.py` +- Modify: `tests/test_bootstrap_cli.py` + +**Interfaces:** +- Consumes: canonical paths `checkpoint_store.MIGRATION_PATH` and `checkpoint_audit.AUDIT_MIGRATION_PATH`. +- Produces: required public names `CheckpointSchemaMigration`, `plan_checkpoint_schema_migrations()`, and `apply_checkpoint_schema_migrations(postgres_dsn)` plus CLI command `init-checkpoint-storage`. + +- [ ] **Step 1: Write the failing public contract tests** + +```python +from hashlib import sha256 +from pg_llm_batch import ( + CheckpointSchemaMigration, + apply_checkpoint_schema_migrations, + plan_checkpoint_schema_migrations, +) + + +def test_plan_uses_canonical_order_and_digests(): + plan = plan_checkpoint_schema_migrations() + assert [item.migration_id for item in plan] == [ + "0007_result_stream_checkpoints", + "0008_result_checkpoint_audit_events", + ] + assert all(isinstance(item, CheckpointSchemaMigration) for item in plan) + assert plan[0].sha256 == sha256(MIGRATION_PATH.read_bytes()).hexdigest() +``` + +Add tests proving immutable descriptors, positive byte counts, lowercase 64-byte hex digests, non-empty/1 MiB bounds before `psycopg.connect`, one advisory lock before SQL, exact SQL order, one commit, and no commit after the second statement fails. + +- [ ] **Step 2: Add the failing CLI route test** + +```python +def test_checkpoint_storage_initialization_emits_bounded_json(monkeypatch, capsys): + monkeypatch.setattr(cli, "apply_checkpoint_schema_migrations", lambda dsn: plan) + assert cli._dispatch([ + "init-checkpoint-storage", "--dsn", "postgresql://x" + ]) == 0 + assert json.loads(capsys.readouterr().out) == { + "schema_version": 1, + "applied_migrations": [descriptor.as_dict() for descriptor in plan], + } +``` + +- [ ] **Step 3: Run the exact RED tests** + +Run: + +```bash +uv run pytest -q \ + tests/test_checkpoint_migration_operator.py \ + tests/test_bootstrap_cli.py +``` + +Expected: collection or import failure because `checkpoint_migrations` and its public exports do not exist. + +- [ ] **Step 4: Commit the RED contract** + +```bash +git add tests/test_checkpoint_migration_operator.py tests/test_bootstrap_cli.py +git commit -m "test(migrations): require atomic checkpoint schema operator" +``` + +### Task 2: Implement bounded planning and atomic application + +**Files:** +- Create: `pg_llm_batch/checkpoint_migrations.py` +- Modify: `pg_llm_batch/__init__.py` + +**Interfaces:** +- Consumes: `checkpoint_store.MIGRATION_PATH`, `checkpoint_audit.AUDIT_MIGRATION_PATH`, `db._require_psycopg`, and `db.psycopg`. +- Produces: + - `CheckpointSchemaMigration(migration_id: str, byte_count: int, sha256: str)` + - `plan_checkpoint_schema_migrations() -> tuple[CheckpointSchemaMigration, ...]` + - `apply_checkpoint_schema_migrations(postgres_dsn: str) -> tuple[CheckpointSchemaMigration, ...]` + +- [ ] **Step 1: Implement the immutable public descriptor** + +```python +@dataclass(frozen=True, slots=True) +class CheckpointSchemaMigration: + """Describe one bounded canonical checkpoint-storage migration.""" + + migration_id: str + byte_count: int + sha256: str + + def as_dict(self) -> dict[str, int | str]: + """Return a stable JSON-compatible evidence object.""" + return { + "migration_id": self.migration_id, + "byte_count": self.byte_count, + "sha256": self.sha256, + } +``` + +Validate exact supported identifiers, positive byte count at most 1 MiB, and lowercase 64-character hexadecimal SHA-256 without coercion. + +- [ ] **Step 2: Load and hash all canonical SQL before database access** + +```python +def _load_checkpoint_schema_migrations() -> tuple[_LoadedMigration, ...]: + loaded = [] + for migration_id, path in _CHECKPOINT_SCHEMA_MIGRATION_PATHS: + sql_bytes = path.read_bytes() + if not sql_bytes or len(sql_bytes) > MAX_CHECKPOINT_SCHEMA_MIGRATION_BYTES: + raise RuntimeError("checkpoint schema migration has an invalid bounded size") + loaded.append(_LoadedMigration( + descriptor=CheckpointSchemaMigration( + migration_id=migration_id, + byte_count=len(sql_bytes), + sha256=hashlib.sha256(sql_bytes).hexdigest(), + ), + sql=sql_bytes.decode("utf-8", errors="strict"), + )) + return tuple(loaded) +``` + +Keep `_LoadedMigration.sql` private so public evidence never exposes SQL text. + +- [ ] **Step 3: Apply the plan under one transaction-level advisory lock** + +```python +def apply_checkpoint_schema_migrations(postgres_dsn: str): + """Apply checkpoint and audit schema atomically in canonical order.""" + loaded = _load_checkpoint_schema_migrations() + _require_psycopg() + with psycopg.connect(postgres_dsn) as connection: + with connection.cursor() as cursor: + cursor.execute( + "SELECT pg_advisory_xact_lock(%s, %s)", + ( + CHECKPOINT_SCHEMA_MIGRATION_LOCK_NAMESPACE, + CHECKPOINT_SCHEMA_MIGRATION_LOCK_OPERATION, + ), + ) + for migration in loaded: + cursor.execute(migration.sql) + connection.commit() + return tuple(migration.descriptor for migration in loaded) +``` + +Use constants `1346849869` (`PGLM`) and `1111577672` (`BATH`) as the reviewed two-key lock namespace. Do not catch and reclassify database exceptions with raw messages. + +- [ ] **Step 4: Export the public contract** + +Add all three public names to `pg_llm_batch.__init__` and `__all__`, and describe the coordinator in the module-level public API docstring. + +- [ ] **Step 5: Run focused tests and refactor** + +Run: + +```bash +uv run pytest -q tests/test_checkpoint_migration_operator.py +uv run ruff check pg_llm_batch/checkpoint_migrations.py tests/test_checkpoint_migration_operator.py +``` + +Expected: PASS with every production branch covered by focused tests. + +- [ ] **Step 6: Commit the GREEN implementation** + +```bash +git add pg_llm_batch/checkpoint_migrations.py pg_llm_batch/__init__.py +git commit -m "feat(migrations): apply checkpoint schemas atomically" +``` + +### Task 3: Add the explicit operator CLI + +**Files:** +- Modify: `pg_llm_batch/cli.py` +- Modify: `tests/test_bootstrap_cli.py` + +**Interfaces:** +- Consumes: `apply_checkpoint_schema_migrations(postgres_dsn)` and each descriptor's `as_dict()`. +- Produces: `python -m pg_llm_batch init-checkpoint-storage --dsn ...`. + +- [ ] **Step 1: Add parser and dispatch behavior** + +```python +p_checkpoint = sub.add_parser( + "init-checkpoint-storage", + help="Atomically apply checkpoint and checkpoint-audit schemas", +) +_add_common(p_checkpoint) +``` + +Dispatch only after DSN resolution: + +```python +if args.command == "init-checkpoint-storage": + applied = apply_checkpoint_schema_migrations(dsn) + print(json.dumps({ + "schema_version": 1, + "applied_migrations": [item.as_dict() for item in applied], + }, separators=(",", ":"), sort_keys=True)) + return 0 +``` + +- [ ] **Step 2: Verify canonical output and unchanged init-db routing** + +Run: + +```bash +uv run pytest -q tests/test_bootstrap_cli.py +``` + +Expected: PASS. Existing `init-db` test still observes only `db.apply_schema`. + +- [ ] **Step 3: Commit the CLI** + +```bash +git add pg_llm_batch/cli.py tests/test_bootstrap_cli.py +git commit -m "feat(cli): add checkpoint storage migration command" +``` + +### Task 4: Prove live rollback and serialization + +**Files:** +- Create: `tests/test_checkpoint_migration_operator_integration.py` +- Modify: `.github/workflows/ci.yml` +- Modify: `tests/test_checkpoint_audit_ci_contract.py` + +**Interfaces:** +- Consumes: public migration coordinator and internal canonical migration path tuple via monkeypatch for bounded fault injection. +- Produces: permanent live PostgreSQL all-or-nothing and advisory-lock evidence. + +- [ ] **Step 1: Write live all-or-nothing regression** + +Create a random temporary database. Copy migration 0007 to a temporary file and use an invalid second SQL file. Invoke the coordinator and require a PostgreSQL error. Reconnect as owner and assert both +`to_regclass('public.llm_result_stream_checkpoints')` and +`to_regclass('public.llm_result_checkpoint_audit_events')` are `NULL`. + +- [ ] **Step 2: Write live lock serialization regression** + +Hold the same two-key `pg_advisory_xact_lock` in one connection. Start the coordinator in a second thread, require it not to complete while the first transaction holds the lock, release the first transaction, and require successful completion. Use bounded thread events and timeouts; always terminate the temporary database sessions in cleanup. + +- [ ] **Step 3: Add the exact file to the permanent PostgreSQL job** + +Change only the checkpoint-audit integration command to: + +```yaml +run: >- + uv run pytest -q + tests/test_checkpoint_audit_integration.py + tests/test_checkpoint_migration_operator_integration.py + -m integration +``` + +Update the workflow contract test so it binds this exact folded command to the exact `checkpoint-audit-integration` job and cannot borrow strings from another job. + +- [ ] **Step 4: Run live verification** + +Run: + +```bash +PG_LLM_BATCH_TEST_DSN=postgresql://postgres:postgres@localhost:5432/postgres \ + uv run pytest -q \ + tests/test_checkpoint_audit_integration.py \ + tests/test_checkpoint_migration_operator_integration.py \ + -m integration +``` + +Expected: PASS; no temporary database or role remains. + +- [ ] **Step 5: Commit live evidence** + +```bash +git add \ + tests/test_checkpoint_migration_operator_integration.py \ + tests/test_checkpoint_audit_ci_contract.py \ + .github/workflows/ci.yml +git commit -m "test(migrations): prove rollback and advisory serialization" +``` + +### Task 5: Make the operator contract authoritative + +**Files:** +- Modify: `README.md` +- Modify: `AGENTS.md` +- Modify: `CLAUDE.md` +- Modify: `ARCHITECTURE.md` +- Modify: `CHANGELOG.md` +- Create: `docs/adr/0010-atomic-checkpoint-schema-operator.md` +- Create: `docs/checkpoint-storage-migrations.md` +- Create: `docs/doctoring/checkpoint-migration-operator.md` +- Create: `tests/test_checkpoint_migration_operator_documentation.py` + +**Interfaces:** +- Consumes: final production and CLI contracts. +- Produces: beginner-readable operator workflow, design authority, failure recovery, and APA 7 evidence. + +- [ ] **Step 1: Write the failing documentation contract** + +Require each authoritative file to state all of: + +- explicit opt-in `init-checkpoint-storage`; +- canonical 0007 → 0008 order; +- both files loaded before database access; +- one transaction-level advisory lock and automatic release; +- one transaction and one commit; +- second-migration failure rolls back the first; +- digest evidence is not signature or attestation; +- `init-db` and separate helpers remain compatible; and +- no migration ledger, downgrade, deletion, or retained-evidence rollback. + +- [ ] **Step 2: Update public and contributor documentation** + +Add a quick-start command and architecture flow: + +```text +init-checkpoint-storage + ├─ load + bound + hash migration 0007 + ├─ load + bound + hash migration 0008 + ├─ pg_advisory_xact_lock(PGLM, BATH) + ├─ execute 0007 → 0008 + └─ commit once → bounded JSON evidence +``` + +- [ ] **Step 3: Record standards in APA 7 form** + +Use NIST SP 800-53 Rev. 5 CM-3/CM-3(2), PostgreSQL 18 system administration functions, transactions, and ROLLBACK documentation. Explain that advisory locking coordinates cooperating operator invocations and does not stop a privileged actor from executing unrelated SQL outside the package boundary. + +- [ ] **Step 4: Run documentation and focused test gates** + +Run: + +```bash +uv run pytest -q \ + tests/test_checkpoint_migration_operator.py \ + tests/test_checkpoint_migration_operator_integration.py \ + tests/test_checkpoint_migration_operator_documentation.py \ + tests/test_bootstrap_cli.py \ + tests/test_checkpoint_audit_ci_contract.py +``` + +Expected: PASS. + +- [ ] **Step 5: Commit authoritative documentation** + +```bash +git add README.md AGENTS.md CLAUDE.md ARCHITECTURE.md CHANGELOG.md \ + docs/adr/0010-atomic-checkpoint-schema-operator.md \ + docs/checkpoint-storage-migrations.md \ + docs/doctoring/checkpoint-migration-operator.md \ + tests/test_checkpoint_migration_operator_documentation.py +git commit -m "docs(migrations): define atomic operator contract" +``` + +### Task 6: Exact-head verification and draft PR gate + +**Files:** +- Update: pull-request description only after exact-head evidence exists. + +**Interfaces:** +- Consumes: complete branch. +- Produces: one stacked draft PR targeting `agent/checkpoint-audit-trail`. + +- [ ] **Step 1: Run deterministic local gates** + +```bash +uv sync --locked --extra test +uv run ruff check . +uv run python -m compileall -q pg_llm_batch tests +uv run pytest -q -m "not integration" \ + --cov=pg_llm_batch --cov-branch --cov-report=term-missing \ + --cov-fail-under=100 +``` + +Require 100% production statement and branch coverage, 100% public docstrings, +lock freshness, package build, Compose validation, and no generated artifacts. + +- [ ] **Step 2: Open one stacked draft PR** + +Target `agent/checkpoint-audit-trail`. Record exact head and exact base SHAs, RED +and GREEN runs, no unresolved valid feedback, and the dependency order: + +```text +.github#790 -> #53 -> #55 -> #56 -> #57 -> #58 -> #59 -> #60 -> #61 -> #62 -> this PR +``` + +- [ ] **Step 3: Inspect every exact-head check and review** + +Do not count queued, pending, cancelled, skipped-required, absent, stale-head, +stale-base, synthetic-merge-only, or infrastructure-only evidence as success. +Address every valid human, CodeRabbit, OpenCode, Noema, Dependabot, code-scanning, +security, and supply-chain finding test-first. + +- [ ] **Step 4: Keep the PR draft and unmerged** + +Do not mark ready, version-bump, publish, attest, or merge until all prerequisites +are integrated into `main`, this branch is reconciled onto that exact integrated +base, every required current-head/current-base gate succeeds, unresolved valid +findings are zero, and a qualifying independent non-author GitHub `APPROVED` +review exists. diff --git a/docs/superpowers/specs/2026-08-07-checkpoint-migration-operator-design.md b/docs/superpowers/specs/2026-08-07-checkpoint-migration-operator-design.md new file mode 100644 index 000000000..a957e9497 --- /dev/null +++ b/docs/superpowers/specs/2026-08-07-checkpoint-migration-operator-design.md @@ -0,0 +1,150 @@ +# Checkpoint Migration Operator Design + +## Status + +Approved bounded vertical slice for implementation on the stacked branch +`agent/checkpoint-migration-operator`. The exact prerequisite is PR #62 head +`2820aa36d8dedf7d89d1b745e5728acf3b913d2b`. + +## Buyer-visible gap + +The package exposes durable checkpoint and checkpoint-audit migrations, but an +operator must currently know and call two independent Python helpers in the +correct order. Each helper opens and commits its own transaction. Existing +PostgreSQL volumes therefore lack one package-owned command that: + +- validates both canonical migration inputs before database mutation; +- serializes concurrent migration attempts; +- applies durable checkpoint schema before checkpoint-audit schema; +- commits the ordered pair atomically or rolls both back; +- returns bounded machine-readable migration identity evidence; and +- preserves the existing opt-in boundary instead of silently changing + `init-db`. + +This is an acquisition-readiness gap because migration ordering and rollback +are currently procedural knowledge rather than a deterministic operator +contract. + +## Decision + +Add a focused `pg_llm_batch.checkpoint_migrations` module and the explicit CLI +command `init-checkpoint-storage`. + +The module will load the canonical package migrations +`0007_result_stream_checkpoints.sql` and +`0008_result_checkpoint_audit_events.sql` in that exact order before opening a +database connection. Each file is non-empty, bounded to 1 MiB, and represented +by an immutable descriptor containing only a stable migration identifier, +byte count, and SHA-256 digest. + +Application will use one PostgreSQL connection and one transaction. Before the +first migration statement, it obtains the fixed two-key transaction-level +advisory lock: + +```sql +SELECT pg_advisory_xact_lock(%s, %s) +``` + +The reviewed keys are package constants derived from the stable ASCII namespace +`PGLM` and operation `BATH`. Transaction-level advisory locks wait for a +competing holder and release automatically at transaction end. The operator +then executes migration 0007 followed by 0008 and commits once. Any load, +connection, lock, or SQL failure propagates before a success report; PostgreSQL +rolls the transaction back and releases the lock. + +The CLI emits one canonical JSON object only after commit: + +```json +{ + "schema_version": 1, + "applied_migrations": [ + { + "migration_id": "0007_result_stream_checkpoints", + "byte_count": 123, + "sha256": "..." + }, + { + "migration_id": "0008_result_checkpoint_audit_events", + "byte_count": 456, + "sha256": "..." + } + ] +} +``` + +The report excludes the DSN, credentials, SQL text, database exception text, +tenant identifiers, checkpoint values, and audit rows. SHA-256 is deterministic +change-identification evidence for the applied package bytes, not a signature or +remote attestation. + +## Compatibility boundary + +- `init-db` remains unchanged and continues to apply only the existing core + schema. +- Existing public `apply_result_checkpoint_schema()` and + `apply_result_checkpoint_audit_schema()` helpers remain source compatible for + hosts that intentionally manage separate transactions. +- The new coordinator is opt-in and independently usable without `naruon` or + `contextual-orchestrator`. +- Fresh Docker data directories keep their existing ordered entrypoint + migrations. Existing volumes use the explicit operator command. +- No migration ledger table is introduced. The canonical migrations are already + idempotent; adding a second state authority would increase recovery and + tamper-analysis scope without solving the bounded operator gap. + +## Failure and recovery contract + +- Both migration files are loaded and hashed before database access. A missing, + empty, unreadable, or oversized second file cannot leave migration 0007 + partially applied. +- The advisory lock is transaction-scoped. Process failure, rollback, or + connection loss releases it through PostgreSQL transaction termination. +- A failure in migration 0008 rolls back migration 0007 from the same operator + invocation. +- A successful rerun is supported by the existing idempotent SQL and emits the + same ordered identity evidence for unchanged package bytes. +- The command does not downgrade, delete retained checkpoint/audit evidence, or + invoke rollback scripts. + +## Verification design + +Strict RED → GREEN tests will prove: + +1. canonical order, identifiers, byte counts, and SHA-256 values; +2. non-empty and 1 MiB input bounds before database connection; +3. one connection, one transaction-level advisory lock, exact SQL order, and + one commit; +4. no success report or commit after the second migration fails; +5. stable public exports and a canonical body-free CLI report; +6. live PostgreSQL all-or-nothing behavior with an intentionally invalid second + migration; +7. live concurrent invocations serialize on the same advisory lock; +8. unchanged `init-db` behavior and separate-helper compatibility; and +9. synchronized README, AGENTS, CLAUDE, architecture, ADR, doctoring, operator, + and changelog contracts with 100% production statement, branch, and public + docstring coverage. + +## Standards basis + +NIST SP 800-53 Rev. 5 CM-3 requires controlled, documented, and reviewed system +changes, while CM-3(2) requires testing, validation, and documentation before +finalization. PostgreSQL 18 documents transaction blocks as all-or-nothing and +`pg_advisory_xact_lock` as an exclusive transaction-level advisory lock that is +released automatically at transaction end. This design uses those controls to +turn two migration scripts into one deterministic, testable operator action. + +## APA 7 references + +Joint Task Force. (2020). *Security and privacy controls for information systems +and organizations* (NIST Special Publication 800-53, Revision 5). National +Institute of Standards and Technology. https://doi.org/10.6028/NIST.SP.800-53r5 + +PostgreSQL Global Development Group. (2026). *System administration functions* +(PostgreSQL 18 documentation). +https://www.postgresql.org/docs/18/functions-admin.html + +PostgreSQL Global Development Group. (2026). *Transactions* (PostgreSQL 18 +documentation). https://www.postgresql.org/docs/18/tutorial-transactions.html + +PostgreSQL Global Development Group. (2026). *ROLLBACK* (PostgreSQL 18 +documentation). https://www.postgresql.org/docs/18/sql-rollback.html diff --git a/pg_llm_batch/__init__.py b/pg_llm_batch/__init__.py index e88f354e4..56355d60a 100644 --- a/pg_llm_batch/__init__.py +++ b/pg_llm_batch/__init__.py @@ -10,6 +10,8 @@ BatchResultCheckpoint -- host-persistable resume evidence PostgresBatchResultCheckpointStore -- tenant-isolated durable checkpoints AuditedPostgresBatchResultCheckpointStore -- append-only accepted-save audit + CheckpointSchemaMigration -- bounded migration identity evidence + apply_checkpoint_schema_migrations -- atomic checkpoint schema operator OpenTelemetryCheckpointStore -- confidential checkpoint observability DurableBatchAPIClient -- standalone durable lifecycle state TenantDurableBatchAPIClient -- tenant-isolated lifecycle state @@ -29,6 +31,11 @@ apply_result_checkpoint_audit_schema, validate_checkpoint_audit_limit, ) +from .checkpoint_migrations import ( + CheckpointSchemaMigration, + apply_checkpoint_schema_migrations, + plan_checkpoint_schema_migrations, +) from .checkpoint_store import ( CheckpointConflictError, PostgresBatchResultCheckpointStore, @@ -76,6 +83,9 @@ "PostgresBatchResultCheckpointStore", "AuditedPostgresBatchResultCheckpointStore", "CheckpointAuditEvent", + "CheckpointSchemaMigration", + "apply_checkpoint_schema_migrations", + "plan_checkpoint_schema_migrations", "OpenTelemetryCheckpointStore", "CheckpointConflictError", "apply_result_checkpoint_schema", diff --git a/pg_llm_batch/checkpoint_migrations.py b/pg_llm_batch/checkpoint_migrations.py new file mode 100644 index 000000000..0121d0c4d --- /dev/null +++ b/pg_llm_batch/checkpoint_migrations.py @@ -0,0 +1,143 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) ContextualWisdomLab. +"""Atomic operator workflow for durable checkpoint and audit migrations.""" + +from __future__ import annotations + +import hashlib +import re +from dataclasses import dataclass +from pathlib import Path + +from .checkpoint_audit import AUDIT_MIGRATION_PATH +from .checkpoint_store import MIGRATION_PATH +from .db import _require_psycopg, psycopg + +MAX_CHECKPOINT_SCHEMA_MIGRATION_BYTES = 1_048_576 +CHECKPOINT_SCHEMA_MIGRATION_LOCK_NAMESPACE = 1_346_849_869 +CHECKPOINT_SCHEMA_MIGRATION_LOCK_OPERATION = 1_111_577_672 +_CHECKPOINT_SCHEMA_MIGRATION_PATHS: tuple[tuple[str, Path], ...] = ( + ("0007_result_stream_checkpoints", MIGRATION_PATH), + ("0008_result_checkpoint_audit_events", AUDIT_MIGRATION_PATH), +) +_SHA256_PATTERN = re.compile(r"[0-9a-f]{64}\Z") + + +def _supported_migration_ids() -> frozenset[str]: + """Return the exact migration identifiers configured for this invocation.""" + return frozenset( + migration_id for migration_id, _path in _CHECKPOINT_SCHEMA_MIGRATION_PATHS + ) + + +@dataclass(frozen=True, slots=True) +class CheckpointSchemaMigration: + """Describe one bounded canonical checkpoint-storage migration.""" + + migration_id: str + byte_count: int + sha256: str + + def __post_init__(self) -> None: + """Reject ambiguous or unbounded public migration evidence.""" + if ( + not isinstance(self.migration_id, str) + or self.migration_id not in _supported_migration_ids() + ): + raise ValueError("migration_id must identify a configured migration") + if ( + isinstance(self.byte_count, bool) + or not isinstance(self.byte_count, int) + or self.byte_count < 1 + or self.byte_count > MAX_CHECKPOINT_SCHEMA_MIGRATION_BYTES + ): + raise ValueError( + "byte_count must be an integer from 1 through " + f"{MAX_CHECKPOINT_SCHEMA_MIGRATION_BYTES}" + ) + if ( + not isinstance(self.sha256, str) + or _SHA256_PATTERN.fullmatch(self.sha256) is None + ): + raise ValueError( + "sha256 must be a lowercase 64-character hexadecimal digest" + ) + + def as_dict(self) -> dict[str, int | str]: + """Return one stable JSON-compatible migration evidence object.""" + return { + "migration_id": self.migration_id, + "byte_count": self.byte_count, + "sha256": self.sha256, + } + + +@dataclass(frozen=True, slots=True) +class _LoadedCheckpointSchemaMigration: + """Keep reviewed SQL private while carrying its public descriptor.""" + + descriptor: CheckpointSchemaMigration + sql: str + + +def _load_checkpoint_schema_migration( + migration_id: str, + migration_path: Path, +) -> _LoadedCheckpointSchemaMigration: + """Load, bound, decode, and identify one migration before database access.""" + with migration_path.open("rb") as migration_file: + sql_bytes = migration_file.read(MAX_CHECKPOINT_SCHEMA_MIGRATION_BYTES + 1) + byte_count = len(sql_bytes) + if byte_count < 1 or byte_count > MAX_CHECKPOINT_SCHEMA_MIGRATION_BYTES: + raise RuntimeError("checkpoint schema migration has an invalid bounded size") + sql = sql_bytes.decode("utf-8", errors="strict") + descriptor = CheckpointSchemaMigration( + migration_id=migration_id, + byte_count=byte_count, + sha256=hashlib.sha256(sql_bytes).hexdigest(), + ) + return _LoadedCheckpointSchemaMigration(descriptor=descriptor, sql=sql) + + +def _load_checkpoint_schema_migrations() -> tuple[_LoadedCheckpointSchemaMigration, ...]: + """Load every configured migration in canonical order before database access.""" + return tuple( + _load_checkpoint_schema_migration(migration_id, migration_path) + for migration_id, migration_path in _CHECKPOINT_SCHEMA_MIGRATION_PATHS + ) + + +def plan_checkpoint_schema_migrations() -> tuple[CheckpointSchemaMigration, ...]: + """Return bounded identity evidence for the canonical migration plan.""" + return tuple( + migration.descriptor for migration in _load_checkpoint_schema_migrations() + ) + + +def apply_checkpoint_schema_migrations( + postgres_dsn: str, +) -> tuple[CheckpointSchemaMigration, ...]: + """Apply checkpoint and audit schema atomically in canonical order. + + Both canonical files are loaded and validated before psycopg or database + access. One package-owned transaction obtains a fixed transaction-level + advisory lock, executes the durable checkpoint migration before the audit + migration, and commits once. Any exception leaves the connection context + before that commit, allowing PostgreSQL to roll the transaction back and + release the advisory lock. + """ + loaded = _load_checkpoint_schema_migrations() + _require_psycopg() + with psycopg.connect(postgres_dsn) as connection: + with connection.cursor() as cursor: + cursor.execute( + "SELECT pg_advisory_xact_lock(%s, %s)", + ( + CHECKPOINT_SCHEMA_MIGRATION_LOCK_NAMESPACE, + CHECKPOINT_SCHEMA_MIGRATION_LOCK_OPERATION, + ), + ) + for migration in loaded: + cursor.execute(migration.sql) + connection.commit() + return tuple(migration.descriptor for migration in loaded) diff --git a/pg_llm_batch/cli.py b/pg_llm_batch/cli.py index 9d137069c..086afc7ec 100644 --- a/pg_llm_batch/cli.py +++ b/pg_llm_batch/cli.py @@ -3,7 +3,8 @@ """Standalone command-line interface: ``python -m pg_llm_batch ...``. Subcommands: - init-db apply the batch schema (idempotent) + init-db apply the core batch schema (idempotent) + init-checkpoint-storage atomically apply checkpoint and audit schemas config set set a KV config value config get read a KV config value config set-secret store a secret (Fernet-encrypted when a key is present) @@ -30,6 +31,7 @@ from . import db from .batch_api_client import BatchAPIClient, config_credentials_provider from .bootstrap import resolve_dsn, resolve_secret_key +from .checkpoint_migrations import apply_checkpoint_schema_migrations from .config import PostgresConfigStore, SecretStore from .exceptions import PgLlmBatchError from .health import check_health, serve_healthz @@ -53,9 +55,15 @@ def build_parser() -> argparse.ArgumentParser: ) sub = parser.add_subparsers(dest="command", required=True) - p_init = sub.add_parser("init-db", help="Apply batch schema (idempotent)") + p_init = sub.add_parser("init-db", help="Apply core batch schema (idempotent)") _add_common(p_init) + p_checkpoint = sub.add_parser( + "init-checkpoint-storage", + help="Atomically apply checkpoint and checkpoint-audit schemas", + ) + _add_common(p_checkpoint) + p_cfg = sub.add_parser("config", help="Manage KV config and secrets") cfg_sub = p_cfg.add_subparsers(dest="config_command", required=True) p_set = cfg_sub.add_parser("set", help="Set a config value") @@ -138,6 +146,20 @@ def _dispatch(argv: Optional[List[str]]) -> int: print("Schema applied.") return 0 + if args.command == "init-checkpoint-storage": + applied = apply_checkpoint_schema_migrations(dsn) + print( + json.dumps( + { + "schema_version": 1, + "applied_migrations": [item.as_dict() for item in applied], + }, + separators=(",", ":"), + sort_keys=True, + ) + ) + return 0 + if args.command == "config": if args.config_command == "set": store = PostgresConfigStore(dsn) diff --git a/tests/test_bootstrap_cli.py b/tests/test_bootstrap_cli.py index a039e7791..09fe1d474 100644 --- a/tests/test_bootstrap_cli.py +++ b/tests/test_bootstrap_cli.py @@ -97,6 +97,49 @@ def set_secret(self, key, value): assert "secret" not in output +def test_checkpoint_storage_initialization_emits_bounded_json(monkeypatch, capsys): + """The opt-in migration command emits only canonical descriptor evidence.""" + calls = [] + + class Descriptor: + def __init__(self, migration_id, byte_count, digest): + self.migration_id = migration_id + self.byte_count = byte_count + self.sha256 = digest + + def as_dict(self): + return { + "migration_id": self.migration_id, + "byte_count": self.byte_count, + "sha256": self.sha256, + } + + plan = ( + Descriptor("0007_result_stream_checkpoints", 123, "a" * 64), + Descriptor("0008_result_checkpoint_audit_events", 456, "b" * 64), + ) + monkeypatch.setattr( + cli, + "apply_checkpoint_schema_migrations", + lambda dsn: calls.append(dsn) or plan, + ) + + assert cli._dispatch( + ["init-checkpoint-storage", "--dsn", "postgresql://secret@database"] + ) == 0 + + assert calls == ["postgresql://secret@database"] + report = json.loads(capsys.readouterr().out) + assert report == { + "schema_version": 1, + "applied_migrations": [item.as_dict() for item in plan], + } + serialized = json.dumps(report) + assert "postgresql://" not in serialized + assert "secret" not in serialized + assert "SELECT" not in serialized + + def test_count_health_and_server_commands(monkeypatch, capsys): """Synchronous operational commands emit machine-readable results.""" class Counter: diff --git a/tests/test_checkpoint_audit_ci_contract.py b/tests/test_checkpoint_audit_ci_contract.py index 322bc5de9..c6a0b06af 100644 --- a/tests/test_checkpoint_audit_ci_contract.py +++ b/tests/test_checkpoint_audit_ci_contract.py @@ -1,5 +1,5 @@ # SPDX-License-Identifier: Apache-2.0 -"""CI contract for the live checkpoint-audit PostgreSQL verification gate.""" +"""CI contract for live checkpoint-storage PostgreSQL verification.""" from __future__ import annotations @@ -79,8 +79,8 @@ def _single_step_with_scalar( return matches[0] -def test_ci_runs_checkpoint_audit_against_ephemeral_postgres() -> None: - """The exact audit job must bind every required service and step setting.""" +def test_ci_runs_checkpoint_storage_against_ephemeral_postgres() -> None: + """The exact storage job binds its service, credentials, and live command.""" workflow = ( Path(__file__).resolve().parents[1] / ".github/workflows/ci.yml" ).read_text(encoding="utf-8") @@ -108,22 +108,22 @@ def test_ci_runs_checkpoint_audit_against_ephemeral_postgres() -> None: integration = _single_step_with_scalar( steps, "name", - "Run live checkpoint audit integration", + "Run live checkpoint storage integration", ) - assert ( - _scalar_value(integration, "run") - == "uv run pytest -q tests/test_checkpoint_audit_integration.py -m integration" + assert _scalar_value(integration, "run") == ( + "uv run pytest -q tests/test_checkpoint_audit_integration.py " + "tests/test_checkpoint_migration_operator_integration.py -m integration" ) def test_ci_scope_parser_excludes_settings_from_other_jobs() -> None: - """A decoy outside the audit job cannot satisfy its bounded contract.""" + """A decoy outside the storage job cannot satisfy its bounded contract.""" workflow = """jobs: decoy: env: PG_LLM_BATCH_TEST_DSN: decoy-only-dsn steps: - - name: Run live checkpoint audit integration + - name: Run live checkpoint storage integration run: decoy-only-command checkpoint-audit-integration: services: @@ -132,7 +132,7 @@ def test_ci_scope_parser_excludes_settings_from_other_jobs() -> None: env: PG_LLM_BATCH_TEST_DSN: target-only-dsn steps: - - name: Run live checkpoint audit integration + - name: Run live checkpoint storage integration run: target-only-command """ job = _single_mapping_block(workflow, "checkpoint-audit-integration", 2) diff --git a/tests/test_checkpoint_migration_bounded_reads.py b/tests/test_checkpoint_migration_bounded_reads.py new file mode 100644 index 000000000..e0c103b27 --- /dev/null +++ b/tests/test_checkpoint_migration_bounded_reads.py @@ -0,0 +1,73 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Security regression for bounded checkpoint migration file reads.""" + +from __future__ import annotations + +from pg_llm_batch import checkpoint_migrations + + +class _BoundedMigrationFile: + """Record the maximum byte request used by one package migration read.""" + + def __init__(self, payload: bytes, read_sizes: list[int]) -> None: + self.payload = payload + self.read_sizes = read_sizes + + def __enter__(self) -> _BoundedMigrationFile: + """Return this deterministic file double.""" + return self + + def __exit__(self, *_exc: object) -> None: + """Leave cleanup to the in-memory test double.""" + return None + + def read(self, size: int = -1) -> bytes: + """Reject unbounded reads and return the configured SQL bytes.""" + self.read_sizes.append(size) + if size < 0: + raise AssertionError("migration SQL read must be bounded") + return self.payload[:size] + + +class _BoundedMigrationPath: + """Expose only an explicitly sized binary read for one fake path.""" + + def __init__(self, payload: bytes, read_sizes: list[int]) -> None: + self.payload = payload + self.read_sizes = read_sizes + + def open(self, mode: str) -> _BoundedMigrationFile: + """Require binary read mode and return the bounded file double.""" + assert mode == "rb" + return _BoundedMigrationFile(self.payload, self.read_sizes) + + def read_bytes(self) -> bytes: + """Fail if production falls back to Path.read_bytes().""" + raise AssertionError("Path.read_bytes() is an unbounded migration read") + + +def test_plan_reads_at_most_one_byte_beyond_the_migration_limit(monkeypatch) -> None: + """Canonical SQL planning never materializes an arbitrarily large file.""" + read_sizes: list[int] = [] + monkeypatch.setattr( + checkpoint_migrations, + "_CHECKPOINT_SCHEMA_MIGRATION_PATHS", + ( + ( + "0007_result_stream_checkpoints", + _BoundedMigrationPath(b"SELECT 1;", read_sizes), + ), + ( + "0008_result_checkpoint_audit_events", + _BoundedMigrationPath(b"SELECT 2;", read_sizes), + ), + ), + ) + + plan = checkpoint_migrations.plan_checkpoint_schema_migrations() + + assert len(plan) == 2 + assert read_sizes == [ + checkpoint_migrations.MAX_CHECKPOINT_SCHEMA_MIGRATION_BYTES + 1, + checkpoint_migrations.MAX_CHECKPOINT_SCHEMA_MIGRATION_BYTES + 1, + ] diff --git a/tests/test_checkpoint_migration_operator.py b/tests/test_checkpoint_migration_operator.py new file mode 100644 index 000000000..c18ad8436 --- /dev/null +++ b/tests/test_checkpoint_migration_operator.py @@ -0,0 +1,295 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Contracts for the atomic checkpoint-storage migration operator.""" + +from __future__ import annotations + +from hashlib import sha256 +from pathlib import Path + +import pytest + +from pg_llm_batch import ( + CheckpointSchemaMigration, + apply_checkpoint_schema_migrations, + plan_checkpoint_schema_migrations, +) +from pg_llm_batch import checkpoint_migrations +from pg_llm_batch.checkpoint_audit import AUDIT_MIGRATION_PATH +from pg_llm_batch.checkpoint_store import MIGRATION_PATH + + +class _RecordingCursor: + """Record SQL calls and optionally fail on one exact migration body.""" + + def __init__(self, fail_sql: str | None = None) -> None: + self.fail_sql = fail_sql + self.executions: list[tuple[str, object | None]] = [] + + def __enter__(self) -> _RecordingCursor: + """Return this cursor for context-managed production use.""" + return self + + def __exit__(self, *_exc: object) -> None: + """Leave cursor cleanup to the deterministic fake.""" + return None + + def execute(self, statement: str, parameters: object | None = None) -> None: + """Record one SQL call and raise for the configured migration body.""" + self.executions.append((statement, parameters)) + if statement == self.fail_sql: + raise RuntimeError("database migration failed") + + +class _RecordingConnection: + """Record transaction ownership without implementing PostgreSQL behavior.""" + + def __init__(self, cursor: _RecordingCursor) -> None: + self.recording_cursor = cursor + self.commit_calls = 0 + self.exit_exception_type: type[BaseException] | None = None + + def __enter__(self) -> _RecordingConnection: + """Return this connection for context-managed production use.""" + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + _exc: BaseException | None, + _traceback: object | None, + ) -> None: + """Remember whether the transaction left through an exception.""" + self.exit_exception_type = exc_type + return None + + def cursor(self) -> _RecordingCursor: + """Return the one deterministic recording cursor.""" + return self.recording_cursor + + def commit(self) -> None: + """Count one explicit production commit.""" + self.commit_calls += 1 + + +class _RecordingPsycopg: + """Expose one deterministic connection through the psycopg seam.""" + + def __init__(self, connection: _RecordingConnection) -> None: + self.connection = connection + self.connect_calls: list[str] = [] + + def connect(self, postgres_dsn: str) -> _RecordingConnection: + """Record the DSN only inside the test fake and return its connection.""" + self.connect_calls.append(postgres_dsn) + return self.connection + + +def _write_migration(path: Path, sql: bytes) -> tuple[str, Path]: + """Write one private test migration and return its reviewed tuple entry.""" + path.write_bytes(sql) + return (path.stem, path) + + +def test_plan_uses_canonical_order_sizes_and_digests() -> None: + """Planning identifies the exact package SQL bytes in dependency order.""" + plan = plan_checkpoint_schema_migrations() + assert tuple(item.migration_id for item in plan) == ( + "0007_result_stream_checkpoints", + "0008_result_checkpoint_audit_events", + ) + assert all(isinstance(item, CheckpointSchemaMigration) for item in plan) + assert tuple(item.byte_count for item in plan) == ( + len(MIGRATION_PATH.read_bytes()), + len(AUDIT_MIGRATION_PATH.read_bytes()), + ) + assert tuple(item.sha256 for item in plan) == ( + sha256(MIGRATION_PATH.read_bytes()).hexdigest(), + sha256(AUDIT_MIGRATION_PATH.read_bytes()).hexdigest(), + ) + + +def test_migration_descriptor_is_immutable_and_json_safe() -> None: + """Public evidence is immutable and contains only bounded scalar fields.""" + descriptor = CheckpointSchemaMigration( + migration_id="0007_result_stream_checkpoints", + byte_count=42, + sha256="a" * 64, + ) + assert descriptor.as_dict() == { + "migration_id": "0007_result_stream_checkpoints", + "byte_count": 42, + "sha256": "a" * 64, + } + with pytest.raises(Exception): + descriptor.byte_count = 43 # type: ignore[misc] + + +@pytest.mark.parametrize( + ("migration_id", "byte_count", "digest"), + ( + ("unexpected_migration", 1, "a" * 64), + (7, 1, "a" * 64), + ("0007_result_stream_checkpoints", True, "a" * 64), + ("0007_result_stream_checkpoints", 0, "a" * 64), + ( + "0007_result_stream_checkpoints", + checkpoint_migrations.MAX_CHECKPOINT_SCHEMA_MIGRATION_BYTES + 1, + "a" * 64, + ), + ("0007_result_stream_checkpoints", 1, None), + ("0007_result_stream_checkpoints", 1, "A" * 64), + ("0007_result_stream_checkpoints", 1, "g" * 64), + ("0007_result_stream_checkpoints", 1, "a" * 63), + ), +) +def test_migration_descriptor_rejects_unbounded_or_ambiguous_values( + migration_id: object, + byte_count: object, + digest: object, +) -> None: + """Descriptor construction is strict, non-coercive, and closed to new IDs.""" + with pytest.raises(ValueError): + CheckpointSchemaMigration( + migration_id=migration_id, # type: ignore[arg-type] + byte_count=byte_count, # type: ignore[arg-type] + sha256=digest, # type: ignore[arg-type] + ) + + +@pytest.mark.parametrize( + "payload", + ( + b"", + b"x" * (1_048_576 + 1), + b"\xff", + ), +) +def test_plan_rejects_empty_oversized_or_non_utf8_migrations( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + payload: bytes, +) -> None: + """Every migration is bounded and decoded before any database operation.""" + first = _write_migration(tmp_path / "first.sql", b"SELECT 1;") + second = _write_migration(tmp_path / "second.sql", payload) + monkeypatch.setattr( + checkpoint_migrations, + "_CHECKPOINT_SCHEMA_MIGRATION_PATHS", + (first, second), + ) + with pytest.raises((RuntimeError, UnicodeDecodeError)): + plan_checkpoint_schema_migrations() + + +def test_apply_loads_every_file_before_psycopg_or_database_access( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """A missing second migration cannot leave the first migration committed.""" + first = _write_migration(tmp_path / "first.sql", b"SELECT 1;") + missing = ("second", tmp_path / "missing.sql") + monkeypatch.setattr( + checkpoint_migrations, + "_CHECKPOINT_SCHEMA_MIGRATION_PATHS", + (first, missing), + ) + require_calls: list[str] = [] + connect_calls: list[str] = [] + monkeypatch.setattr( + checkpoint_migrations, + "_require_psycopg", + lambda: require_calls.append("required"), + ) + monkeypatch.setattr( + checkpoint_migrations, + "psycopg", + type( + "ForbiddenPsycopg", + (), + {"connect": staticmethod(lambda dsn: connect_calls.append(dsn))}, + )(), + ) + + with pytest.raises(OSError): + apply_checkpoint_schema_migrations("postgresql://secret@database") + + assert require_calls == [] + assert connect_calls == [] + + +def test_apply_uses_one_lock_exact_order_and_one_commit( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """One owned transaction serializes and commits both migrations together.""" + first_sql = "SELECT 'first';" + second_sql = "SELECT 'second';" + monkeypatch.setattr( + checkpoint_migrations, + "_CHECKPOINT_SCHEMA_MIGRATION_PATHS", + ( + _write_migration(tmp_path / "first.sql", first_sql.encode()), + _write_migration(tmp_path / "second.sql", second_sql.encode()), + ), + ) + cursor = _RecordingCursor() + connection = _RecordingConnection(cursor) + provider = _RecordingPsycopg(connection) + require_calls: list[str] = [] + monkeypatch.setattr(checkpoint_migrations, "psycopg", provider) + monkeypatch.setattr( + checkpoint_migrations, + "_require_psycopg", + lambda: require_calls.append("required"), + ) + + applied = apply_checkpoint_schema_migrations("postgresql://operator") + + assert require_calls == ["required"] + assert provider.connect_calls == ["postgresql://operator"] + assert cursor.executions == [ + ( + "SELECT pg_advisory_xact_lock(%s, %s)", + ( + checkpoint_migrations.CHECKPOINT_SCHEMA_MIGRATION_LOCK_NAMESPACE, + checkpoint_migrations.CHECKPOINT_SCHEMA_MIGRATION_LOCK_OPERATION, + ), + ), + (first_sql, None), + (second_sql, None), + ] + assert connection.commit_calls == 1 + assert connection.exit_exception_type is None + assert tuple(item.migration_id for item in applied) == ("first", "second") + + +def test_apply_propagates_second_failure_without_commit_or_success( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """A second-statement failure exits the owned transaction before commit.""" + first_sql = "SELECT 'first';" + second_sql = "SELECT 'second';" + monkeypatch.setattr( + checkpoint_migrations, + "_CHECKPOINT_SCHEMA_MIGRATION_PATHS", + ( + _write_migration(tmp_path / "first.sql", first_sql.encode()), + _write_migration(tmp_path / "second.sql", second_sql.encode()), + ), + ) + cursor = _RecordingCursor(fail_sql=second_sql) + connection = _RecordingConnection(cursor) + monkeypatch.setattr( + checkpoint_migrations, + "psycopg", + _RecordingPsycopg(connection), + ) + monkeypatch.setattr(checkpoint_migrations, "_require_psycopg", lambda: None) + + with pytest.raises(RuntimeError, match="database migration failed"): + apply_checkpoint_schema_migrations("postgresql://operator") + + assert cursor.executions[-2:] == [(first_sql, None), (second_sql, None)] + assert connection.commit_calls == 0 + assert connection.exit_exception_type is RuntimeError diff --git a/tests/test_checkpoint_migration_operator_documentation.py b/tests/test_checkpoint_migration_operator_documentation.py new file mode 100644 index 000000000..ab6c979bc --- /dev/null +++ b/tests/test_checkpoint_migration_operator_documentation.py @@ -0,0 +1,94 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Documentation contracts for the atomic checkpoint migration operator.""" + +from __future__ import annotations + +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] + + +def _read(path: str) -> str: + """Read one authoritative repository document as UTF-8 text.""" + return (ROOT / path).read_text(encoding="utf-8") + + +def test_public_operator_documentation_is_complete_and_body_free() -> None: + """README and operator guide explain the safe opt-in workflow to beginners.""" + combined = "\n".join( + ( + _read("README.md"), + _read("docs/checkpoint-storage-migrations.md"), + ) + ) + required = ( + "init-checkpoint-storage", + "0007_result_stream_checkpoints", + "0008_result_checkpoint_audit_events", + "before database access", + "pg_advisory_xact_lock", + "one transaction", + "one commit", + "rolls back", + "1 MiB", + "not a signature", + "init-db", + "existing PostgreSQL volumes", + ) + for phrase in required: + assert phrase in combined + forbidden = ( + "prints the DSN", + "prints SQL text", + "migration ledger table", + "downgrade retained evidence", + ) + for phrase in forbidden: + assert phrase not in combined + + +def test_contributor_invariants_define_atomic_migration_ownership() -> None: + """Agent contracts preserve ordering, locking, compatibility, and evidence.""" + for path in ("AGENTS.md", "CLAUDE.md"): + content = _read(path) + required = ( + "Checkpoint migration operator", + "init-checkpoint-storage", + "0007_result_stream_checkpoints", + "0008_result_checkpoint_audit_events", + "transaction-level advisory lock", + "before database access", + "one commit", + "not a signature", + "100% production statement, branch, and public-docstring coverage", + ) + for phrase in required: + assert phrase in content + + +def test_architecture_adr_changelog_and_doctoring_are_synchronized() -> None: + """Authoritative design and assurance records describe one shared boundary.""" + documents = ( + _read("ARCHITECTURE.md"), + _read("CHANGELOG.md"), + _read("docs/adr/0010-atomic-checkpoint-schema-operator.md"), + _read("docs/doctoring/checkpoint-migration-operator.md"), + ) + for content in documents: + required = ( + "init-checkpoint-storage", + "0007_result_stream_checkpoints", + "0008_result_checkpoint_audit_events", + "pg_advisory_xact_lock", + "SHA-256", + ) + for phrase in required: + assert phrase in content + + doctoring = documents[-1] + assert "NIST Special Publication 800-53, Revision 5" in doctoring + assert "PostgreSQL Global Development Group. (2026)." in doctoring + assert "CM-3" in doctoring + assert "transaction-level" in doctoring + assert "administrator" in doctoring diff --git a/tests/test_checkpoint_migration_operator_integration.py b/tests/test_checkpoint_migration_operator_integration.py new file mode 100644 index 000000000..573a0c2df --- /dev/null +++ b/tests/test_checkpoint_migration_operator_integration.py @@ -0,0 +1,182 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Live PostgreSQL verification for the checkpoint migration operator.""" + +from __future__ import annotations + +import os +import threading +import time +import uuid +from collections.abc import Iterator +from contextlib import contextmanager +from pathlib import Path + +import pytest + +from pg_llm_batch import apply_checkpoint_schema_migrations +from pg_llm_batch import checkpoint_migrations +from pg_llm_batch.checkpoint_store import MIGRATION_PATH + +pytestmark = pytest.mark.integration + +ADMIN_DSN = os.environ.get("PG_LLM_BATCH_TEST_DSN") +skip_no_db = pytest.mark.skipif( + not ADMIN_DSN, + reason="PG_LLM_BATCH_TEST_DSN not set; skipping live migration integration", +) + + +@contextmanager +def _temporary_database() -> Iterator[str]: + """Yield one isolated PostgreSQL database and remove it after the test.""" + import psycopg + from psycopg import sql + from psycopg.conninfo import make_conninfo + + suffix = uuid.uuid4().hex[:12] + database_name = f"migration_operator_{suffix}" + database_created = False + database_dsn = make_conninfo(ADMIN_DSN, dbname=database_name) + try: + with psycopg.connect(ADMIN_DSN, autocommit=True) as cluster_admin: + with cluster_admin.cursor() as cursor: + cursor.execute( + sql.SQL("CREATE DATABASE {}").format( + sql.Identifier(database_name) + ) + ) + database_created = True + with psycopg.connect(database_dsn, autocommit=True) as database_admin: + with database_admin.cursor() as cursor: + cursor.execute('CREATE EXTENSION IF NOT EXISTS "uuid-ossp"') + yield database_dsn + finally: + if database_created: + with psycopg.connect(ADMIN_DSN, autocommit=True) as cluster_admin: + with cluster_admin.cursor() as cursor: + cursor.execute( + "SELECT pg_terminate_backend(pid) FROM pg_stat_activity " + "WHERE datname = %s AND pid <> pg_backend_pid()", + (database_name,), + ) + cursor.execute( + sql.SQL("DROP DATABASE IF EXISTS {}").format( + sql.Identifier(database_name) + ) + ) + + +def _table_names(database_dsn: str) -> tuple[str | None, str | None]: + """Return the durable checkpoint and audit table identities, if present.""" + import psycopg + + with psycopg.connect(database_dsn) as connection: + with connection.cursor() as cursor: + cursor.execute( + "SELECT " + "to_regclass('public.llm_result_stream_checkpoints')::text, " + "to_regclass('public.llm_result_checkpoint_audit_events')::text" + ) + row = cursor.fetchone() + assert row is not None + return row[0], row[1] + + +@skip_no_db +def test_second_migration_failure_rolls_back_the_first( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """Invalid migration 0008 leaves neither package table partially applied.""" + import psycopg + + checkpoint_path = tmp_path / "0007_result_stream_checkpoints.sql" + checkpoint_path.write_bytes(MIGRATION_PATH.read_bytes()) + invalid_audit_path = tmp_path / "0008_result_checkpoint_audit_events.sql" + invalid_audit_path.write_text("SELECT FROM;", encoding="utf-8") + monkeypatch.setattr( + checkpoint_migrations, + "_CHECKPOINT_SCHEMA_MIGRATION_PATHS", + ( + ("0007_result_stream_checkpoints", checkpoint_path), + ("0008_result_checkpoint_audit_events", invalid_audit_path), + ), + ) + + with _temporary_database() as database_dsn: + with pytest.raises(psycopg.Error): + apply_checkpoint_schema_migrations(database_dsn) + assert _table_names(database_dsn) == (None, None) + + +@skip_no_db +def test_concurrent_operator_waits_for_the_transaction_advisory_lock() -> None: + """A competing invocation waits until the held package lock is released.""" + import psycopg + from psycopg.conninfo import make_conninfo + + application_name = f"checkpoint_migration_{uuid.uuid4().hex[:12]}" + worker_errors: list[BaseException] = [] + worker_started = threading.Event() + worker_finished = threading.Event() + + with _temporary_database() as database_dsn: + worker_dsn = make_conninfo(database_dsn, application_name=application_name) + with psycopg.connect(database_dsn) as lock_holder: + with lock_holder.cursor() as cursor: + cursor.execute( + "SELECT pg_advisory_xact_lock(%s, %s)", + ( + checkpoint_migrations.CHECKPOINT_SCHEMA_MIGRATION_LOCK_NAMESPACE, + checkpoint_migrations.CHECKPOINT_SCHEMA_MIGRATION_LOCK_OPERATION, + ), + ) + + def run_worker() -> None: + """Apply the migration plan and retain only test-local failures.""" + worker_started.set() + try: + apply_checkpoint_schema_migrations(worker_dsn) + except BaseException as exc: # pragma: no cover - asserted below + worker_errors.append(exc) + finally: + worker_finished.set() + + worker = threading.Thread(target=run_worker, daemon=True) + worker.start() + assert worker_started.wait(timeout=2.0) + + waiting_observed = False + deadline = time.monotonic() + 5.0 + with psycopg.connect(database_dsn, autocommit=True) as observer: + with observer.cursor() as cursor: + while time.monotonic() < deadline: + cursor.execute( + "SELECT EXISTS (" + "SELECT 1 FROM pg_locks AS migration_locks " + "JOIN pg_stat_activity AS migration_sessions " + "ON migration_sessions.pid = migration_locks.pid " + "WHERE migration_sessions.application_name = %s " + "AND migration_locks.locktype = 'advisory' " + "AND NOT migration_locks.granted" + ")", + (application_name,), + ) + row = cursor.fetchone() + if row and row[0] is True: + waiting_observed = True + break + time.sleep(0.02) + + assert waiting_observed + assert not worker_finished.is_set() + lock_holder.commit() + assert worker_finished.wait(timeout=10.0) + worker.join(timeout=1.0) + + assert not worker.is_alive() + assert worker_errors == [] + assert _table_names(database_dsn) == ( + "llm_result_stream_checkpoints", + "llm_result_checkpoint_audit_events", + ) diff --git a/tests/test_workflow_contracts.py b/tests/test_workflow_contracts.py index b130b234a..d2ca3a3a3 100644 --- a/tests/test_workflow_contracts.py +++ b/tests/test_workflow_contracts.py @@ -41,6 +41,18 @@ def test_ci_workflow_enforces_supported_versions_and_quality_gates() -> None: _assert_external_actions_are_pinned(workflow) +def test_ci_checks_out_and_verifies_the_exact_source_head_in_every_job() -> None: + """Required CI evidence must execute the exact PR head rather than a merge ref.""" + workflow = _read(".github/workflows/ci.yml") + exact_source_expression = "${{ github.event.pull_request.head.sha || github.sha }}" + + assert workflow.count(f"ref: {exact_source_expression}") == 4 + assert workflow.count("name: Verify exact source head") == 4 + assert workflow.count( + f'test "$(git rev-parse HEAD)" = "{exact_source_expression}"' + ) == 4 + + def test_hourly_workflow_repairs_revalidates_and_merges_pull_requests() -> None: workflow = _read(".github/workflows/hourly-maintenance.yml") scheduler_sha = "5983b41ace75040c1d81818171ca7d0f3653254e"