diff --git a/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/DECISIONS.md b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/DECISIONS.md index 374be4dc..37b57fb6 100644 --- a/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/DECISIONS.md +++ b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/DECISIONS.md @@ -408,9 +408,14 @@ The v0.1 extraction-policy limits are fixed, not caller-selectable: |---|---:|---|---| | parser input per item | 32 MiB | 03B3A framework before dispatch | `limit_exceeded` | | canonical output per item | 4 MiB | 03B3A streaming output collector | `limit_exceeded` | +| subprocess CPU time per item | 30 seconds | 03B3A OS resource limit | `limit_exceeded` | | aggregate agent material | 12 MiB | 03B4 input assembler | `limit_exceeded` | | subprocess wall time per item | 60 seconds | 03B3A process supervisor | `limit_exceeded` | | subprocess address space | 512 MiB | 03B3A OS isolation boundary | `limit_exceeded` | +| subprocess output file | 4 MiB | 03B3A OS file-size limit | `limit_exceeded` | +| subprocess open descriptors | 32 | 03B3A OS descriptor limit | `limit_exceeded` | +| subprocess children/core dumps | 0 | 03B3A OS process/core limits | `parser_failure` | +| JSON container depth | 64 | 03B3A JSON adapter | `limit_exceeded` | | container entries | 2,000 | 03B2 container inspector | `limit_exceeded` | | decompressed container bytes | 128 MiB | 03B2 container inspector | `limit_exceeded` | | container nesting depth | 8 | 03B2 container inspector | `limit_exceeded` | @@ -429,14 +434,30 @@ termination, and executor-loss tests are mandatory. Executor loss leaves no successful extraction usage record; a current-generation retry starts from fresh materialization and authority. +03B3A accepts only UTF-8 text-family input with at most one leading UTF-8 BOM, +normalizes CRLF/CR to LF, and rejects NUL or controls other than tab/LF. +Markdown is bounded text, not rendered markup. JSON rejects duplicate keys and +non-finite numbers and uses sorted-key compact UTF-8 serialization. CSV uses the +fixed strict Python `excel` dialect and serializes exact row arrays as compact +UTF-8 JSON. After trusted imports, the Linux extraction child installs a +default-deny libseccomp filter with an explicit descriptor-only syscall +allowlist. Parsing then uses only pre-opened standard descriptors; unavailable +isolation fails closed. An exact-lineage durable budget permits the initial +materialization plus at most one fresh-authority retry for `parser_failure` or +current-lineage cancellation. Deterministic terminal outcomes replay without +another materialization. Failed attempts may retain +bounded status evidence, but never canonical output payload, successful usage, +or a report. + ## D43 - Canonical Extraction Records, Not Implicit Provider Writes -v0.1 persists an immutable content-derived extraction representation in -PostgreSQL keyed by original content, format, extractor/version, and policy -version, with output digest, omission facts, status, and bounded error code. A -separate immutable usage record binds it to the exact guide binding, source -item, setup run, and generation. AUTH-04B grants read and binding only, so ART -does not use read authority to create a provider object. +v0.1 persists bounded immutable attempt evidence with status/error separately +from the successful content-derived representation. Successful content is +keyed by original content, format, extractor/version, and policy version and +stores canonical output, its digest, and omission facts without an error code. +A separate immutable usage record binds that success to the exact guide +binding, source item, setup run, and generation. AUTH-04B grants read and +binding only, so ART does not use read authority to create a provider object. ## D44 - Sufficiency Consumes Complete Verified Material diff --git a/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/REVIEW_LOG.md b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/REVIEW_LOG.md index 04ff8c85..802ca8d3 100644 --- a/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/REVIEW_LOG.md +++ b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/REVIEW_LOG.md @@ -203,3 +203,15 @@ the passing repository-owned hosted docstring gate. - Senior-engineering, security, and QA repair-delta re-reviews pass with no blockers before the repaired PR head is published. + +## WS-ART-001-03B3A + +- Implemented the hidden default-deny extraction framework and standard-library + text, Markdown, JSON, and CSV canonicalization slice. +- Initial L1 review found blocking sandbox, provenance, concurrency, workspace, + retry, terminal replay, schema-proof, and test-surface issues. +- Repairs added exact guide/content locks, classification predicates, durable + two-slot retry custody, extracted-attempt usage fencing, cleanup recovery, + fixture-only resource probes, and expanded boundary/migration evidence. +- Architecture, security, senior engineering, product/ops, QA, docs, reuse, + CI-integrity, and test-delta repair reviews pass with no blockers. diff --git a/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/chunks/WS-ART-001-03B3A-extraction-framework-text.md b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/chunks/WS-ART-001-03B3A-extraction-framework-text.md index 6a590057..85b1c13f 100644 --- a/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/chunks/WS-ART-001-03B3A-extraction-framework-text.md +++ b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/chunks/WS-ART-001-03B3A-extraction-framework-text.md @@ -26,10 +26,33 @@ and standard-library text, Markdown, JSON, and CSV extraction. - no-network subprocess enforces input/output, CPU/time/memory, row/cell, encoding, nesting, cancellation, and cleanup limits; +- after trusted standard-library imports, the Linux child installs a + default-deny libseccomp filter with an explicit syscall allowlist before + parsing. Network/process creation, new file opens, filesystem mutation, and + kernel-introspection surfaces return `EPERM`. Parsing is + descriptor-only over pre-opened stdin/stdout/stderr; missing filter support + fails closed as `parser_failure`. Launch uses `shell=False`, closed + extraneous file descriptors, one scratch-owned working directory, and a minimal + allowlisted environment containing no provider, proxy, database, OpenAI, or + authorization secrets; - the framework rejects parser input above 32 MiB, canonical output above - 4 MiB, execution beyond 60 seconds, or address-space use above 512 MiB; + 4 MiB, CPU use beyond 30 seconds, wall execution beyond 60 seconds, + address-space use above 512 MiB, output files above 4 MiB, more than 32 open + descriptors, any child process, or any core dump. Numeric resource breaches + record `limit_exceeded`; prohibited process creation, unavailable isolation, + or abnormal child termination records `parser_failure`; CSV rejects more than 100,000 rows, 1,000,000 cells, or 32,768 characters in - one cell; every breach deterministically records `limit_exceeded`; + one cell; every CSV numeric breach records `limit_exceeded`; +- text-family input accepts UTF-8 only, permits at most one leading UTF-8 BOM, + normalizes CRLF and CR to LF, rejects NUL and control characters other than + tab/LF, and otherwise preserves text exactly. Markdown uses the same rules + without rendering or parsing; +- JSON rejects duplicate object keys, non-finite numbers, invalid UTF-8, and + nesting deeper than 64 containers; it canonicalizes with recursively sorted + keys, compact separators, UTF-8 characters unescaped where JSON permits, and + no trailing newline. CSV uses the fixed Python `excel` dialect with strict + quoting, preserves empty/ragged cells, and canonicalizes to compact UTF-8 JSON + containing the exact row arrays; - tests cover each exact boundary and one-over boundary, timeout and memory termination, cancellation, executor loss, scratch cleanup, and retry through fresh materialization and fresh authority with no partial durable output; @@ -38,13 +61,24 @@ and standard-library text, Markdown, JSON, and CSV extraction. and policy; separate usage records name item, binding, run, and generation; - text/Markdown/JSON/CSV canonicalization and error statuses are deterministic; - unsupported raw input never reaches an agent or provider write; +- failed attempts may persist one bounded status/error record, but only an + `extracted` current-generation result may retain canonical output and create a + usage record; failure, cancellation, or executor loss creates no successful + usage, report, or partial output payload; +- transient failure evidence is attempt-scoped and cannot occupy or poison the + deterministic successful content key; a later fresh-authority/materialization + retry may publish the one immutable successful content record; +- a durable exact-lineage budget reserves at most the initial materialization + and one retry. Only `parser_failure` or current-lineage cancellation may use + the second slot; deterministic terminal outcomes replay without another + provider read; - changed subsystem coverage is at least 90%; repository coverage stays 78%. ## Verification ```bash (cd backend && .venv/bin/python -m ruff check app tests scripts) -(cd backend && WORKSTREAM_TEST_DATABASE_URL=postgresql+asyncpg://workstream:workstream@localhost:5433/workstream_test .venv/bin/pytest tests/test_alembic.py tests/test_guide_extraction.py -q --cov=app --cov-report=term-missing --cov-fail-under=0) +(cd backend && WORKSTREAM_TEST_DATABASE_URL=postgresql+asyncpg://workstream:workstream@localhost:5433/workstream_test .venv/bin/pytest tests/test_alembic.py tests/test_guide_bindings.py tests/test_guide_extraction.py -q --cov=app --cov-report=term-missing --cov-fail-under=0) (cd backend && .venv/bin/coverage report --precision=2 --fail-under=78) (cd backend && .venv/bin/coverage report --include='app/modules/artifacts/*,app/core/config.py,app/interfaces/artifacts.py' --precision=2 --fail-under=90) python3 scripts/check_stale_artifact_contracts.py @@ -53,6 +87,14 @@ python3 scripts/test_agent_gates.py git diff --check ``` +Architecture/security tests must also prove that 03B3A introduces no direct +provider read/write, concrete adapter import, public raw-extraction route, +in-process untrusted parser, plugin discovery, production parser dependency, +agent invocation, AUTH availability edit, secret-bearing child environment, or +cross-lineage extraction lookup. A real child probe must fail DNS/socket access. +The same probe must fail reads of a known outside-scratch file and writes both +inside and outside scratch after the descriptor-only filter is installed. + The exact PR head must pass hosted `Backend / test` and `Agent Gates / agent-gates`. Every changed production module, including repository, schema, migration-owned service, and composition surfaces, must be included in a dedicated retained diff --git a/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/reviews/WS-ART-001-03B3A-external-review-response.md b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/reviews/WS-ART-001-03B3A-external-review-response.md new file mode 100644 index 00000000..7c5dba9c --- /dev/null +++ b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/reviews/WS-ART-001-03B3A-external-review-response.md @@ -0,0 +1,99 @@ +# External Review Response: WS-ART-001-03B3A + +## Comments addressed + +- Made cancellation evidence persistence resistant to repeated cancellation and + bounded by a five-second timeout while preserving cancellation as the caller + result. +- Made concurrent extracted-content publication deterministic with PostgreSQL + `ON CONFLICT DO NOTHING`, locked reload, and exact-output comparison. +- Added durable TTL-bearing extraction-workspace custody to the canonical + scratch ledger and stale crash cleanup before ownership release. +- Added a real outside-scratch write denial probe. +- Normalized relative imports in the extraction architecture boundary test. +- Repaired hosted database tests that inserted dependent extraction rows before + their classification was flushed and invoked synchronous Alembic from an + active event loop. +- Rebased onto merged AUTH-12A and moved guide extraction to Alembic revision + `0042_guide_extraction`, preserving the AUTH project-mutation evidence head. +- Made successful replay selection deterministic and replaced duplicated policy + literals with the canonical extraction-policy constant. +- Moved extraction workspace ownership behind a public + `ArtifactScratchManager` context manager, made cleanup recursively bounded and + descriptor-relative, and made every ledger mutation preserve explicit + workspace custody without hidden write-time backfill. +- Added nested parser-residue cleanup coverage and made request fixtures use + explicit lineage fields. +- Scoped successful replay to the current extraction policy on both the attempt + and canonical content, with PostgreSQL regression coverage for obsolete-policy + success evidence. +- Prevented a second materialization slot from being claimed unless a current- + policy `parser_failure` or `cancelled` attempt durably proves retry authority. +- Serialized the remaining retry slot with a PostgreSQL row lock and added a + concurrent-claim regression proving exactly one request receives slot two. +- Made unresolved seccomp syscall names fail as `isolation_unavailable`, made + CSV's parser ceiling subordinate to the policy cell limit, validated every + child error envelope before persistence, and made worker result writes + short-write safe. +- Scoped failed-attempt numbering to the current extraction policy, replaced + the large-text control scan with the equivalent bounded control-range regex, + and repaired the final stale refused-downgrade head assertion. + +## Comments deferred + +- The suggestion to add compat and x32 ABIs to the libseccomp allow-list was not + adopted. The native-only libseccomp filter rejects an architecture mismatch; + adding alternate architectures would expand the accepted syscall surface. + Failure to install or load the native filter remains + `isolation_unavailable`. +- The suggestion to return successful extracted content when workspace cleanup + fails was not adopted. An uncertain cleanup result is a scratch-custody + failure, so the operation remains fail closed while the workspace stays in + the manager-owned pending-cleanup set. +- The suggestion to let a claimed slot with no durable attempt authorize slot + two was not adopted. The reviewed contract permits slot two only after a + durable current-policy `parser_failure` or `cancelled` attempt; process death + before that evidence is an intentionally fail-closed custody outcome. +- The suggestion to replace the pinned descriptor workspace path with a real + pathname was not adopted. Extraction is deliberately process-local and + descriptor-scoped; converting it back to a name-resolved path would weaken + the containment boundary. Cleanup failure also intentionally remains the + visible result rather than publishing output while scratch custody is + uncertain. + +## Human decisions needed + +None. + +## Commands rerun + +- `ruff format` and `ruff check` on every repair file. +- Focused extraction isolation, workspace lifecycle, stale cleanup, and + architecture tests: 6 passed. +- After rebasing onto `64dd9c98`, changed-file Ruff format/check, Alembic + single-head inspection, diff integrity, Markdown links, and stale-wording + checks pass. The earlier 37-test focused run passed before the rebase repair; + the refreshed PostgreSQL and focused proof is delegated to hosted CI because + this worktree's incomplete local test environment resolves a conflicting + global pytest plugin. Hosted checks must be refreshed on the new exact PR + head. +- The final worker/protocol repair slice passed 10 focused synchronous tests; + direct probes also confirmed a 200,000-byte CSV cell is classified as + `limit_exceeded/csv_cell_size_limit` and non-string child output as + `parser_failure/invalid_executor_output`. + +## Remaining risks + +The full PostgreSQL, schema-fingerprint, and repository coverage proof remains +delegated to the hosted Backend gate. No CI threshold or test assertion was +weakened. + +The first reconciled-head Backend run migrated successfully through +`0042_guide_extraction` and observed the expected changed public-schema +fingerprint, then failed closed because the committed custody constant still +described the pre-extraction schema. The constant is updated only from that +hosted isolated-database observation; the next exact head must rerun every lane. +The next hosted run passed 2,280 tests across the other lanes and exposed one +new obsolete-policy replay fixture whose classification dependency had not been +flushed before its attempt insert. The fixture now flushes the classification +first; this is test setup ordering, not a production-path relaxation. diff --git a/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/reviews/WS-ART-001-03B3A-internal-review-evidence.md b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/reviews/WS-ART-001-03B3A-internal-review-evidence.md new file mode 100644 index 00000000..17ffb82a --- /dev/null +++ b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/reviews/WS-ART-001-03B3A-internal-review-evidence.md @@ -0,0 +1,78 @@ +# Internal Review Evidence: WS-ART-001-03B3A + +Reviewed against trusted main: `64dd9c98` + +Reviewed at: `2026-07-29` + +## Candidate + +Hidden, bounded extraction framework for verified guide text, Markdown, JSON, +and CSV with exact provenance, deterministic successful content, and no agent, +provider-write, Celery, submission, or AUTH activation behavior. + +## Deterministic Evidence + +- changed-file Ruff, mapper configuration, Python compilation, and + `git diff --check`: PASS; +- the pre-rebase candidate's focused extraction plus architecture suite passed + 41 tests; refreshed focused and PostgreSQL proof remains delegated to hosted + CI for the reconciled head; +- real default-deny seccomp probes deny network, file opens, filesystem writes, + and process creation after trusted imports; +- real child probes cover CPU, wall, memory, and abnormal executor outcomes; +- schema-contract coverage includes successful extraction/usage, extracted- + attempt fencing, retry outcomes, and populated downgrade refusal; +- stale artifact-contract and Markdown-link scans: PASS; +- exact hosted Backend and Agent Gates remain required on the PR head. + +## Reviewer Results + +| Reviewer | Result | Blocking findings | +|---|---|---| +| senior engineering | PASS WITH LOW RISKS | none | +| architecture | PASS WITH LOW RISKS | none | +| QA/test | PASS WITH LOW RISKS | none | +| security/auth | PASS WITH LOW RISKS | none | +| product/ops | PASS WITH LOW RISKS | none | +| reuse/dedup | PASS WITH LOW RISKS | none | +| CI integrity | PASS WITH LOW RISKS | none | +| test delta | PASS WITH LOW RISKS | none | +| docs | PASS | none | + +The pre-rebase CodeRabbit/hosted-CI repair delta was re-reviewed by senior engineering, +architecture/reuse, QA/test-delta, security, product/ops, and CI/docs. All +tracks pass after one valid legacy-ledger compatibility blocker was repaired +with exact prior-v2 normalization and restart coverage. +The final worker-coverage test delta separately passed security, QA/test-delta, +and CI/docs review; it leaves every real subprocess isolation probe intact. +The AUTH-12A rebase and final external-review repair delta passed refreshed +senior, architecture/reuse, QA/test-delta, security, product/ops, and CI/docs +review after policy replay and concurrent retry-custody findings were repaired. +The subsequent hosted schema-fingerprint and CodeRabbit worker/protocol delta +also passed focused QA, security, senior/reuse, and CI/docs re-review. + +## Material Repairs + +- replaced default-allow syscall denial with a default-deny allowlist; +- serialized guide generation and shared-content publication through canonical + row locks; +- revalidated digest, size, media type, detector, binding, run, and generation; +- fenced successful usage to an exact `extracted` attempt in PostgreSQL; +- added cleanup-tracked scratch workspaces and post-launch child reaping; +- added a durable two-slot exact-lineage materialization budget so only + executor failure or current-lineage cancellation may retry; +- scoped deterministic success replay to the current extraction policy and + denied second-slot claims without durable current-policy retry evidence; +- moved destructive resource probes out of the production worker. + +## Accepted Low Risks + +- A later complex-format chunk should make the registry explicitly table-driven. +- The existing and extraction lineage queries remain similar but not yet + identical enough to justify a shared selector. +- Hosted PostgreSQL concurrency and repository coverage gates remain the final + publication proof. + +Valid findings addressed: yes + +Open sub-agent sessions: none diff --git a/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/reviews/WS-ART-001-03B3A-pr-trust-bundle.md b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/reviews/WS-ART-001-03B3A-pr-trust-bundle.md new file mode 100644 index 00000000..2f7a17d5 --- /dev/null +++ b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/reviews/WS-ART-001-03B3A-pr-trust-bundle.md @@ -0,0 +1,60 @@ +# PR Trust Bundle: WS-ART-001-03B3A + +## Chunk + +`WS-ART-001-03B3A` — Extraction Framework And Text Formats (L1) + +## Goal And Intent + +Extract bounded canonical text from exact verified guide content without +weakening artifact custody or allowing raw binaries, provider access, or caller +authority to cross into the parser boundary. + +## What Changed + +- Added a default-deny, resource-limited, descriptor-only extraction child. +- Added deterministic UTF-8 text/Markdown, strict JSON, and strict CSV + canonicalization. +- Added immutable attempts, successful extracted content, exact usage, and a + durable two-slot retry budget with composite database custody. +- Added extraction-specific scratch workspace custody and cleanup recovery. +- Added exact generation/classification revalidation and focused security, + boundary, retry, migration, and provenance tests. + +## Scope Control + +No PDF/OOXML/image parser, provider write, agent invocation, Celery continuation, +public route, submission behavior, legacy cutover, or AUTH availability change +is included. + +## Tests And CI Integrity + +The pre-rebase focused extraction and architecture suite passed locally, +including real outside-scratch write denial and stale workspace crash recovery. +After rebasing onto merged AUTH-12A, Ruff, Python compilation, single-head +migration inspection, stale-contract scan, Markdown links, and diff integrity +pass. Refreshed focused and PostgreSQL proof remains delegated to hosted CI +because this worktree's incomplete test environment resolves a conflicting +global pytest plugin. CodeRabbit's valid findings and the first hosted database-test +failures were repaired without weakening a workflow, threshold, or assertion. +The isolated worker now also has additive parent-process unit coverage while +its real kernel-isolation subprocess probes remain unchanged. +The migration chain has one head at `0042_guide_extraction`; exact hosted +Backend and Agent Gates remain required on the refreshed PR head before merge. + +## Internal Review + +All required L1 reviewer tracks pass after repairs. Details are in the paired +internal review evidence. + +## Remaining Boundary And Next Gate + +The feature remains hidden. `artifact.guide_source.binding.create` and +`artifact.guide_source.read` remain planned and unavailable. ART-03B3B is the +same-initiative successor and requires a separate explicit start after merge. + +## Human Review Focus + +Confirm default-deny parser isolation, exact lineage locks, two-slot retry +custody, successful-attempt usage fencing, and absence of provider/agent/AUTH +scope expansion. The user retains merge approval for this PR. diff --git a/.agent-loop/merge-intents/WS-ART-001-03B3A.json b/.agent-loop/merge-intents/WS-ART-001-03B3A.json new file mode 100644 index 00000000..6fb0f093 --- /dev/null +++ b/.agent-loop/merge-intents/WS-ART-001-03B3A.json @@ -0,0 +1,9 @@ +{ + "chunk_id": "WS-ART-001-03B3A", + "chunk_title": "Extraction Framework And Text Formats", + "initiative_id": "WS-ART-001", + "next_chunk_id": "WS-ART-001-03B3B", + "next_chunk_title": "Durable Guide Sufficiency Continuation", + "next_requires_explicit_start": true, + "schema_version": 2 +} diff --git a/backend/alembic/versions/0042_guide_extraction.py b/backend/alembic/versions/0042_guide_extraction.py new file mode 100644 index 00000000..8fbc1717 --- /dev/null +++ b/backend/alembic/versions/0042_guide_extraction.py @@ -0,0 +1,277 @@ +"""add bounded guide extraction provenance + +Revision ID: 0042_guide_extraction +Revises: 0041_project_mutation_evidence +Create Date: 2026-07-29 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "0042_guide_extraction" +down_revision = "0041_project_mutation_evidence" +branch_labels = depends_on = None + + +def upgrade() -> None: + op.create_unique_constraint( + "uq_guide_bindings_extraction_attempt_lineage", + "guide_source_artifact_bindings", + ["id", "content_id", "setup_generation"], + ) + op.create_unique_constraint( + "uq_guide_bindings_extraction_lineage", + "guide_source_artifact_bindings", + ["id", "content_id", "source_item_id", "project_setup_run_id", "setup_generation"], + ) + op.create_unique_constraint( + "uq_guide_classifications_extraction_lineage", + "guide_source_format_classifications", + ["id", "binding_id", "content_id", "setup_generation"], + ) + op.create_table( + "guide_source_extraction_attempts", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("binding_id", sa.String(36), nullable=False), + sa.Column("content_id", sa.String(36), nullable=False), + sa.Column("classification_id", sa.String(36), nullable=False), + sa.Column("setup_generation", sa.BigInteger(), nullable=False), + sa.Column("detected_format", sa.String(40), nullable=False), + sa.Column("extractor_name", sa.String(100), nullable=False), + sa.Column("extractor_version", sa.String(40), nullable=False), + sa.Column("policy_version", sa.String(80), nullable=False), + sa.Column("attempt_number", sa.BigInteger(), nullable=False), + sa.Column("status", sa.String(40), nullable=False), + sa.Column("error_code", sa.String(80), nullable=True), + sa.Column("bounded_facts", sa.JSON(), nullable=False), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.ForeignKeyConstraint( + ["binding_id", "content_id", "setup_generation"], + [ + "guide_source_artifact_bindings.id", + "guide_source_artifact_bindings.content_id", + "guide_source_artifact_bindings.setup_generation", + ], + name="fk_guide_extraction_attempts_exact_binding", + ), + sa.ForeignKeyConstraint( + ["classification_id", "binding_id", "content_id", "setup_generation"], + [ + "guide_source_format_classifications.id", + "guide_source_format_classifications.binding_id", + "guide_source_format_classifications.content_id", + "guide_source_format_classifications.setup_generation", + ], + name="fk_guide_extraction_attempts_exact_classification", + ), + sa.UniqueConstraint( + "binding_id", "policy_version", "attempt_number", name="uq_guide_extraction_attempts" + ), + sa.UniqueConstraint( + "id", + "binding_id", + "content_id", + "setup_generation", + "status", + name="uq_guide_extraction_attempts_exact_usage", + ), + sa.CheckConstraint("attempt_number > 0", name="ck_guide_extraction_attempts_number"), + sa.CheckConstraint( + "(status = 'extracted') = (error_code is null)", + name="ck_guide_extraction_attempts_error", + ), + sa.CheckConstraint( + "status in ('extracted','unsupported','ambiguous','malformed','limit_exceeded','parser_failure','cancelled','artifact_incident')", + name="ck_guide_extraction_attempts_status", + ), + ) + op.create_table( + "guide_source_extracted_contents", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column( + "content_id", + sa.String(36), + sa.ForeignKey("artifact_contents.id", ondelete="RESTRICT"), + nullable=False, + ), + sa.Column("detected_format", sa.String(40), nullable=False), + sa.Column("extractor_name", sa.String(100), nullable=False), + sa.Column("extractor_version", sa.String(40), nullable=False), + sa.Column("policy_version", sa.String(80), nullable=False), + sa.Column("source_sha256", sa.String(71), nullable=False), + sa.Column("source_byte_count", sa.BigInteger(), nullable=False), + sa.Column("status", sa.String(40), nullable=False), + sa.Column("output_sha256", sa.String(71), nullable=False), + sa.Column("canonical_output", sa.Text(), nullable=False), + sa.Column("omission_facts", sa.JSON(), nullable=False), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.UniqueConstraint( + "content_id", + "detected_format", + "extractor_name", + "extractor_version", + "policy_version", + name="uq_guide_extracted_contents_identity", + ), + sa.UniqueConstraint("id", "content_id", name="uq_guide_extracted_contents_exact_usage"), + sa.CheckConstraint("status = 'extracted'", name="ck_guide_extracted_contents_status"), + sa.CheckConstraint( + "source_sha256 ~ '^sha256:[0-9a-f]{64}$'", + name="ck_guide_extracted_contents_source_sha256", + ), + sa.CheckConstraint( + "output_sha256 ~ '^sha256:[0-9a-f]{64}$'", + name="ck_guide_extracted_contents_output_sha256", + ), + sa.CheckConstraint( + "source_byte_count >= 0", name="ck_guide_extracted_contents_source_size" + ), + sa.CheckConstraint( + "octet_length(canonical_output) <= 4194304", + name="ck_guide_extracted_contents_output_size", + ), + ) + op.create_table( + "guide_source_extraction_retry_budgets", + sa.Column("binding_id", sa.String(36), primary_key=True), + sa.Column("content_id", sa.String(36), nullable=False), + sa.Column("classification_id", sa.String(36), nullable=False), + sa.Column("setup_generation", sa.BigInteger(), nullable=False), + sa.Column("policy_version", sa.String(80), nullable=False), + sa.Column("claimed_slots", sa.Integer(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.ForeignKeyConstraint( + ["binding_id", "content_id", "setup_generation"], + ["guide_source_artifact_bindings.id", "guide_source_artifact_bindings.content_id", "guide_source_artifact_bindings.setup_generation"], + name="fk_guide_extraction_retry_budgets_exact_binding", + ), + sa.ForeignKeyConstraint( + ["classification_id", "binding_id", "content_id", "setup_generation"], + ["guide_source_format_classifications.id", "guide_source_format_classifications.binding_id", "guide_source_format_classifications.content_id", "guide_source_format_classifications.setup_generation"], + name="fk_guide_extraction_retry_budgets_exact_classification", + ), + sa.CheckConstraint("claimed_slots between 1 and 2", name="ck_guide_extraction_retry_budgets_slots"), + ) + op.create_table( + "guide_source_extraction_usages", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("extracted_content_id", sa.String(36), nullable=False), + sa.Column("extraction_attempt_id", sa.String(36), nullable=False), + sa.Column("attempt_status", sa.String(40), nullable=False), + sa.Column("binding_id", sa.String(36), nullable=False), + sa.Column("content_id", sa.String(36), nullable=False), + sa.Column("source_item_id", sa.String(36), nullable=False), + sa.Column("project_setup_run_id", sa.String(36), nullable=False), + sa.Column("setup_generation", sa.BigInteger(), nullable=False), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.ForeignKeyConstraint( + [ + "binding_id", + "content_id", + "source_item_id", + "project_setup_run_id", + "setup_generation", + ], + [ + "guide_source_artifact_bindings.id", + "guide_source_artifact_bindings.content_id", + "guide_source_artifact_bindings.source_item_id", + "guide_source_artifact_bindings.project_setup_run_id", + "guide_source_artifact_bindings.setup_generation", + ], + name="fk_guide_extraction_usages_exact_binding", + ), + sa.ForeignKeyConstraint( + [ + "extraction_attempt_id", + "binding_id", + "content_id", + "setup_generation", + "attempt_status", + ], + [ + "guide_source_extraction_attempts.id", + "guide_source_extraction_attempts.binding_id", + "guide_source_extraction_attempts.content_id", + "guide_source_extraction_attempts.setup_generation", + "guide_source_extraction_attempts.status", + ], + name="fk_guide_extraction_usages_exact_attempt", + ), + sa.ForeignKeyConstraint( + ["extracted_content_id", "content_id"], + ["guide_source_extracted_contents.id", "guide_source_extracted_contents.content_id"], + name="fk_guide_extraction_usages_exact_content", + ), + sa.UniqueConstraint( + "binding_id", "extracted_content_id", name="uq_guide_extraction_usages" + ), + sa.CheckConstraint( + "attempt_status = 'extracted'", + name="ck_guide_extraction_usages_successful_attempt", + ), + ) + for table, columns in { + "guide_source_extraction_attempts": ("binding_id", "content_id"), + "guide_source_extracted_contents": ("content_id",), + "guide_source_extraction_usages": ( + "extracted_content_id", + "binding_id", + "content_id", + "source_item_id", + "project_setup_run_id", + ), + }.items(): + for column in columns: + op.create_index(op.f(f"ix_{table}_{column}"), table, [column]) + + +def downgrade() -> None: + bind = op.get_bind() + bind.execute( + sa.text( + "lock table guide_source_extraction_usages, guide_source_extracted_contents, guide_source_extraction_retry_budgets, guide_source_extraction_attempts in access exclusive mode" + ) + ) + if bind.execute( + sa.text( + "select exists(select 1 from guide_source_extraction_usages) or exists(select 1 from guide_source_extracted_contents) or exists(select 1 from guide_source_extraction_attempts) or exists(select 1 from guide_source_extraction_retry_budgets)" + ) + ).scalar_one(): + raise RuntimeError("cannot downgrade populated guide extraction evidence") + op.drop_table("guide_source_extraction_usages") + op.drop_table("guide_source_extraction_retry_budgets") + op.drop_table("guide_source_extracted_contents") + op.drop_table("guide_source_extraction_attempts") + op.drop_constraint( + "uq_guide_classifications_extraction_lineage", + "guide_source_format_classifications", + type_="unique", + ) + op.drop_constraint( + "uq_guide_bindings_extraction_lineage", "guide_source_artifact_bindings", type_="unique" + ) + op.drop_constraint( + "uq_guide_bindings_extraction_attempt_lineage", + "guide_source_artifact_bindings", + type_="unique", + ) diff --git a/backend/app/db/models.py b/backend/app/db/models.py index 7c002422..461e7ea9 100644 --- a/backend/app/db/models.py +++ b/backend/app/db/models.py @@ -12,6 +12,10 @@ ArtifactContent, GuideSourceArtifactBinding, GuideSourceArtifactIncident, + GuideSourceExtractedContent, + GuideSourceExtractionAttempt, + GuideSourceExtractionRetryBudget, + GuideSourceExtractionUsage, GuideSourceFormatClassification, ArtifactOperationReceipt, ArtifactReplica, diff --git a/backend/app/modules/artifacts/guide_extraction.py b/backend/app/modules/artifacts/guide_extraction.py new file mode 100644 index 00000000..fbc45693 --- /dev/null +++ b/backend/app/modules/artifacts/guide_extraction.py @@ -0,0 +1,163 @@ +"""Bounded subprocess framework for canonical guide extraction.""" + +from __future__ import annotations + +from dataclasses import dataclass +import hashlib +import json +import os +from pathlib import Path +import signal +import subprocess +import sys +from typing import BinaryIO + + +EXTRACTION_POLICY_VERSION = "guide-extraction-v1" +EXTRACTOR_VERSION = "1" +MAXIMUM_INPUT_BYTES = 32 * 1024 * 1024 +MAXIMUM_OUTPUT_BYTES = 4 * 1024 * 1024 +MAXIMUM_PROTOCOL_BYTES = (MAXIMUM_OUTPUT_BYTES * 6) + 1024 +WALL_TIMEOUT_SECONDS = 60 +_SUPPORTED = frozenset({"plain_text", "markdown", "json", "csv"}) + + +@dataclass(frozen=True, slots=True) +class GuideExtractionResult: + """One bounded child result without raw diagnostics.""" + + status: str + error_code: str | None + canonical_output: str | None + output_sha256: str | None + extractor_name: str + extractor_version: str = EXTRACTOR_VERSION + policy_version: str = EXTRACTION_POLICY_VERSION + + +@dataclass(frozen=True, slots=True) +class BoundGuideExtractor: + """Typed prepared-artifact inspector for one classified format.""" + + runner: GuideExtractionRunner + detected_format: str + + def inspect(self, reader: BinaryIO, workspace: Path) -> GuideExtractionResult: + return self.runner.extract( + reader, detected_format=self.detected_format, workspace=workspace + ) + + +class GuideExtractionRegistry: + """Fixed typed registry for the v0.1 standard-library extractors.""" + + def __init__(self, runner: GuideExtractionRunner | None = None) -> None: + self._runner = runner or GuideExtractionRunner() + + def resolve(self, detected_format: str) -> BoundGuideExtractor: + """Return one policy-bound extractor; unsupported input stays bounded.""" + return BoundGuideExtractor(runner=self._runner, detected_format=detected_format) + + +class GuideExtractionRunner: + """Supervise one fixed, secret-free, descriptor-only extraction child.""" + + def __init__(self) -> None: + self._worker_path = Path(__file__).with_name("guide_extraction_worker.py") + + def extract( + self, + reader: BinaryIO, + *, + detected_format: str, + workspace: Path, + ) -> GuideExtractionResult: + if detected_format not in _SUPPORTED: + return self._result(detected_format, "unsupported", "unsupported_format", None) + payload = reader.read(MAXIMUM_INPUT_BYTES + 1) + if len(payload) > MAXIMUM_INPUT_BYTES: + return self._result(detected_format, "limit_exceeded", "input_limit", None) + environment = {"LANG": "C.UTF-8", "LC_ALL": "C.UTF-8", "PATH": "/usr/bin:/bin"} + process = None + try: + process = subprocess.Popen( + [sys.executable, "-I", str(self._worker_path), detected_format], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + env=environment, + cwd=workspace, + shell=False, + close_fds=True, + start_new_session=True, + ) + stdout, _ = process.communicate(payload, timeout=WALL_TIMEOUT_SECONDS) + except subprocess.TimeoutExpired: + assert process is not None + self._terminate(process) + return self._result(detected_format, "limit_exceeded", "wall_time_limit", None) + except (OSError, ValueError): + if process is not None: + self._terminate(process) + return self._result(detected_format, "parser_failure", "executor_unavailable", None) + if process.returncode == -signal.SIGXCPU: + return self._result(detected_format, "limit_exceeded", "cpu_time_limit", None) + if process.returncode != 0 or len(stdout) > MAXIMUM_PROTOCOL_BYTES: + return self._result(detected_format, "parser_failure", "executor_lost", None) + try: + result = json.loads(stdout) + status = result["status"] + error_code = result["error_code"] + output = result["output"] + except (KeyError, TypeError, ValueError, UnicodeDecodeError): + return self._result(detected_format, "parser_failure", "invalid_executor_output", None) + if status not in { + "extracted", + "unsupported", + "malformed", + "limit_exceeded", + "parser_failure", + }: + return self._result(detected_format, "parser_failure", "invalid_executor_output", None) + if output is not None and not isinstance(output, str): + return self._result(detected_format, "parser_failure", "invalid_executor_output", None) + if isinstance(output, str) and len(output.encode("utf-8")) > MAXIMUM_OUTPUT_BYTES: + return self._result(detected_format, "limit_exceeded", "output_limit", None) + if (status == "extracted" and (error_code is not None or not isinstance(output, str))) or ( + status != "extracted" + and ( + not isinstance(error_code, str) + or len(error_code.encode("utf-8")) > 80 + or output is not None + ) + ): + return self._result(detected_format, "parser_failure", "invalid_executor_output", None) + return self._result(detected_format, status, error_code, output) + + @staticmethod + def _terminate(process: subprocess.Popen[bytes]) -> None: + """Reap a started executor on every uncertain supervision path.""" + if process.poll() is None: + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + process.wait() + + @staticmethod + def _result( + detected_format: str, + status: str, + error_code: str | None, + output: str | None, + ) -> GuideExtractionResult: + if status != "extracted": + output = None + digest = None if output is None else "sha256:" + hashlib.sha256(output.encode()).hexdigest() + return GuideExtractionResult( + status=status, + error_code=error_code, + canonical_output=output, + output_sha256=digest, + extractor_name=f"workstream.{detected_format}", + ) diff --git a/backend/app/modules/artifacts/guide_extraction_service.py b/backend/app/modules/artifacts/guide_extraction_service.py new file mode 100644 index 00000000..ab5764bb --- /dev/null +++ b/backend/app/modules/artifacts/guide_extraction_service.py @@ -0,0 +1,523 @@ +"""Exact-lineage persistence for bounded guide extraction.""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass +from typing import Protocol +from uuid import UUID, uuid4 + +from sqlalchemy import func, select +from sqlalchemy.dialects.postgresql import insert +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from app.core.cancellation import await_cancellation_resistant + +from app.modules.artifacts.guide_extraction import ( + EXTRACTION_POLICY_VERSION, + GuideExtractionRegistry, +) +from app.modules.artifacts.guide_formats import DETECTOR_NAME, DETECTOR_VERSION +from app.modules.artifacts.models import ( + ArtifactContent, + GuideSourceArtifactBinding, + GuideSourceExtractedContent, + GuideSourceExtractionAttempt, + GuideSourceExtractionRetryBudget, + GuideSourceExtractionUsage, + GuideSourceFormatClassification, +) +from app.modules.artifacts.sources import PreparedArtifact +from app.modules.projects.models import ( + GuideSourceSnapshot, + GuideSourceSnapshotItem, + ProjectGuide, + ProjectSetupRun, +) + + +class GuideExtractionError(RuntimeError): + """Concealed guide extraction persistence failure.""" + + +class FreshAuthorizedGuideMaterializer(Protocol): + """Supply one newly authorized and newly materialized exact guide source.""" + + async def materialize_with_fresh_authority( + self, request: GuideExtractionRequest + ) -> PreparedArtifact: + """Return a fresh prepared artifact after current authority revalidation.""" + + +@dataclass(frozen=True, slots=True) +class GuideExtractionRequest: + """Exact lineage selected for one prepared extraction.""" + + project_id: UUID + guide_id: UUID + source_snapshot_id: UUID + source_item_id: UUID + project_setup_run_id: UUID + setup_generation: int + binding_id: UUID + classification_id: UUID + + +@dataclass(frozen=True, slots=True) +class GuideExtractionPersistenceResult: + """Bounded durable extraction outcome.""" + + attempt_id: UUID + status: str + error_code: str | None + extracted_content_id: UUID | None + usage_id: UUID | None + replayed: bool + + +@dataclass(frozen=True, slots=True) +class _ExtractionFacts: + project_id: str + guide_id: str + snapshot_id: str + item_id: str + setup_run_id: str + setup_generation: int + binding_id: str + classification_id: str + content_id: str + sha256: str + byte_count: int + detected_format: str + classification_status: str + + +class GuideExtractionService: + """Extract prepared verified bytes and publish exact immutable provenance.""" + + def __init__( + self, + session_factory: async_sessionmaker[AsyncSession], + registry: GuideExtractionRegistry, + ) -> None: + self._session_factory = session_factory + self._registry = registry + + async def extract_prepared( + self, + request: GuideExtractionRequest, + prepared: PreparedArtifact, + ) -> GuideExtractionPersistenceResult: + """Revalidate, extract in isolation, revalidate, and persist one outcome.""" + async with self._session_factory() as session, session.begin(): + before = await self._load_facts(session, request) + if before is None or before.classification_status != "classified": + raise GuideExtractionError("guide extraction is unavailable") + if ( + prepared.commitment.sha256 != before.sha256 + or prepared.commitment.byte_count != before.byte_count + ): + raise GuideExtractionError("guide extraction is unavailable") + try: + extracted = await prepared.extract_guide(self._registry.resolve(before.detected_format)) + except asyncio.CancelledError as cancellation: + + async def record_cancellation() -> None: + async with self._session_factory() as session, session.begin(): + after = await self._load_facts(session, request) + if after == before: + await self._persist_failure( + session, + before, + status="cancelled", + error_code="extraction_cancelled", + ) + + recovery = asyncio.create_task(asyncio.wait_for(record_cancellation(), timeout=5.0)) + try: + await await_cancellation_resistant(recovery) + except Exception: + pass + raise cancellation from None + async with self._session_factory() as session, session.begin(): + after = await self._load_facts(session, request) + if after != before: + raise GuideExtractionError("guide extraction is unavailable") + if extracted.policy_version != EXTRACTION_POLICY_VERSION: + raise GuideExtractionError("guide extraction result conflicts") + attempt_number = 1 + int( + await session.scalar( + select( + func.coalesce(func.max(GuideSourceExtractionAttempt.attempt_number), 0) + ).where( + GuideSourceExtractionAttempt.binding_id == before.binding_id, + GuideSourceExtractionAttempt.policy_version == extracted.policy_version, + ) + ) + or 0 + ) + attempt = GuideSourceExtractionAttempt( + id=str(uuid4()), + binding_id=before.binding_id, + content_id=before.content_id, + classification_id=before.classification_id, + setup_generation=before.setup_generation, + detected_format=before.detected_format, + extractor_name=extracted.extractor_name, + extractor_version=extracted.extractor_version, + policy_version=extracted.policy_version, + attempt_number=attempt_number, + status=extracted.status, + error_code=extracted.error_code, + bounded_facts={}, + ) + session.add(attempt) + await session.flush() + if extracted.status != "extracted": + return self._result(attempt) + if extracted.canonical_output is None or extracted.output_sha256 is None: + raise GuideExtractionError("guide extraction result conflicts") + content = await session.scalar( + select(GuideSourceExtractedContent) + .where( + GuideSourceExtractedContent.content_id == before.content_id, + GuideSourceExtractedContent.detected_format == before.detected_format, + GuideSourceExtractedContent.extractor_name == extracted.extractor_name, + GuideSourceExtractedContent.extractor_version == extracted.extractor_version, + GuideSourceExtractedContent.policy_version == extracted.policy_version, + ) + .with_for_update() + ) + replayed = content is not None + if content is None: + content_id = str(uuid4()) + inserted_id = await session.scalar( + insert(GuideSourceExtractedContent) + .values( + id=content_id, + content_id=before.content_id, + detected_format=before.detected_format, + extractor_name=extracted.extractor_name, + extractor_version=extracted.extractor_version, + policy_version=extracted.policy_version, + source_sha256=before.sha256, + source_byte_count=before.byte_count, + status="extracted", + output_sha256=extracted.output_sha256, + canonical_output=extracted.canonical_output, + omission_facts={"truncated": False, "omitted": False}, + ) + .on_conflict_do_nothing(constraint="uq_guide_extracted_contents_identity") + .returning(GuideSourceExtractedContent.id) + ) + replayed = inserted_id is None + content = await session.scalar( + select(GuideSourceExtractedContent) + .where( + GuideSourceExtractedContent.content_id == before.content_id, + GuideSourceExtractedContent.detected_format == before.detected_format, + GuideSourceExtractedContent.extractor_name == extracted.extractor_name, + GuideSourceExtractedContent.extractor_version + == extracted.extractor_version, + GuideSourceExtractedContent.policy_version == extracted.policy_version, + ) + .with_for_update() + ) + if content is None: + raise GuideExtractionError("guide extraction result conflicts") + if ( + content.source_sha256 != before.sha256 + or content.source_byte_count != before.byte_count + or content.output_sha256 != extracted.output_sha256 + or content.canonical_output != extracted.canonical_output + ): + raise GuideExtractionError("guide extraction result conflicts") + usage = await session.scalar( + select(GuideSourceExtractionUsage) + .where( + GuideSourceExtractionUsage.binding_id == before.binding_id, + GuideSourceExtractionUsage.extracted_content_id == content.id, + ) + .with_for_update() + ) + if usage is None: + usage = GuideSourceExtractionUsage( + id=str(uuid4()), + extracted_content_id=content.id, + extraction_attempt_id=attempt.id, + attempt_status="extracted", + binding_id=before.binding_id, + content_id=before.content_id, + source_item_id=before.item_id, + project_setup_run_id=before.setup_run_id, + setup_generation=before.setup_generation, + ) + session.add(usage) + await session.flush() + return self._result(attempt, content=content, usage=usage, replayed=replayed) + + async def claim_materialization_slot( + self, request: GuideExtractionRequest + ) -> GuideExtractionPersistenceResult | None: + """Atomically reserve one of two durable materialization slots.""" + async with self._session_factory() as session, session.begin(): + facts = await self._load_facts(session, request) + if facts is None: + raise GuideExtractionError("guide extraction is unavailable") + successful = ( + await session.execute( + select( + GuideSourceExtractionAttempt, + GuideSourceExtractedContent, + GuideSourceExtractionUsage, + ) + .join( + GuideSourceExtractionUsage, + GuideSourceExtractionUsage.extraction_attempt_id + == GuideSourceExtractionAttempt.id, + ) + .join( + GuideSourceExtractedContent, + GuideSourceExtractedContent.id + == GuideSourceExtractionUsage.extracted_content_id, + ) + .where( + GuideSourceExtractionUsage.binding_id == facts.binding_id, + GuideSourceExtractionAttempt.policy_version == EXTRACTION_POLICY_VERSION, + GuideSourceExtractedContent.policy_version == EXTRACTION_POLICY_VERSION, + ) + .order_by( + GuideSourceExtractionAttempt.attempt_number.asc(), + GuideSourceExtractionUsage.created_at.asc(), + GuideSourceExtractionUsage.id.asc(), + ) + .limit(1) + ) + ).one_or_none() + if successful is not None: + attempt, content, usage = successful + return self._result(attempt, content=content, usage=usage, replayed=True) + latest_attempt = await session.scalar( + select(GuideSourceExtractionAttempt) + .where( + GuideSourceExtractionAttempt.binding_id == facts.binding_id, + GuideSourceExtractionAttempt.content_id == facts.content_id, + GuideSourceExtractionAttempt.classification_id == facts.classification_id, + GuideSourceExtractionAttempt.setup_generation == facts.setup_generation, + GuideSourceExtractionAttempt.policy_version == EXTRACTION_POLICY_VERSION, + ) + .order_by(GuideSourceExtractionAttempt.attempt_number.desc()) + .limit(1) + ) + if latest_attempt is not None and latest_attempt.status not in { + "parser_failure", + "cancelled", + }: + return self._result(latest_attempt) + budget = await session.scalar( + select(GuideSourceExtractionRetryBudget) + .where(GuideSourceExtractionRetryBudget.binding_id == facts.binding_id) + .with_for_update() + ) + if budget is None: + session.add( + GuideSourceExtractionRetryBudget( + binding_id=facts.binding_id, + content_id=facts.content_id, + classification_id=facts.classification_id, + setup_generation=facts.setup_generation, + policy_version=EXTRACTION_POLICY_VERSION, + claimed_slots=1, + ) + ) + await session.flush() + return None + if ( + budget.content_id != facts.content_id + or budget.classification_id != facts.classification_id + or budget.setup_generation != facts.setup_generation + or budget.policy_version != EXTRACTION_POLICY_VERSION + ): + raise GuideExtractionError("guide extraction is unavailable") + if latest_attempt is None: + raise GuideExtractionError("guide extraction is unavailable") + if budget.claimed_slots < 2: + budget.claimed_slots += 1 + await session.flush() + return None + if latest_attempt is not None: + return self._result(latest_attempt) + raise GuideExtractionError("guide extraction retry budget is exhausted") + + async def _load_facts( + self, + session: AsyncSession, + request: GuideExtractionRequest, + ) -> _ExtractionFacts | None: + latest_generation = ( + select(func.max(ProjectSetupRun.setup_generation)) + .where(ProjectSetupRun.guide_id == str(request.guide_id)) + .scalar_subquery() + ) + row = ( + await session.execute( + select(GuideSourceArtifactBinding, GuideSourceFormatClassification, ArtifactContent) + .join( + GuideSourceFormatClassification, + GuideSourceFormatClassification.binding_id == GuideSourceArtifactBinding.id, + ) + .join(ArtifactContent, ArtifactContent.id == GuideSourceArtifactBinding.content_id) + .join(ProjectGuide, ProjectGuide.id == GuideSourceArtifactBinding.guide_id) + .join( + GuideSourceSnapshot, + GuideSourceSnapshot.id == GuideSourceArtifactBinding.source_snapshot_id, + ) + .join( + GuideSourceSnapshotItem, + GuideSourceSnapshotItem.id == GuideSourceArtifactBinding.source_item_id, + ) + .join( + ProjectSetupRun, + ProjectSetupRun.id == GuideSourceArtifactBinding.project_setup_run_id, + ) + .where( + GuideSourceArtifactBinding.id == str(request.binding_id), + GuideSourceArtifactBinding.project_id == str(request.project_id), + GuideSourceArtifactBinding.guide_id == str(request.guide_id), + GuideSourceArtifactBinding.source_snapshot_id + == str(request.source_snapshot_id), + GuideSourceArtifactBinding.source_item_id == str(request.source_item_id), + GuideSourceArtifactBinding.project_setup_run_id + == str(request.project_setup_run_id), + GuideSourceArtifactBinding.setup_generation == request.setup_generation, + GuideSourceFormatClassification.id == str(request.classification_id), + GuideSourceFormatClassification.content_id + == GuideSourceArtifactBinding.content_id, + GuideSourceFormatClassification.setup_generation + == GuideSourceArtifactBinding.setup_generation, + ProjectGuide.status == "draft", + GuideSourceSnapshot.project_id == GuideSourceArtifactBinding.project_id, + GuideSourceSnapshot.guide_id == GuideSourceArtifactBinding.guide_id, + GuideSourceSnapshot.guide_version == ProjectGuide.version, + GuideSourceSnapshotItem.source_snapshot_id + == GuideSourceArtifactBinding.source_snapshot_id, + ProjectSetupRun.project_id == GuideSourceArtifactBinding.project_id, + ProjectSetupRun.guide_version == ProjectGuide.version, + ProjectSetupRun.source_snapshot_hash == GuideSourceSnapshot.bundle_hash, + ProjectSetupRun.setup_generation == latest_generation, + GuideSourceFormatClassification.sha256 == ArtifactContent.sha256, + GuideSourceFormatClassification.byte_count == ArtifactContent.byte_count, + GuideSourceFormatClassification.media_type == ArtifactContent.media_type, + GuideSourceFormatClassification.detector_name == DETECTOR_NAME, + GuideSourceFormatClassification.detector_version == DETECTOR_VERSION, + ) + .with_for_update( + of=( + GuideSourceArtifactBinding, + ProjectGuide, + ProjectSetupRun, + ArtifactContent, + ) + ) + ) + ).one_or_none() + if row is None: + return None + binding, classification, content = row + return _ExtractionFacts( + project_id=binding.project_id, + guide_id=binding.guide_id, + snapshot_id=binding.source_snapshot_id, + item_id=binding.source_item_id, + setup_run_id=binding.project_setup_run_id, + setup_generation=binding.setup_generation, + binding_id=binding.id, + classification_id=classification.id, + content_id=content.id, + sha256=content.sha256, + byte_count=content.byte_count, + detected_format=classification.detected_format, + classification_status=classification.status, + ) + + async def _persist_failure( + self, + session: AsyncSession, + facts: _ExtractionFacts, + *, + status: str, + error_code: str, + ) -> GuideExtractionPersistenceResult: + attempt_number = 1 + int( + await session.scalar( + select( + func.coalesce(func.max(GuideSourceExtractionAttempt.attempt_number), 0) + ).where( + GuideSourceExtractionAttempt.binding_id == facts.binding_id, + GuideSourceExtractionAttempt.policy_version == EXTRACTION_POLICY_VERSION, + ) + ) + or 0 + ) + attempt = GuideSourceExtractionAttempt( + id=str(uuid4()), + binding_id=facts.binding_id, + content_id=facts.content_id, + classification_id=facts.classification_id, + setup_generation=facts.setup_generation, + detected_format=facts.detected_format, + extractor_name=f"workstream.{facts.detected_format}", + extractor_version="1", + policy_version=EXTRACTION_POLICY_VERSION, + attempt_number=attempt_number, + status=status, + error_code=error_code, + bounded_facts={}, + ) + session.add(attempt) + await session.flush() + return self._result(attempt) + + @staticmethod + def _result( + attempt: GuideSourceExtractionAttempt, + *, + content: GuideSourceExtractedContent | None = None, + usage: GuideSourceExtractionUsage | None = None, + replayed: bool = False, + ) -> GuideExtractionPersistenceResult: + return GuideExtractionPersistenceResult( + attempt_id=UUID(attempt.id), + status=attempt.status, + error_code=attempt.error_code, + extracted_content_id=None if content is None else UUID(content.id), + usage_id=None if usage is None else UUID(usage.id), + replayed=replayed, + ) + + +class GuideExtractionCoordinator: + """Apply the one permitted fresh-authority/materialization executor retry.""" + + def __init__( + self, + service: GuideExtractionService, + materializer: FreshAuthorizedGuideMaterializer, + ) -> None: + self._service = service + self._materializer = materializer + + async def extract(self, request: GuideExtractionRequest) -> GuideExtractionPersistenceResult: + """Retry one executor failure only after obtaining a completely fresh source.""" + for attempt_index in range(2): + exhausted = await self._service.claim_materialization_slot(request) + if exhausted is not None: + return exhausted + prepared = await self._materializer.materialize_with_fresh_authority(request) + try: + result = await self._service.extract_prepared(request, prepared) + finally: + await prepared.close() + if result.status != "parser_failure" or attempt_index == 1: + return result + raise AssertionError("bounded guide extraction retry exhausted") diff --git a/backend/app/modules/artifacts/guide_extraction_worker.py b/backend/app/modules/artifacts/guide_extraction_worker.py new file mode 100644 index 00000000..62f590c2 --- /dev/null +++ b/backend/app/modules/artifacts/guide_extraction_worker.py @@ -0,0 +1,199 @@ +"""Descriptor-only standard-library guide extraction child.""" + +from __future__ import annotations + +import csv +import ctypes +from io import StringIO +import json +import math +import os +import re +import resource +import sys + + +_ALLOW = 0x7FFF0000 +_ERRNO_EPERM = 0x00050001 +_MAXIMUM_INPUT_BYTES = 32 * 1024 * 1024 +_INVALID_CONTROL = re.compile(r"[\x00-\x08\x0b-\x1f\x7f-\x9f]") +_ALLOWED_SYSCALLS = ( + "read", + "write", + "close", + "fstat", + "lseek", + "mmap", + "mprotect", + "mremap", + "munmap", + "brk", + "madvise", + "futex", + "clock_gettime", + "rt_sigaction", + "rt_sigprocmask", + "rt_sigreturn", + "sigaltstack", + "getrandom", + "exit", + "exit_group", +) + + +class ExtractionFailure(Exception): + """Carry one bounded child outcome.""" + + def __init__(self, status: str, code: str) -> None: + super().__init__(code) + self.status = status + self.code = code + + +def _install_limits(*, cpu_soft_seconds: int = 29, cpu_hard_seconds: int = 30) -> None: + resource.setrlimit(resource.RLIMIT_CORE, (0, 0)) + resource.setrlimit(resource.RLIMIT_CPU, (cpu_soft_seconds, cpu_hard_seconds)) + resource.setrlimit(resource.RLIMIT_AS, (512 * 1024 * 1024,) * 2) + resource.setrlimit(resource.RLIMIT_FSIZE, (4 * 1024 * 1024,) * 2) + resource.setrlimit(resource.RLIMIT_NOFILE, (32, 32)) + if hasattr(resource, "RLIMIT_NPROC"): + resource.setrlimit(resource.RLIMIT_NPROC, (0, 0)) + + +def _install_seccomp() -> None: + try: + library = ctypes.CDLL("libseccomp.so.2", use_errno=True) + except OSError as exc: + raise ExtractionFailure("parser_failure", "isolation_unavailable") from exc + library.seccomp_init.argtypes = [ctypes.c_uint32] + library.seccomp_init.restype = ctypes.c_void_p + library.seccomp_rule_add.argtypes = [ + ctypes.c_void_p, + ctypes.c_uint32, + ctypes.c_int, + ctypes.c_uint, + ] + library.seccomp_rule_add.restype = ctypes.c_int + library.seccomp_syscall_resolve_name.argtypes = [ctypes.c_char_p] + library.seccomp_syscall_resolve_name.restype = ctypes.c_int + library.seccomp_load.argtypes = [ctypes.c_void_p] + library.seccomp_load.restype = ctypes.c_int + library.seccomp_release.argtypes = [ctypes.c_void_p] + context = library.seccomp_init(_ERRNO_EPERM) + if not context: + raise ExtractionFailure("parser_failure", "isolation_unavailable") + try: + for name in _ALLOWED_SYSCALLS: + number = library.seccomp_syscall_resolve_name(name.encode("ascii")) + if number < 0 or library.seccomp_rule_add(context, _ALLOW, number, 0) != 0: + raise ExtractionFailure("parser_failure", "isolation_unavailable") + if library.seccomp_load(context) != 0: + raise ExtractionFailure("parser_failure", "isolation_unavailable") + finally: + library.seccomp_release(context) + + +def _decode_text(payload: bytes) -> str: + if payload.startswith(b"\xef\xbb\xbf"): + payload = payload[3:] + if b"\xef\xbb\xbf" in payload: + raise ExtractionFailure("malformed", "invalid_encoding") + try: + value = payload.decode("utf-8") + except UnicodeDecodeError as exc: + raise ExtractionFailure("malformed", "invalid_encoding") from exc + value = value.replace("\r\n", "\n").replace("\r", "\n") + if _INVALID_CONTROL.search(value): + raise ExtractionFailure("malformed", "invalid_control_character") + return value + + +def _reject_constant(_: str) -> None: + raise ExtractionFailure("malformed", "invalid_json_number") + + +def _object(pairs: list[tuple[str, object]]) -> dict[str, object]: + result: dict[str, object] = {} + for key, value in pairs: + if key in result: + raise ExtractionFailure("malformed", "duplicate_json_key") + result[key] = value + return result + + +def _validate_json(value: object, depth: int = 0) -> None: + if isinstance(value, (dict, list)): + if depth > 64: + raise ExtractionFailure("limit_exceeded", "json_nesting_limit") + values = value.values() if isinstance(value, dict) else value + for child in values: + _validate_json(child, depth + 1) + elif isinstance(value, float) and not math.isfinite(value): + raise ExtractionFailure("malformed", "invalid_json_number") + + +def _extract(payload: bytes, detected_format: str) -> str: + text = _decode_text(payload) + if detected_format in {"plain_text", "markdown"}: + return text + if detected_format == "json": + try: + value = json.loads(text, object_pairs_hook=_object, parse_constant=_reject_constant) + except ExtractionFailure: + raise + except (json.JSONDecodeError, RecursionError, ValueError) as exc: + raise ExtractionFailure("malformed", "invalid_json") from exc + _validate_json(value, 1) + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + if detected_format == "csv": + rows: list[list[str]] = [] + cells = 0 + try: + csv.field_size_limit(_MAXIMUM_INPUT_BYTES + 1) + reader = csv.reader(StringIO(text, newline=""), dialect="excel", strict=True) + for row in reader: + if len(rows) >= 100_000: + raise ExtractionFailure("limit_exceeded", "csv_row_limit") + cells += len(row) + if cells > 1_000_000: + raise ExtractionFailure("limit_exceeded", "csv_cell_limit") + if any(len(cell) > 32_768 for cell in row): + raise ExtractionFailure("limit_exceeded", "csv_cell_size_limit") + rows.append(row) + except csv.Error as exc: + raise ExtractionFailure("malformed", "invalid_csv") from exc + return json.dumps(rows, separators=(",", ":"), ensure_ascii=False) + raise ExtractionFailure("unsupported", "unsupported_format") + + +def main() -> int: + """Apply resource/isolation controls and emit one bounded JSON result.""" + detected_format = sys.argv[1] if len(sys.argv) == 2 else "" + try: + _install_limits() + _install_seccomp() + payload = sys.stdin.buffer.read(_MAXIMUM_INPUT_BYTES + 1) + if len(payload) > _MAXIMUM_INPUT_BYTES: + raise ExtractionFailure("limit_exceeded", "input_limit") + output = _extract(payload, detected_format) + if len(output.encode("utf-8")) > 4 * 1024 * 1024: + raise ExtractionFailure("limit_exceeded", "output_limit") + result = {"status": "extracted", "error_code": None, "output": output} + except ExtractionFailure as exc: + result = {"status": exc.status, "error_code": exc.code, "output": None} + except MemoryError: + result = {"status": "limit_exceeded", "error_code": "memory_limit", "output": None} + except BaseException: + result = {"status": "parser_failure", "error_code": "parser_failure", "output": None} + encoded = json.dumps(result, separators=(",", ":"), ensure_ascii=True).encode("utf-8") + view = memoryview(encoded) + while view: + written = os.write(1, view) + if written <= 0: + raise OSError("guide extraction result write made no progress") + view = view[written:] + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/app/modules/artifacts/models.py b/backend/app/modules/artifacts/models.py index 3b68f8e5..7684e55b 100644 --- a/backend/app/modules/artifacts/models.py +++ b/backend/app/modules/artifacts/models.py @@ -14,6 +14,7 @@ Integer, JSON, String, + Text, UniqueConstraint, ) from sqlalchemy.orm import Mapped, mapped_column @@ -28,6 +29,7 @@ "[89ab][0-9a-f]{{3}}-[0-9a-f]{{12}}$'" ) + class ArtifactUploadSession(Base): """Mutable staging authority for one bounded artifact set.""" @@ -176,7 +178,11 @@ class ArtifactBinding(Base): __tablename__ = "artifact_bindings" __table_args__ = ( UniqueConstraint( - "project_id", "resource_type", "resource_id", "logical_role", "scope_version", + "project_id", + "resource_type", + "resource_id", + "logical_role", + "scope_version", name="uq_artifact_binding_scope_version", ), UniqueConstraint("supersedes_binding_id", name="uq_artifact_binding_supersedes"), @@ -253,6 +259,20 @@ class GuideSourceArtifactBinding(Base): "setup_generation", name="uq_guide_bindings_item_generation", ), + UniqueConstraint( + "id", + "content_id", + "setup_generation", + name="uq_guide_bindings_extraction_attempt_lineage", + ), + UniqueConstraint( + "id", + "content_id", + "source_item_id", + "project_setup_run_id", + "setup_generation", + name="uq_guide_bindings_extraction_lineage", + ), UniqueConstraint( "id", "content_id", @@ -300,6 +320,13 @@ class GuideSourceFormatClassification(Base): name="fk_guide_classifications_exact_binding", ), UniqueConstraint("binding_id", name="uq_guide_classifications_binding"), + UniqueConstraint( + "id", + "binding_id", + "content_id", + "setup_generation", + name="uq_guide_classifications_extraction_lineage", + ), CheckConstraint( "status in ('classified', 'unsupported', 'ambiguous', 'malformed', 'limit_exceeded')", name="ck_guide_classifications_status", @@ -371,13 +398,230 @@ class GuideSourceArtifactIncident(Base): created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) +class GuideSourceExtractionAttempt(Base): + """Immutable bounded outcome for one exact extraction execution.""" + + __tablename__ = "guide_source_extraction_attempts" + __table_args__ = ( + ForeignKeyConstraint( + ["binding_id", "content_id", "setup_generation"], + [ + "guide_source_artifact_bindings.id", + "guide_source_artifact_bindings.content_id", + "guide_source_artifact_bindings.setup_generation", + ], + name="fk_guide_extraction_attempts_exact_binding", + ), + ForeignKeyConstraint( + ["classification_id", "binding_id", "content_id", "setup_generation"], + [ + "guide_source_format_classifications.id", + "guide_source_format_classifications.binding_id", + "guide_source_format_classifications.content_id", + "guide_source_format_classifications.setup_generation", + ], + name="fk_guide_extraction_attempts_exact_classification", + ), + CheckConstraint( + "status in ('extracted','unsupported','ambiguous','malformed','limit_exceeded'," + "'parser_failure','cancelled','artifact_incident')", + name="ck_guide_extraction_attempts_status", + ), + CheckConstraint("attempt_number > 0", name="ck_guide_extraction_attempts_number"), + CheckConstraint( + "(status = 'extracted') = (error_code is null)", + name="ck_guide_extraction_attempts_error", + ), + UniqueConstraint( + "binding_id", "policy_version", "attempt_number", name="uq_guide_extraction_attempts" + ), + UniqueConstraint( + "id", + "binding_id", + "content_id", + "setup_generation", + "status", + name="uq_guide_extraction_attempts_exact_usage", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True) + binding_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + content_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + classification_id: Mapped[str] = mapped_column(String(36), nullable=False) + setup_generation: Mapped[int] = mapped_column(BigInteger, nullable=False) + detected_format: Mapped[str] = mapped_column(String(40), nullable=False) + extractor_name: Mapped[str] = mapped_column(String(100), nullable=False) + extractor_version: Mapped[str] = mapped_column(String(40), nullable=False) + policy_version: Mapped[str] = mapped_column(String(80), nullable=False) + attempt_number: Mapped[int] = mapped_column(BigInteger, nullable=False) + status: Mapped[str] = mapped_column(String(40), nullable=False) + error_code: Mapped[str | None] = mapped_column(String(80)) + bounded_facts: Mapped[dict] = mapped_column(JSON, nullable=False) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + + +class GuideSourceExtractionRetryBudget(Base): + """Durable two-slot materialization budget for one exact extraction lineage.""" + + __tablename__ = "guide_source_extraction_retry_budgets" + __table_args__ = ( + ForeignKeyConstraint( + ["binding_id", "content_id", "setup_generation"], + [ + "guide_source_artifact_bindings.id", + "guide_source_artifact_bindings.content_id", + "guide_source_artifact_bindings.setup_generation", + ], + name="fk_guide_extraction_retry_budgets_exact_binding", + ), + ForeignKeyConstraint( + ["classification_id", "binding_id", "content_id", "setup_generation"], + [ + "guide_source_format_classifications.id", + "guide_source_format_classifications.binding_id", + "guide_source_format_classifications.content_id", + "guide_source_format_classifications.setup_generation", + ], + name="fk_guide_extraction_retry_budgets_exact_classification", + ), + CheckConstraint( + "claimed_slots between 1 and 2", + name="ck_guide_extraction_retry_budgets_slots", + ), + ) + + binding_id: Mapped[str] = mapped_column(String(36), primary_key=True) + content_id: Mapped[str] = mapped_column(String(36), nullable=False) + classification_id: Mapped[str] = mapped_column(String(36), nullable=False) + setup_generation: Mapped[int] = mapped_column(BigInteger, nullable=False) + policy_version: Mapped[str] = mapped_column(String(80), nullable=False) + claimed_slots: Mapped[int] = mapped_column(Integer, nullable=False) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now() + ) + + +class GuideSourceExtractedContent(Base): + """Immutable deterministic successful extraction keyed only by content semantics.""" + + __tablename__ = "guide_source_extracted_contents" + __table_args__ = ( + UniqueConstraint( + "content_id", + "detected_format", + "extractor_name", + "extractor_version", + "policy_version", + name="uq_guide_extracted_contents_identity", + ), + UniqueConstraint("id", "content_id", name="uq_guide_extracted_contents_exact_usage"), + CheckConstraint("status = 'extracted'", name="ck_guide_extracted_contents_status"), + CheckConstraint( + SHA256_CHECK.format(column="source_sha256"), + name="ck_guide_extracted_contents_source_sha256", + ), + CheckConstraint( + SHA256_CHECK.format(column="output_sha256"), + name="ck_guide_extracted_contents_output_sha256", + ), + CheckConstraint("source_byte_count >= 0", name="ck_guide_extracted_contents_source_size"), + CheckConstraint( + "octet_length(canonical_output) <= 4194304", + name="ck_guide_extracted_contents_output_size", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True) + content_id: Mapped[str] = mapped_column( + ForeignKey("artifact_contents.id", ondelete="RESTRICT"), nullable=False, index=True + ) + detected_format: Mapped[str] = mapped_column(String(40), nullable=False) + extractor_name: Mapped[str] = mapped_column(String(100), nullable=False) + extractor_version: Mapped[str] = mapped_column(String(40), nullable=False) + policy_version: Mapped[str] = mapped_column(String(80), nullable=False) + source_sha256: Mapped[str] = mapped_column(String(71), nullable=False) + source_byte_count: Mapped[int] = mapped_column(BigInteger, nullable=False) + status: Mapped[str] = mapped_column(String(40), nullable=False) + output_sha256: Mapped[str] = mapped_column(String(71), nullable=False) + canonical_output: Mapped[str] = mapped_column(Text, nullable=False) + omission_facts: Mapped[dict] = mapped_column(JSON, nullable=False) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + + +class GuideSourceExtractionUsage(Base): + """Exact current-lineage use of one deterministic extracted content record.""" + + __tablename__ = "guide_source_extraction_usages" + __table_args__ = ( + ForeignKeyConstraint( + [ + "binding_id", + "content_id", + "source_item_id", + "project_setup_run_id", + "setup_generation", + ], + [ + "guide_source_artifact_bindings.id", + "guide_source_artifact_bindings.content_id", + "guide_source_artifact_bindings.source_item_id", + "guide_source_artifact_bindings.project_setup_run_id", + "guide_source_artifact_bindings.setup_generation", + ], + name="fk_guide_extraction_usages_exact_binding", + ), + ForeignKeyConstraint( + [ + "extraction_attempt_id", + "binding_id", + "content_id", + "setup_generation", + "attempt_status", + ], + [ + "guide_source_extraction_attempts.id", + "guide_source_extraction_attempts.binding_id", + "guide_source_extraction_attempts.content_id", + "guide_source_extraction_attempts.setup_generation", + "guide_source_extraction_attempts.status", + ], + name="fk_guide_extraction_usages_exact_attempt", + ), + ForeignKeyConstraint( + ["extracted_content_id", "content_id"], + ["guide_source_extracted_contents.id", "guide_source_extracted_contents.content_id"], + name="fk_guide_extraction_usages_exact_content", + ), + UniqueConstraint("binding_id", "extracted_content_id", name="uq_guide_extraction_usages"), + CheckConstraint( + "attempt_status = 'extracted'", + name="ck_guide_extraction_usages_successful_attempt", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True) + extracted_content_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + extraction_attempt_id: Mapped[str] = mapped_column(String(36), nullable=False) + attempt_status: Mapped[str] = mapped_column(String(40), nullable=False) + binding_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + content_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + source_item_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + project_setup_run_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + setup_generation: Mapped[int] = mapped_column(BigInteger, nullable=False) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + + class ArtifactStorageNamespace(Base): """Immutable singleton fencing one deployment to one storage namespace.""" __tablename__ = "artifact_storage_namespaces" __table_args__ = ( CheckConstraint("id = 'primary'", name="singleton_id"), - CheckConstraint(SHA256_CHECK.format(column="namespace_fingerprint"), name="fingerprint_shape"), + CheckConstraint( + SHA256_CHECK.format(column="namespace_fingerprint"), name="fingerprint_shape" + ), UniqueConstraint( "namespace_fingerprint", name="uq_artifact_storage_namespace_fingerprint", @@ -678,7 +922,9 @@ class ArtifactReplica(Base): "integrity_state in ('unknown', 'valid', 'invalid')", name="integrity_state", ), - CheckConstraint(SHA256_CHECK.format(column="namespace_fingerprint"), name="fingerprint_shape"), + CheckConstraint( + SHA256_CHECK.format(column="namespace_fingerprint"), name="fingerprint_shape" + ), ) id: Mapped[str] = mapped_column(String(36), primary_key=True) @@ -869,12 +1115,8 @@ class ArtifactRecoveryAttempt(Base): "client_idempotency_key", name="uq_artifact_recovery_idempotency", ), - UniqueConstraint( - "source_verification_job_id", name="uq_artifact_recovery_source_job" - ), - UniqueConstraint( - "retry_verification_job_id", name="uq_artifact_recovery_retry_job" - ), + UniqueConstraint("source_verification_job_id", name="uq_artifact_recovery_source_job"), + UniqueConstraint("retry_verification_job_id", name="uq_artifact_recovery_retry_job"), CheckConstraint( "source_verification_job_id <> retry_verification_job_id", name="distinct_jobs", diff --git a/backend/app/modules/artifacts/preparation.py b/backend/app/modules/artifacts/preparation.py index aef86802..dc91bc58 100644 --- a/backend/app/modules/artifacts/preparation.py +++ b/backend/app/modules/artifacts/preparation.py @@ -37,6 +37,7 @@ CommittedArtifactSource, PreparedArtifact, PreparedArtifactInspector, + PreparedGuideExtractor, ) @@ -45,6 +46,7 @@ _LEDGER_VERSION = 2 _LEDGER_MAXIMUM_BYTES = 1024 * 1024 _RESERVATION_ID = re.compile(r"^[0-9a-f]{32}$") +_WORKSPACE_ID = re.compile(r"^extract_[0-9a-f]{32}$") _SCRATCH_FILE = re.compile(r"^prep_([0-9a-f]{32})\.bin$") _LEDGER_TEMP_FILE = re.compile(r"^\.ledger\.[0-9a-f]{32}\.tmp$") _PROCESS_IDENTITY = re.compile(r"^[0-9a-f]{64}$") @@ -214,23 +216,31 @@ def __init__(self, *, root: Path, limits: ArtifactPreparationLimits) -> None: self._poisoned_reservations: dict[str, _ScratchReservation] = {} self._pending_allocation_releases: dict[str, _ScratchReservation] = {} self._pending_readers: dict[str, _PendingReaderCleanup] = {} + self._pending_workspaces: set[str] = set() expected_root = self._initialize_root(root) self._root_fd = os.open( self._root, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0), ) self._files_fd = -1 + self._workspaces_fd = -1 try: self._assert_opened_root(self._root_fd, expected_root=expected_root) self._initialize_layout() self._files_fd = self._open_directory("files") + self._workspaces_fd = self._open_directory("workspaces") with self._locked_ledger(): if self._read_ledger_optional() is None: - self._write_ledger({"version": _LEDGER_VERSION, "reservations": []}) + self._write_ledger( + {"version": _LEDGER_VERSION, "reservations": [], "workspaces": []} + ) except BaseException: if self._files_fd >= 0: os.close(self._files_fd) self._files_fd = -1 + if self._workspaces_fd >= 0: + os.close(self._workspaces_fd) + self._workspaces_fd = -1 os.close(self._root_fd) self._root_fd = -1 raise @@ -247,6 +257,7 @@ def pending_cleanup_count(self) -> int: len(self._pending_allocation_releases) + len(self._pending_readers) + len(self._poisoned_reservations) + + len(self._pending_workspaces) ) async def allocate(self) -> tuple[_ScratchReservation, int]: @@ -281,14 +292,10 @@ async def allocate(self) -> tuple[_ScratchReservation, int]: asyncio.to_thread(self._release_sync, reservation) ) except BaseException: - self._pending_allocation_releases[ - reservation.reservation_id - ] = reservation + self._pending_allocation_releases[reservation.reservation_id] = reservation raise cancellation except _AllocationCleanupRequired as exc: - self._pending_allocation_releases[exc.reservation.reservation_id] = ( - exc.reservation - ) + self._pending_allocation_releases[exc.reservation.reservation_id] = exc.reservation raise except _DescriptorOwnershipUncertain as exc: self._retain_uncertain_descriptor(exc.reservation) @@ -314,8 +321,8 @@ async def seal_for_read(self, reservation: _ScratchReservation, descriptor: int) raise cancellation from None except BaseException: if reader is not None: - self._pending_readers[reservation.reservation_id] = ( - _PendingReaderCleanup(reservation=reservation, reader=reader) + self._pending_readers[reservation.reservation_id] = _PendingReaderCleanup( + reservation=reservation, reader=reader ) raise cancellation from None raise @@ -327,13 +334,9 @@ async def release(self, reservation: _ScratchReservation) -> None: """Remove one scratch file and reservation under the ledger lock.""" with self._tracked_operation(): if reservation.reservation_id in self._poisoned_reservations: - raise ArtifactScratchIntegrityError( - "artifact descriptor ownership is uncertain" - ) + raise ArtifactScratchIntegrityError("artifact descriptor ownership is uncertain") if reservation.reservation_id in self._pending_readers: - raise ArtifactScratchIntegrityError( - "artifact reader cleanup is still pending" - ) + raise ArtifactScratchIntegrityError("artifact reader cleanup is still pending") async def release_and_forget() -> None: """Durably release and clear local ownership as one handoff.""" @@ -364,14 +367,75 @@ async def cleanup_stale(self, *, now_unix_ns: int | None = None) -> int: protected_reservation_ids = frozenset( self._owned_reservations | self._poisoned_reservations ) - cleaned_reservation_ids = await self._run_io( + cleaned_reservation_ids, cleaned_workspace_names = await self._run_io( self._cleanup_stale_sync, now, protected_reservation_ids, ) for reservation_id in cleaned_reservation_ids: self._pending_allocation_releases.pop(reservation_id, None) - return len(cleaned_reservation_ids) + return len(cleaned_reservation_ids) + len(cleaned_workspace_names) + + def _reserve_workspace_sync(self, workspace_name: str) -> None: + """Publish crash-recovery ownership before creating a workspace.""" + now = time.time_ns() + expires_at = now + int(self._limits.reservation_ttl_seconds * 1_000_000_000) + with self._locked_ledger(): + ledger = self._read_ledger() + workspaces = list(ledger["workspaces"]) + workspaces.append( + { + "workspace_name": workspace_name, + "created_at_unix_ns": now, + "expires_at_unix_ns": expires_at, + "owner_pid": os.getpid(), + "owner_process_identity": self._owner_process_identity, + } + ) + self._write_ledger({**ledger, "workspaces": workspaces}) + + def _release_workspace_sync(self, workspace_name: str) -> None: + """Remove residue before releasing durable workspace ownership.""" + with self._locked_ledger(): + ledger = self._read_ledger() + try: + self._cleanup_workspace_sync(workspace_name) + except FileNotFoundError: + pass + self._write_ledger( + { + **ledger, + "workspaces": [ + entry + for entry in ledger["workspaces"] + if entry["workspace_name"] != workspace_name + ], + } + ) + + @contextmanager + def extraction_workspace(self) -> Iterator[Path]: + """Own one crash-recoverable private extraction workspace.""" + with self._tracked_operation(): + workspace_name = f"extract_{uuid4().hex}" + self._reserve_workspace_sync(workspace_name) + workspace_fd = -1 + try: + os.mkdir(workspace_name, mode=0o700, dir_fd=self._workspaces_fd) + workspace_fd = os.open( + workspace_name, + os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0), + dir_fd=self._workspaces_fd, + ) + yield Path(f"/proc/self/fd/{workspace_fd}") + finally: + if workspace_fd >= 0: + os.close(workspace_fd) + try: + self._release_workspace_sync(workspace_name) + except BaseException: + self._pending_workspaces.add(workspace_name) + raise async def retry_pending_cleanup(self) -> int: """Retry manager-owned cleanup retained before service handoff.""" @@ -387,16 +451,12 @@ async def close_reader_and_release() -> None: self._pending_readers.pop(reservation_id, None) try: - await await_completion_preserving_cancellation( - close_reader_and_release() - ) + await await_completion_preserving_cancellation(close_reader_and_release()) except Exception: continue else: cleaned += 1 - for reservation_id, reservation in tuple( - self._pending_allocation_releases.items() - ): + for reservation_id, reservation in tuple(self._pending_allocation_releases.items()): async def release_reservation() -> None: """Release one retained reservation and clear manager ownership.""" @@ -409,26 +469,73 @@ async def release_reservation() -> None: continue else: cleaned += 1 + for workspace_name in tuple(self._pending_workspaces): + try: + await self._run_io(self._release_workspace_sync, workspace_name) + except Exception: + continue + self._pending_workspaces.remove(workspace_name) + cleaned += 1 return cleaned + def _cleanup_workspace_sync(self, workspace_name: str) -> None: + """Remove bounded workspace residue without following links.""" + workspace_fd = os.open( + workspace_name, + os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0), + dir_fd=self._workspaces_fd, + ) + try: + remaining_entries = [self._limits.maximum_files] + self._cleanup_workspace_entries_sync(workspace_fd, remaining_entries) + finally: + os.close(workspace_fd) + os.rmdir(workspace_name, dir_fd=self._workspaces_fd) + os.fsync(self._workspaces_fd) + + def _cleanup_workspace_entries_sync( + self, + directory_fd: int, + remaining_entries: list[int], + ) -> None: + """Remove a bounded tree bottom-up without following links.""" + for entry in os.listdir(directory_fd): + remaining_entries[0] -= 1 + if remaining_entries[0] < 0: + raise ArtifactScratchIntegrityError( + "artifact extraction workspace entry limit exceeded" + ) + metadata = os.stat(entry, dir_fd=directory_fd, follow_symlinks=False) + if stat_module.S_ISDIR(metadata.st_mode): + child_fd = os.open( + entry, + os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0), + dir_fd=directory_fd, + ) + try: + self._cleanup_workspace_entries_sync(child_fd, remaining_entries) + finally: + os.close(child_fd) + os.rmdir(entry, dir_fd=directory_fd) + else: + os.unlink(entry, dir_fd=directory_fd) + def close(self) -> None: """Release pinned scratch directory descriptors.""" with self._lifecycle_lock: if self._closed: return - if ( - self._in_flight_operations - or self.pending_cleanup_count - or self._owned_reservations - ): + if self._in_flight_operations or self.pending_cleanup_count or self._owned_reservations: raise ArtifactScratchIntegrityError("artifact scratch cleanup is still pending") self._closed = True files_fd = getattr(self, "_files_fd", -1) + workspaces_fd = getattr(self, "_workspaces_fd", -1) root_fd = getattr(self, "_root_fd", -1) self._files_fd = -1 + self._workspaces_fd = -1 self._root_fd = -1 failure: BaseException | None = None - for descriptor in (files_fd, root_fd): + for descriptor in (files_fd, workspaces_fd, root_fd): if descriptor < 0: continue try: @@ -573,6 +680,12 @@ def _initialize_layout_locked(self) -> None: except FileExistsError: pass os.fsync(self._root_fd) + if "workspaces" not in entries: + try: + os.mkdir("workspaces", mode=0o700, dir_fd=self._root_fd) + except FileExistsError: + pass + os.fsync(self._root_fd) def _validate_existing_root_marker(self) -> None: """Require the root marker to match this manager's canonical limits.""" @@ -621,16 +734,13 @@ def _validate_layout_entries( allowed = { _ROOT_MARKER, "files", + "workspaces", ".ledger.lock", ".ledger.json", } - temporary_entries = { - entry for entry in entries if _LEDGER_TEMP_FILE.fullmatch(entry) - } + temporary_entries = {entry for entry in entries if _LEDGER_TEMP_FILE.fullmatch(entry)} allowed_temps = ( - temporary_entries - if allow_marked_ledger_temps and _ROOT_MARKER in entries - else set() + temporary_entries if allow_marked_ledger_temps and _ROOT_MARKER in entries else set() ) non_bootstrap_entries = entries - {".ledger.lock"} - allowed_temps if entries - allowed - allowed_temps or ( @@ -683,9 +793,7 @@ def _allocate_sync(self) -> tuple[_ScratchReservation, int]: filesystem = os.fstatvfs(self._files_fd) available = filesystem.f_bavail * filesystem.f_frsize required = ( - reserved_bytes - + HARD_MAXIMUM_ARTIFACT_BYTES - + self._limits.minimum_free_bytes + reserved_bytes + HARD_MAXIMUM_ARTIFACT_BYTES + self._limits.minimum_free_bytes ) if available < required: raise ArtifactScratchCapacityError("artifact scratch free-space floor reached") @@ -701,7 +809,7 @@ def _allocate_sync(self) -> tuple[_ScratchReservation, int]: } ) try: - self._write_ledger({"version": _LEDGER_VERSION, "reservations": entries}) + self._write_ledger({**ledger, "reservations": entries}) except BaseException: try: self._restore_failed_reservation_ledger( @@ -739,9 +847,7 @@ def _allocate_sync(self) -> tuple[_ScratchReservation, int]: for entry in entries if entry["reservation_id"] != reservation.reservation_id ] - self._write_ledger( - {"version": _LEDGER_VERSION, "reservations": remaining} - ) + self._write_ledger({**ledger, "reservations": remaining}) except BaseException as cleanup_error: raise _AllocationCleanupRequired(reservation) from cleanup_error raise @@ -753,13 +859,12 @@ def _restore_failed_reservation_ledger( previous_entries: list[dict[str, Any]], ) -> None: """Prove an ambiguously published reservation was durably removed.""" - current = self._read_ledger()["reservations"] + ledger = self._read_ledger() + current = ledger["reservations"] if not any(entry["reservation_id"] == reservation_id for entry in current): return try: - self._write_ledger( - {"version": _LEDGER_VERSION, "reservations": previous_entries} - ) + self._write_ledger({**ledger, "reservations": previous_entries}) except BaseException as exc: raise ArtifactScratchIntegrityError( "failed artifact scratch reservation could not be rolled back" @@ -782,9 +887,7 @@ def _seal_for_read_sync(self, reservation: _ScratchReservation, descriptor: int) if entry["reservation_id"] == reservation.reservation_id ] if len(matching) != 1 or matching[0]["filename"] != reservation.filename: - raise ArtifactScratchIntegrityError( - "artifact scratch reservation is invalid" - ) + raise ArtifactScratchIntegrityError("artifact scratch reservation is invalid") os.fsync(descriptor) os.fchmod(descriptor, 0o400) read_descriptor = os.open( @@ -855,7 +958,7 @@ def _release_sync(self, reservation: _ScratchReservation) -> None: remaining = [ entry for entry in entries if entry["reservation_id"] != reservation.reservation_id ] - self._write_ledger({"version": _LEDGER_VERSION, "reservations": remaining}) + self._write_ledger({**ledger, "reservations": remaining}) def _usage_sync(self) -> ArtifactScratchUsage: """Read aggregate ledger usage while holding the cross-process lock.""" @@ -882,11 +985,12 @@ def _cleanup_stale_sync( self, now_unix_ns: int, protected_reservation_ids: frozenset[str], - ) -> tuple[str, ...]: + ) -> tuple[tuple[str, ...], tuple[str, ...]]: """Durably remove expired files before deleting their ledger entries.""" with self._locked_ledger(): ledger = self._read_ledger() entries = list(ledger["reservations"]) + workspaces = list(ledger["workspaces"]) stale = sorted( ( entry @@ -902,17 +1006,42 @@ def _cleanup_stale_sync( ) for entry in stale: self._unlink_regular_optional(entry["filename"]) - if stale: + stale_workspaces = sorted( + ( + entry + for entry in workspaces + if entry["expires_at_unix_ns"] <= now_unix_ns + and not self._owner_process_matches( + entry["owner_pid"], entry["owner_process_identity"] + ) + ), + key=lambda entry: entry["workspace_name"], + ) + for entry in stale_workspaces: + try: + self._cleanup_workspace_sync(entry["workspace_name"]) + except FileNotFoundError: + pass + if stale or stale_workspaces: stale_ids = {entry["reservation_id"] for entry in stale} + stale_workspace_names = {entry["workspace_name"] for entry in stale_workspaces} self._write_ledger( { - "version": _LEDGER_VERSION, + **ledger, "reservations": [ entry for entry in entries if entry["reservation_id"] not in stale_ids ], + "workspaces": [ + entry + for entry in workspaces + if entry["workspace_name"] not in stale_workspace_names + ], } ) - return tuple(entry["reservation_id"] for entry in stale) + return ( + tuple(entry["reservation_id"] for entry in stale), + tuple(entry["workspace_name"] for entry in stale_workspaces), + ) def _owner_process_matches(self, owner_pid: int, expected_identity: str) -> bool: """Retain custody only while PID and process-start identity still match.""" @@ -926,12 +1055,8 @@ def _owner_process_matches(self, owner_pid: int, expected_identity: str) -> bool def _read_process_identity(owner_pid: int) -> str | None: """Bind a Linux PID to its boot and process-start identity.""" try: - boot_id = Path("/proc/sys/kernel/random/boot_id").read_text( - encoding="ascii" - ).strip() - process_stat = Path(f"/proc/{owner_pid}/stat").read_text( - encoding="ascii" - ) + boot_id = Path("/proc/sys/kernel/random/boot_id").read_text(encoding="ascii").strip() + process_stat = Path(f"/proc/{owner_pid}/stat").read_text(encoding="ascii") except FileNotFoundError: return None _, separator, remainder = process_stat.rpartition(")") @@ -992,6 +1117,12 @@ def _read_ledger_optional(self) -> dict[str, Any] | None: ledger = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise ArtifactScratchIntegrityError("artifact scratch ledger is invalid") from exc + if ( + isinstance(ledger, dict) + and set(ledger) == {"version", "reservations"} + and ledger.get("version") == _LEDGER_VERSION + ): + ledger = {**ledger, "workspaces": []} self._validate_ledger(ledger) return ledger @@ -1052,7 +1183,11 @@ def _write_ledger(self, ledger: dict[str, Any]) -> None: def _validate_ledger(self, ledger: object) -> None: """Validate the complete bounded ledger schema and reservation set.""" - if not isinstance(ledger, dict) or set(ledger) != {"version", "reservations"}: + if not isinstance(ledger, dict) or set(ledger) != { + "version", + "reservations", + "workspaces", + }: raise ArtifactScratchIntegrityError("artifact scratch ledger is invalid") if type(ledger["version"]) is not int or ledger["version"] != _LEDGER_VERSION: raise ArtifactScratchIntegrityError("artifact scratch ledger is invalid") @@ -1097,6 +1232,38 @@ def _validate_ledger(self, ledger: object) -> None: or _PROCESS_IDENTITY.fullmatch(entry["owner_process_identity"]) is None ): raise ArtifactScratchIntegrityError("artifact scratch ledger is invalid") + workspaces = ledger["workspaces"] + if not isinstance(workspaces, list) or len(workspaces) > 1024: + raise ArtifactScratchIntegrityError("artifact scratch ledger is invalid") + seen_workspaces: set[str] = set() + expected_workspace = { + "workspace_name", + "created_at_unix_ns", + "expires_at_unix_ns", + "owner_pid", + "owner_process_identity", + } + for entry in workspaces: + if not isinstance(entry, dict) or set(entry) != expected_workspace: + raise ArtifactScratchIntegrityError("artifact scratch ledger is invalid") + workspace_name = entry["workspace_name"] + integer_values = ( + entry["created_at_unix_ns"], + entry["expires_at_unix_ns"], + entry["owner_pid"], + ) + if ( + not isinstance(workspace_name, str) + or _WORKSPACE_ID.fullmatch(workspace_name) is None + or workspace_name in seen_workspaces + or any(type(value) is not int or value < 0 for value in integer_values) + or entry["expires_at_unix_ns"] <= entry["created_at_unix_ns"] + or entry["owner_pid"] <= 0 + or not isinstance(entry["owner_process_identity"], str) + or _PROCESS_IDENTITY.fullmatch(entry["owner_process_identity"]) is None + ): + raise ArtifactScratchIntegrityError("artifact scratch ledger is invalid") + seen_workspaces.add(workspace_name) @staticmethod def _validate_reservation(reservation: _ScratchReservation) -> None: @@ -1132,9 +1299,7 @@ def _open_directory(self, name: str) -> int: try: descriptor = os.open( name, - os.O_RDONLY - | getattr(os, "O_DIRECTORY", 0) - | getattr(os, "O_NOFOLLOW", 0), + os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0), dir_fd=self._root_fd, ) except OSError as exc: @@ -1387,6 +1552,32 @@ def inspect_and_rewind() -> _InspectionResult: "artifact preparation deadline exceeded" ) from None + async def extract_prepared_guide( + self, + binding: object, + extractor: PreparedGuideExtractor[_InspectionResult], + ) -> _InspectionResult: + """Run one trusted inspector in a private scratch-owned empty directory.""" + active = self._active.get(binding) + if active is None or not active.handle_issued or active.stream_claimed: + raise ArtifactScratchIntegrityError("prepared artifact source is unavailable") + + def inspect_and_cleanup() -> _InspectionResult: + with self._manager.extraction_workspace() as workspace: + active.reader.seek(0) + try: + return extractor.inspect(active.reader, workspace) + finally: + active.reader.seek(0) + + try: + async with asyncio.timeout_at(active.deadline): + return await self._run_io(inspect_and_cleanup) + except TimeoutError: + raise ArtifactPreparationDeadlineError( + "artifact preparation deadline exceeded" + ) from None + async def _release_unhanded_preparation( self, pending: _PendingPreparationCleanup, @@ -1472,9 +1663,7 @@ async def _verify_reader_commitment( f"sha256:{digest.hexdigest()}" != commitment.sha256 or byte_count != commitment.byte_count ): - raise ArtifactScratchIntegrityError( - "prepared artifact bytes changed after commitment" - ) + raise ArtifactScratchIntegrityError("prepared artifact bytes changed after commitment") await self._run_io(reader.seek, 0) @staticmethod diff --git a/backend/app/modules/artifacts/sources.py b/backend/app/modules/artifacts/sources.py index 1cd39c44..21136f60 100644 --- a/backend/app/modules/artifacts/sources.py +++ b/backend/app/modules/artifacts/sources.py @@ -5,6 +5,7 @@ import asyncio from collections.abc import AsyncIterator from dataclasses import dataclass +from pathlib import Path from typing import BinaryIO, Protocol, TypeVar, final from app.core.cancellation import await_completion_preserving_cancellation @@ -22,6 +23,13 @@ def inspect(self, reader: BinaryIO) -> _InspectionResultCo: """Return bounded structural facts without retaining the reader.""" +class PreparedGuideExtractor(Protocol[_InspectionResultCo]): + """Artifact-owned guide extractor over sealed bytes and ephemeral scratch.""" + + def inspect(self, reader: BinaryIO, workspace: Path) -> _InspectionResultCo: + """Return bounded facts without retaining either capability.""" + + class _PreparedArtifactOwner(Protocol): """Private lifecycle operations retained by one preparation service.""" @@ -42,6 +50,13 @@ async def inspect_prepared_artifact( ) -> _InspectionResult: """Run a trusted bounded inspector against the owned scratch reader.""" + async def extract_prepared_guide( + self, + binding: object, + extractor: PreparedGuideExtractor[_InspectionResult], + ) -> _InspectionResult: + """Run a child inspector in one scratch-owned workspace.""" + def claim_prepared_commitment(self, binding: object) -> ArtifactCommitment: """Claim the server-computed commitment for one registered binding.""" @@ -217,6 +232,15 @@ async def inspect( raise RuntimeError("prepared artifact is closed") return await self._owner.inspect_prepared_artifact(self._binding, inspector) + async def extract_guide( + self, + extractor: PreparedGuideExtractor[_InspectionResult], + ) -> _InspectionResult: + """Inspect through one scratch-owned workspace without retaining it.""" + if self._closed: + raise RuntimeError("prepared artifact is closed") + return await self._owner.extract_prepared_guide(self._binding, extractor) + async def __aenter__(self) -> CommittedArtifactSource: """Enter the bounded provider-I/O lifetime.""" return self.committed_source diff --git a/backend/scripts/run_test_lanes.py b/backend/scripts/run_test_lanes.py index 0f28d2d5..abc23acd 100644 --- a/backend/scripts/run_test_lanes.py +++ b/backend/scripts/run_test_lanes.py @@ -105,6 +105,7 @@ class TestLane: "tests/test_guide_artifacts.py", "tests/test_guide_bindings.py", "tests/test_guide_formats.py", + "tests/test_guide_extraction.py", "tests/test_local_artifact_store.py", "tests/test_s3_artifact_store.py", "tests/test_test_lane_evidence.py", diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 5a6c311e..4d91281c 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -21,7 +21,7 @@ from scripts.run_isolated_tests import LOOPBACK, NAME_RE, ROLE_RE DDL_LOCK_DIRECTORY = Path("/tmp") -EXPECTED_PUBLIC_SCHEMA_SHA256 = "9d6cb126924b2ddd0d805d0bdc21feb55f2827608a26c26e0ecb80072880447a" +EXPECTED_PUBLIC_SCHEMA_SHA256 = "649ad4f05aa677cc72a8f4cffe04291ed8f41fd6f4be0f1bc4aac6809ca96491" PROTECTED_TEST_TABLES = ( "actor_profile_migration_state", "alembic_version", @@ -57,6 +57,10 @@ "guide_source_artifact_ingests", "guide_source_artifact_incidents", "guide_source_artifact_bindings", + "guide_source_extracted_contents", + "guide_source_extraction_attempts", + "guide_source_extraction_retry_budgets", + "guide_source_extraction_usages", "guide_source_format_classifications", "guide_source_snapshot_items", "guide_source_snapshots", diff --git a/backend/tests/fixtures/guide_extraction_probe_worker.py b/backend/tests/fixtures/guide_extraction_probe_worker.py new file mode 100644 index 00000000..a32e90ed --- /dev/null +++ b/backend/tests/fixtures/guide_extraction_probe_worker.py @@ -0,0 +1,78 @@ +"""Test-only child for exercising production extraction sandbox limits.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +import socket +import sys + + +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from app.modules.artifacts.guide_extraction_worker import ( # noqa: E402 + ExtractionFailure, + _install_limits, + _install_seccomp, +) + + +def _denied(probe) -> bool: + try: + probe() + except PermissionError: + return True + except OSError: + return False + return False + + +def main() -> int: + probe = sys.argv[1] if len(sys.argv) == 2 else "" + try: + _install_limits( + cpu_soft_seconds=1 if probe == "__cpu_probe__" else 29, + cpu_hard_seconds=2 if probe == "__cpu_probe__" else 30, + ) + _install_seccomp() + if probe == "__isolation_probe__": + output = json.dumps( + { + "network": _denied(socket.socket), + "outside_read": _denied(lambda: os.open("/etc/passwd", os.O_RDONLY)), + "workspace_write": _denied( + lambda: os.open("probe", os.O_WRONLY | os.O_CREAT, 0o600) + ), + "outside_write": _denied( + lambda: os.open( + "/tmp/workstream-extraction-probe", os.O_WRONLY | os.O_CREAT, 0o600 + ) + ), + "process": _denied(os.fork), + }, + sort_keys=True, + separators=(",", ":"), + ) + elif probe in {"__cpu_probe__", "__wall_probe__"}: + while True: + pass + elif probe == "__memory_probe__": + _ = bytearray(600 * 1024 * 1024) + output = "unexpected" + elif probe == "__abnormal_probe__": + os._exit(3) + else: + raise ExtractionFailure("parser_failure", "invalid_test_probe") + result = {"status": "extracted", "error_code": None, "output": output} + except ExtractionFailure as exc: + result = {"status": exc.status, "error_code": exc.code, "output": None} + except MemoryError: + result = {"status": "limit_exceeded", "error_code": "memory_limit", "output": None} + encoded = json.dumps(result, separators=(",", ":")).encode() + os.write(1, encoded) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/tests/test_alembic.py b/backend/tests/test_alembic.py index de3b06b7..72c8b957 100644 --- a/backend/tests/test_alembic.py +++ b/backend/tests/test_alembic.py @@ -34,6 +34,7 @@ ActionOwner, PermissionId, ) + from app.modules.actors.legacy_classification import ( CLASSIFICATION_FILE_ENV, LegacyActorClassification, @@ -54,6 +55,8 @@ snapshot_existing_service_rows, ) +HEAD_REVISION = "0042_guide_extraction" + pytestmark = pytest.mark.postgres_schema_contract _OBSOLETE_ART_UPLOAD_IDS = tuple( @@ -1640,7 +1643,7 @@ def test_artifact_recovery_schema_and_empty_downgrade( command.downgrade(config, "base") command.upgrade(config, "head") assert asyncio.run(_artifact_recovery_schema(isolated_database_env)) == { - "revision": "0041_project_mutation_evidence", + "revision": HEAD_REVISION, "constraints": { "artifact_recovery_attempt_custody", "artifact_verification_lineage_custody", @@ -1793,10 +1796,7 @@ async def _setup_generations_after_0039(database_url: str) -> list[tuple[str, in try: async with engine.connect() as connection: rows = await connection.execute( - text( - "select id, setup_generation from project_setup_runs " - "order by id" - ) + text("select id, setup_generation from project_setup_runs order by id") ) return sorted((str(row.id), int(row.setup_generation)) for row in rows) finally: @@ -1963,9 +1963,7 @@ def test_0035_project_read_action_evidence_refuses_nonempty_downgrade( RuntimeError, match="cannot downgrade non-empty project-read action evidence" ): command.downgrade(config, "0034_project_role_issue_evidence") - assert asyncio.run(_current_revision(isolated_database_env)) == ( - "0041_project_mutation_evidence" - ) + assert asyncio.run(_current_revision(isolated_database_env)) == (HEAD_REVISION) finally: asyncio.run(_remove_authority_audit_fixture(isolated_database_env, event_id=event_id)) command.downgrade(config, "base") @@ -2014,9 +2012,7 @@ def test_0041_project_mutation_action_evidence_refuses_downgrade( """Committed direct and idempotency-linked evidence must preserve vocabulary.""" config = _alembic_config() definitions = tuple( - item - for item in ACTION_DEFINITIONS - if item.owner in _PROJECT_MUTATION_OWNERS + item for item in ACTION_DEFINITIONS if item.owner in _PROJECT_MUTATION_OWNERS ) assert len(definitions) == 18 event_id = "" @@ -2041,9 +2037,7 @@ def test_0041_project_mutation_action_evidence_refuses_downgrade( match="cannot downgrade non-empty project-mutation action evidence", ): command.downgrade(config, "0040_guide_materialization") - asyncio.run( - _remove_authority_audit_fixture(isolated_database_env, event_id=event_id) - ) + asyncio.run(_remove_authority_audit_fixture(isolated_database_env, event_id=event_id)) event_id = "" definition = definitions[-1] @@ -2066,21 +2060,15 @@ def test_0041_project_mutation_action_evidence_refuses_downgrade( match="cannot downgrade non-empty project-mutation action evidence", ): command.downgrade(config, "0040_guide_materialization") - assert asyncio.run(_current_revision(isolated_database_env)) == ( - "0041_project_mutation_evidence" - ) + assert asyncio.run(_current_revision(isolated_database_env)) == HEAD_REVISION finally: if event_id: asyncio.run( - _remove_authority_audit_fixture( - isolated_database_env, event_id=event_id - ) + _remove_authority_audit_fixture(isolated_database_env, event_id=event_id) ) if linked_event_id: asyncio.run( - _remove_authority_audit_fixture( - isolated_database_env, event_id=linked_event_id - ) + _remove_authority_audit_fixture(isolated_database_env, event_id=linked_event_id) ) asyncio.run( _remove_authority_idempotency_fixture( @@ -2209,9 +2197,7 @@ def test_0036_art_auth_catalogue_refuses_obsolete_evidence( ) record_id = "" command.upgrade(config, "head") - assert asyncio.run(_current_revision(isolated_database_env)) == ( - "0041_project_mutation_evidence" - ) + assert asyncio.run(_current_revision(isolated_database_env)) == (HEAD_REVISION) finally: for event_id in reversed(event_ids): asyncio.run( @@ -2607,7 +2593,7 @@ def test_project_role_migration_constraints_and_immutable_history( command.upgrade(config, "head") result = asyncio.run(_exercise_project_role_migration(isolated_database_env)) assert result == { - "revision": "0041_project_mutation_evidence", + "revision": HEAD_REVISION, "role_count": 3, "invalid_availability": "23514", "duplicate_role": "23505", @@ -2819,7 +2805,7 @@ def test_project_role_downgrade_refuses_each_reserved_evidence_predicate( ): command.downgrade(config, "0030_artifact_verification") assert asyncio.run(_project_role_refusal_state(isolated_database_env))[:3] == ( - "0041_project_mutation_evidence", + HEAD_REVISION, True, True, ) @@ -2846,7 +2832,7 @@ def test_project_role_downgrade_refuses_each_reserved_evidence_predicate( ): command.downgrade(config, "0030_artifact_verification") assert asyncio.run(_project_role_refusal_state(isolated_database_env))[:3] == ( - "0041_project_mutation_evidence", + HEAD_REVISION, True, True, ) @@ -2874,7 +2860,7 @@ def test_outbox_migration_schema_and_downgrade_writer_guard( command.upgrade(config, "head") schema = asyncio.run(_outbox_schema(isolated_database_env)) assert schema == { - "revision": "0041_project_mutation_evidence", + "revision": HEAD_REVISION, "columns": { "aggregate_id", "aggregate_type", @@ -2933,9 +2919,7 @@ def test_outbox_migration_schema_and_downgrade_writer_guard( ) ) assert committed == "refused_after_commit" - assert asyncio.run(_current_revision(isolated_database_env)) == ( - "0041_project_mutation_evidence" - ) + assert asyncio.run(_current_revision(isolated_database_env)) == (HEAD_REVISION) asyncio.run(_remove_outbox_migration_row(isolated_database_env, committed_project_id)) command.downgrade(config, "0028_artifact_admission") assert "outbox_events" not in asyncio.run(_fetch_table_names(isolated_database_env)) diff --git a/backend/tests/test_artifact_architecture.py b/backend/tests/test_artifact_architecture.py index 5c911fac..cf9b5700 100644 --- a/backend/tests/test_artifact_architecture.py +++ b/backend/tests/test_artifact_architecture.py @@ -510,3 +510,43 @@ def test_artifact_repository_does_not_own_actor_persistence() -> None: assert not any(module.startswith("app.modules.actors") for module in imported_modules) assert "actor_profiles" not in path.read_text() assert "actor_identity_links" not in path.read_text() + + +def test_guide_extraction_has_no_provider_agent_auth_or_route_boundary() -> None: + extraction_paths = ( + APP_ROOT / "modules" / "artifacts" / "guide_extraction.py", + APP_ROOT / "modules" / "artifacts" / "guide_extraction_worker.py", + APP_ROOT / "modules" / "artifacts" / "guide_extraction_service.py", + ) + forbidden_prefixes = ( + "app.adapters", + "app.interfaces.artifacts", + "app.modules.authorization", + "app.modules.agents", + "boto", + "celery", + ) + for path in extraction_paths: + imported_modules: set[str] = set() + for node in ast.walk(_tree(path)): + if isinstance(node, ast.ImportFrom): + if node.level: + package = "app.modules.artifacts".split(".") + base = package[: len(package) - node.level + 1] + if node.module is not None: + imported_modules.add(".".join((*base, *node.module.split(".")))) + else: + imported_modules.update( + ".".join((*base, alias.name)) for alias in node.names + ) + elif node.module is not None: + imported_modules.add(node.module) + elif isinstance(node, ast.Import): + imported_modules.update(alias.name for alias in node.names) + assert not any(module.startswith(forbidden_prefixes) for module in imported_modules), ( + path.name + ) + source = path.read_text(encoding="utf-8") + assert "ArtifactStore" not in source + assert "PreparedAuthorizationHandle" not in source + assert "APIRouter" not in source diff --git a/backend/tests/test_artifact_preparation.py b/backend/tests/test_artifact_preparation.py index f57c66dd..35c7e9d7 100644 --- a/backend/tests/test_artifact_preparation.py +++ b/backend/tests/test_artifact_preparation.py @@ -7,6 +7,7 @@ import errno import fcntl import hashlib +import json import multiprocessing import os from pathlib import Path @@ -1238,6 +1239,26 @@ async def test_malformed_ledger_and_layout_fail_closed(tmp_path: Path) -> None: ArtifactScratchManager(root=unsafe_root, limits=preparation_limits()) +@pytest.mark.asyncio +async def test_prior_scratch_ledger_shape_is_normalized_on_reopen(tmp_path: Path) -> None: + """Preserve restart compatibility when workspace custody is introduced.""" + root = tmp_path / "scratch" + initial = ArtifactScratchManager(root=root, limits=preparation_limits()) + initial.close() + ledger_path = root / ".ledger.json" + ledger = json.loads(ledger_path.read_text(encoding="utf-8")) + ledger.pop("workspaces") + ledger_path.write_text(json.dumps(ledger), encoding="utf-8") + + reopened = ArtifactScratchManager(root=root, limits=preparation_limits()) + reservation, descriptor = await reopened.allocate() + os.close(descriptor) + await reopened.release(reservation) + + assert json.loads(ledger_path.read_text(encoding="utf-8"))["workspaces"] == [] + reopened.close() + + @pytest.mark.asyncio async def test_shared_scratch_root_rejects_mismatched_process_limits(tmp_path: Path) -> None: """Fail closed when another process is configured with different root limits.""" diff --git a/backend/tests/test_guide_bindings.py b/backend/tests/test_guide_bindings.py index ec1b2197..3e4e3e03 100644 --- a/backend/tests/test_guide_bindings.py +++ b/backend/tests/test_guide_bindings.py @@ -14,7 +14,8 @@ import pytest from alembic import command from alembic.config import Config -from sqlalchemy import func, select +from sqlalchemy import func, select, text +from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app.core.hashing import canonical_json_hash @@ -29,6 +30,15 @@ GuideSourceBindingService, ) from app.modules.artifacts.guide_formats import GuideFormatDetector, GuideFormatLimits +from app.modules.artifacts.guide_extraction import ( + EXTRACTION_POLICY_VERSION, + GuideExtractionRegistry, +) +from app.modules.artifacts.guide_extraction_service import ( + GuideExtractionError, + GuideExtractionRequest, + GuideExtractionService, +) from app.modules.artifacts.guide_materialization import ( ArtifactMaterializationService, GuideSourceMaterializationError, @@ -43,6 +53,10 @@ GuideSourceArtifactBinding, GuideSourceArtifactIncident, GuideSourceFormatClassification, + GuideSourceExtractedContent, + GuideSourceExtractionAttempt, + GuideSourceExtractionRetryBudget, + GuideSourceExtractionUsage, ) from app.modules.artifacts.preparation import ( HARD_MAXIMUM_ARTIFACT_BYTES, @@ -51,7 +65,10 @@ ArtifactPreparationService, ArtifactScratchManager, ) -from app.modules.artifacts.service import ArtifactStorageNamespaceError, ArtifactStorageNamespaceSpec +from app.modules.artifacts.service import ( + ArtifactStorageNamespaceError, + ArtifactStorageNamespaceSpec, +) from app.modules.artifacts.schemas import ( ArtifactAuthorityDeniedError, GuideSourceBindingAuthorityFacts, @@ -142,6 +159,10 @@ async def open(self, provider_object_ref: str) -> AsyncIterator[bytes]: yield self.payload +async def _byte_stream(payload: bytes) -> AsyncIterator[bytes]: + yield payload + + def _preparation( tmp_path: Path, **limit_changes: Any ) -> tuple[ArtifactPreparationService, ArtifactScratchManager]: @@ -423,6 +444,335 @@ async def _create_binding(factory, ids: dict[str, UUID]) -> UUID: return result.binding_id +@pytest.mark.asyncio +@pytest.mark.postgres_schema_contract +async def test_extraction_publishes_deterministic_content_and_exact_usage( + isolated_database_env: str, tmp_path: Path, migration_lock +) -> None: + payload = b'{"z":2,"a":1}' + digest = "sha256:" + hashlib.sha256(payload).hexdigest() + engine = create_async_engine(isolated_database_env) + factory = async_sessionmaker(engine, expire_on_commit=False) + preparation, manager = _preparation(tmp_path) + prepared = None + try: + async with factory() as session: + ids = await _seed_binding_lineage( + session, + sha256=digest, + byte_count=len(payload), + media_type="application/json", + ) + binding_id = await _create_binding(factory, ids) + classification_id = uuid4() + async with factory() as session, session.begin(): + session.add( + GuideSourceFormatClassification( + id=str(classification_id), + binding_id=str(binding_id), + content_id=str(ids["content"]), + verified_replica_id=str(ids["replica"]), + setup_generation=1, + sha256=digest, + byte_count=len(payload), + media_type="application/json", + detected_format="json", + status="classified", + detector_name="workstream.guide_format", + detector_version="1", + classification_facts={}, + ) + ) + prepared = await preparation.prepare(_byte_stream(payload), media_type="application/json") + request = GuideExtractionRequest( + project_id=ids["project"], + guide_id=ids["guide"], + source_snapshot_id=ids["snapshot"], + source_item_id=ids["item"], + project_setup_run_id=ids["run"], + setup_generation=1, + binding_id=binding_id, + classification_id=classification_id, + ) + result = await GuideExtractionService(factory, GuideExtractionRegistry()).extract_prepared( + request, prepared + ) + assert result.status == "extracted" + assert result.extracted_content_id is not None + assert result.usage_id is not None + async with factory() as session: + content = await session.get( + GuideSourceExtractedContent, str(result.extracted_content_id) + ) + assert content is not None + assert content.canonical_output == '{"a":1,"z":2}' + assert await session.scalar(select(func.count(GuideSourceExtractionAttempt.id))) == 1 + assert await session.scalar(select(func.count(GuideSourceExtractionUsage.id))) == 1 + async with factory() as session: + with pytest.raises(IntegrityError): + async with session.begin(): + await session.execute( + text( + "update guide_source_extraction_usages " + "set attempt_status = 'parser_failure' where id = :usage_id" + ), + {"usage_id": str(result.usage_id)}, + ) + config = Config(str(Path(__file__).resolve().parents[1] / "alembic.ini")) + config.set_main_option( + "script_location", str(Path(__file__).resolve().parents[1] / "alembic") + ) + with ( + migration_lock(), + pytest.raises( + RuntimeError, match="cannot downgrade populated guide extraction evidence" + ), + ): + await asyncio.to_thread( + command.downgrade, + config, + "0041_project_mutation_evidence", + ) + finally: + if prepared is not None: + await prepared.close() + manager.close() + await engine.dispose() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("status", "retry_allowed"), + [ + ("malformed", False), + ("limit_exceeded", False), + ("unsupported", False), + ("parser_failure", True), + ("cancelled", True), + (None, None), + ], +) +async def test_retry_budget_replays_terminal_outcomes_and_only_claims_transient_slot( + isolated_database_env: str, status: str | None, retry_allowed: bool | None +) -> None: + payload = b"guide" + digest = "sha256:" + hashlib.sha256(payload).hexdigest() + engine = create_async_engine(isolated_database_env) + factory = async_sessionmaker(engine, expire_on_commit=False) + try: + async with factory() as session: + ids = await _seed_binding_lineage( + session, + sha256=digest, + byte_count=len(payload), + media_type="text/plain", + ) + binding_id = await _create_binding(factory, ids) + classification_id = uuid4() + attempt_id = uuid4() + async with factory() as session, session.begin(): + session.add( + GuideSourceFormatClassification( + id=str(classification_id), + binding_id=str(binding_id), + content_id=str(ids["content"]), + verified_replica_id=str(ids["replica"]), + setup_generation=1, + sha256=digest, + byte_count=len(payload), + media_type="text/plain", + detected_format="plain_text", + status="classified", + detector_name="workstream.guide_format", + detector_version="1", + classification_facts={}, + ) + ) + await session.flush() + session.add( + GuideSourceExtractionRetryBudget( + binding_id=str(binding_id), + content_id=str(ids["content"]), + classification_id=str(classification_id), + setup_generation=1, + policy_version=EXTRACTION_POLICY_VERSION, + claimed_slots=1, + ) + ) + if status is not None: + session.add( + GuideSourceExtractionAttempt( + id=str(attempt_id), + binding_id=str(binding_id), + content_id=str(ids["content"]), + classification_id=str(classification_id), + setup_generation=1, + detected_format="plain_text", + extractor_name="workstream.plain_text", + extractor_version="1", + policy_version=EXTRACTION_POLICY_VERSION, + attempt_number=1, + status=status, + error_code="test_failure", + bounded_facts={}, + ) + ) + request = GuideExtractionRequest( + project_id=ids["project"], + guide_id=ids["guide"], + source_snapshot_id=ids["snapshot"], + source_item_id=ids["item"], + project_setup_run_id=ids["run"], + setup_generation=1, + binding_id=binding_id, + classification_id=classification_id, + ) + service = GuideExtractionService(factory, GuideExtractionRegistry()) + if retry_allowed is None: + with pytest.raises(GuideExtractionError, match="unavailable"): + await service.claim_materialization_slot(request) + result = None + elif retry_allowed: + concurrent_results = await asyncio.gather( + service.claim_materialization_slot(request), + service.claim_materialization_slot(request), + ) + assert sum(result is None for result in concurrent_results) == 1 + replayed = next(result for result in concurrent_results if result is not None) + assert replayed.attempt_id == attempt_id + assert replayed.status == status + result = None + else: + result = await service.claim_materialization_slot(request) + async with factory() as session: + budget = await session.get(GuideSourceExtractionRetryBudget, str(binding_id)) + assert budget is not None + assert budget.claimed_slots == (2 if retry_allowed is True else 1) + if retry_allowed is None: + return + if retry_allowed: + assert result is None + else: + assert result is not None + assert result.attempt_id == attempt_id + assert result.status == status + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_successful_replay_requires_the_current_extraction_policy( + isolated_database_env: str, +) -> None: + payload = b"guide" + digest = "sha256:" + hashlib.sha256(payload).hexdigest() + output = "old output" + output_digest = "sha256:" + hashlib.sha256(output.encode()).hexdigest() + engine = create_async_engine(isolated_database_env) + factory = async_sessionmaker(engine, expire_on_commit=False) + try: + async with factory() as session: + ids = await _seed_binding_lineage( + session, + sha256=digest, + byte_count=len(payload), + media_type="text/plain", + ) + binding_id = await _create_binding(factory, ids) + classification_id = uuid4() + attempt_id = uuid4() + extracted_content_id = uuid4() + async with factory() as session, session.begin(): + session.add( + GuideSourceFormatClassification( + id=str(classification_id), + binding_id=str(binding_id), + content_id=str(ids["content"]), + verified_replica_id=str(ids["replica"]), + setup_generation=1, + sha256=digest, + byte_count=len(payload), + media_type="text/plain", + detected_format="plain_text", + status="classified", + detector_name="workstream.guide_format", + detector_version="1", + classification_facts={}, + ) + ) + await session.flush() + session.add_all( + [ + GuideSourceExtractionAttempt( + id=str(attempt_id), + binding_id=str(binding_id), + content_id=str(ids["content"]), + classification_id=str(classification_id), + setup_generation=1, + detected_format="plain_text", + extractor_name="workstream.plain_text", + extractor_version="1", + policy_version="guide-extraction-obsolete", + attempt_number=1, + status="extracted", + error_code=None, + bounded_facts={}, + ), + GuideSourceExtractedContent( + id=str(extracted_content_id), + content_id=str(ids["content"]), + detected_format="plain_text", + extractor_name="workstream.plain_text", + extractor_version="1", + policy_version="guide-extraction-obsolete", + source_sha256=digest, + source_byte_count=len(payload), + status="extracted", + output_sha256=output_digest, + canonical_output=output, + omission_facts={}, + ), + ] + ) + await session.flush() + session.add( + GuideSourceExtractionUsage( + id=str(uuid4()), + extracted_content_id=str(extracted_content_id), + extraction_attempt_id=str(attempt_id), + attempt_status="extracted", + binding_id=str(binding_id), + content_id=str(ids["content"]), + source_item_id=str(ids["item"]), + project_setup_run_id=str(ids["run"]), + setup_generation=1, + ) + ) + request = GuideExtractionRequest( + project_id=ids["project"], + guide_id=ids["guide"], + source_snapshot_id=ids["snapshot"], + source_item_id=ids["item"], + project_setup_run_id=ids["run"], + setup_generation=1, + binding_id=binding_id, + classification_id=classification_id, + ) + + result = await GuideExtractionService( + factory, GuideExtractionRegistry() + ).claim_materialization_slot(request) + + assert result is None + async with factory() as session: + budget = await session.get(GuideSourceExtractionRetryBudget, str(binding_id)) + assert budget is not None + assert budget.policy_version == EXTRACTION_POLICY_VERSION + assert budget.claimed_slots == 1 + finally: + await engine.dispose() + + @pytest.mark.asyncio async def test_materialization_denies_before_provider_read( isolated_database_env: str, tmp_path: Path @@ -672,9 +1022,7 @@ async def advance_setup_generation() -> None: guide_id=str(ids["guide"]), guide_version="v1", source_snapshot_id=str(ids["snapshot"]), - source_snapshot_hash=canonical_json_hash( - {"item": str(ids["item"])} - ), + source_snapshot_hash=canonical_json_hash({"item": str(ids["item"])}), setup_generation=2, status="queued", current_step="queued", @@ -747,12 +1095,8 @@ async def test_cross_resource_materialization_denies_before_authority_and_provid ) wrong_value: UUID | int = 2 if changed_field == "setup_generation" else uuid4() - request_binding_id = ( - wrong_value if changed_field == "binding_id" else binding_id - ) - request_changes = ( - {} if changed_field == "binding_id" else {changed_field: wrong_value} - ) + request_binding_id = wrong_value if changed_field == "binding_id" else binding_id + request_changes = {} if changed_field == "binding_id" else {changed_field: wrong_value} with pytest.raises(GuideSourceMaterializationError, match="unavailable"): await service.materialize_guide_source( _materialization_request( @@ -876,6 +1220,7 @@ async def test_materialization_inspection_timeout_cleans_scratch_and_records_inc session, sha256=digest, byte_count=len(payload), media_type="application/pdf" ) binding_id = await _create_binding(factory, ids) + async def deterministic_timeout(*_args: Any, **_kwargs: Any) -> None: raise ArtifactPreparationDeadlineError("artifact preparation deadline exceeded") @@ -1060,9 +1405,12 @@ def test_0039_refuses_populated_binding_downgrade( str(Path(__file__).resolve().parents[1] / "alembic"), ) asyncio.run(_create_populated_binding(isolated_database_env)) - with migration_lock(), pytest.raises( - RuntimeError, - match="cannot downgrade populated guide source artifact bindings", + with ( + migration_lock(), + pytest.raises( + RuntimeError, + match="cannot downgrade populated guide source artifact bindings", + ), ): command.downgrade(config, "0038_guide_source_ingest") @@ -1093,9 +1441,12 @@ def test_0040_refuses_populated_classification_downgrade( str(Path(__file__).resolve().parents[1] / "alembic"), ) asyncio.run(_create_populated_classification(isolated_database_env)) - with migration_lock(), pytest.raises( - RuntimeError, - match="cannot downgrade populated guide materialization evidence", + with ( + migration_lock(), + pytest.raises( + RuntimeError, + match="cannot downgrade populated guide materialization evidence", + ), ): command.downgrade(config, "0039_guide_source_bindings") @@ -1140,9 +1491,12 @@ def test_0040_refuses_incident_only_downgrade( str(Path(__file__).resolve().parents[1] / "alembic"), ) asyncio.run(_create_populated_incident(isolated_database_env)) - with migration_lock(), pytest.raises( - RuntimeError, - match="cannot downgrade populated guide materialization evidence", + with ( + migration_lock(), + pytest.raises( + RuntimeError, + match="cannot downgrade populated guide materialization evidence", + ), ): command.downgrade(config, "0039_guide_source_bindings") @@ -1264,9 +1618,7 @@ async def test_default_live_binding_authority_denies(isolated_database_env: str) authority = _AllowBindingAuthority() with pytest.raises(ArtifactAuthorityDeniedError): async with factory() as session, session.begin(): - await GuideSourceBindingService(session).bind_guide_source( - _request(ids, authority) - ) + await GuideSourceBindingService(session).bind_guide_source(_request(ids, authority)) finally: await engine.dispose() diff --git a/backend/tests/test_guide_extraction.py b/backend/tests/test_guide_extraction.py new file mode 100644 index 00000000..a7457b11 --- /dev/null +++ b/backend/tests/test_guide_extraction.py @@ -0,0 +1,705 @@ +"""Focused proofs for bounded canonical guide extraction.""" + +from __future__ import annotations + +import asyncio +from io import BytesIO +import json +from pathlib import Path +import subprocess +import sys +import threading +from types import SimpleNamespace +from uuid import uuid4 + +import pytest + +import app.modules.artifacts.guide_extraction as extraction_module +import app.modules.artifacts.guide_extraction_worker as worker_module +from app.modules.artifacts.guide_extraction import ( + MAXIMUM_INPUT_BYTES, + MAXIMUM_OUTPUT_BYTES, + BoundGuideExtractor, + GuideExtractionRunner, +) +from app.modules.artifacts.preparation import ( + HARD_MAXIMUM_ARTIFACT_BYTES, + ArtifactPreparationLimits, + ArtifactPreparationService, + ArtifactScratchManager, +) +from app.modules.artifacts.guide_extraction_service import ( + GuideExtractionCoordinator, + GuideExtractionPersistenceResult, + GuideExtractionRequest, +) + + +async def _bytes(payload: bytes): + yield payload + + +@pytest.fixture +def runner() -> GuideExtractionRunner: + return GuideExtractionRunner() + + +def test_worker_controls_and_canonical_extractors_execute_in_parent_coverage( + monkeypatch: pytest.MonkeyPatch, +) -> None: + limits: list[tuple[int, tuple[int, int]]] = [] + monkeypatch.setattr( + worker_module.resource, "setrlimit", lambda key, value: limits.append((key, value)) + ) + worker_module._install_limits() + assert len(limits) >= 5 + + class Function: + def __init__(self, result): + self.result = result + + def __call__(self, *_args): + return self.result + + library = SimpleNamespace( + seccomp_init=Function(1), + seccomp_rule_add=Function(0), + seccomp_syscall_resolve_name=Function(0), + seccomp_load=Function(0), + seccomp_release=Function(None), + ) + monkeypatch.setattr(worker_module.ctypes, "CDLL", lambda *_args, **_kwargs: library) + worker_module._install_seccomp() + library.seccomp_syscall_resolve_name.result = -1 + with pytest.raises(worker_module.ExtractionFailure) as unresolved: + worker_module._install_seccomp() + assert (unresolved.value.status, unresolved.value.code) == ( + "parser_failure", + "isolation_unavailable", + ) + + assert worker_module._extract(b"\xef\xbb\xbfhello\r\n", "plain_text") == "hello\n" + assert worker_module._extract(b'{"z":2,"a":1}', "json") == '{"a":1,"z":2}' + assert worker_module._extract(b"a,b\r\n1,2\r\n", "csv") == '[["a","b"],["1","2"]]' + with pytest.raises(worker_module.ExtractionFailure): + worker_module._extract(b"opaque", "unsupported_binary") + with pytest.raises(worker_module.ExtractionFailure): + worker_module._extract(b'{"a":1,"a":2}', "json") + with pytest.raises(worker_module.ExtractionFailure): + worker_module._extract(b"\xff", "plain_text") + + +@pytest.mark.parametrize( + ("failure", "status", "error_code"), + [ + (worker_module.ExtractionFailure("malformed", "invalid_json"), "malformed", "invalid_json"), + (MemoryError(), "limit_exceeded", "memory_limit"), + (RuntimeError("private"), "parser_failure", "parser_failure"), + ], +) +def test_worker_main_bounds_every_exception_class( + monkeypatch: pytest.MonkeyPatch, + failure: BaseException, + status: str, + error_code: str, +) -> None: + writes: list[bytes] = [] + monkeypatch.setattr(worker_module, "_install_limits", lambda: None) + monkeypatch.setattr(worker_module, "_install_seccomp", lambda: None) + monkeypatch.setattr(worker_module, "_extract", lambda *_args: (_ for _ in ()).throw(failure)) + monkeypatch.setattr(worker_module.sys, "argv", ["worker", "json"]) + monkeypatch.setattr(worker_module.sys, "stdin", SimpleNamespace(buffer=BytesIO(b"{}"))) + + def write(_fd, value): + writes.append(bytes(value)) + return len(value) + + monkeypatch.setattr(worker_module.os, "write", write) + + assert worker_module.main() == 0 + assert json.loads(writes[0]) == {"status": status, "error_code": error_code, "output": None} + + +def test_worker_main_writes_the_complete_result_after_short_writes( + monkeypatch: pytest.MonkeyPatch, +) -> None: + writes: list[bytes] = [] + monkeypatch.setattr(worker_module, "_install_limits", lambda: None) + monkeypatch.setattr(worker_module, "_install_seccomp", lambda: None) + monkeypatch.setattr(worker_module, "_extract", lambda *_args: "complete") + monkeypatch.setattr(worker_module.sys, "argv", ["worker", "plain_text"]) + monkeypatch.setattr(worker_module.sys, "stdin", SimpleNamespace(buffer=BytesIO(b"guide"))) + + def short_write(_fd, value): + chunk = bytes(value[:7]) + writes.append(chunk) + return len(chunk) + + monkeypatch.setattr(worker_module.os, "write", short_write) + + assert worker_module.main() == 0 + assert json.loads(b"".join(writes)) == { + "status": "extracted", + "error_code": None, + "output": "complete", + } + + +@pytest.mark.parametrize( + ("detected_format", "payload", "expected"), + [ + ("plain_text", b"\xef\xbb\xbfhello\r\nworld\r", "hello\nworld\n"), + ("markdown", b"# Guide\r\n\r\nBody\n", "# Guide\n\nBody\n"), + ("json", b'{"z":2,"a":[true,null]}', '{"a":[true,null],"z":2}'), + ("csv", b"name,value\r\na,1\r\n", '[["name","value"],["a","1"]]'), + ], +) +def test_runner_produces_canonical_content( + runner: GuideExtractionRunner, + tmp_path: Path, + detected_format: str, + payload: bytes, + expected: str, +) -> None: + result = runner.extract(BytesIO(payload), detected_format=detected_format, workspace=tmp_path) + + assert result.status == "extracted" + assert result.error_code is None + assert result.canonical_output == expected + assert result.output_sha256 is not None + assert list(tmp_path.iterdir()) == [] + + +@pytest.mark.parametrize( + ("detected_format", "payload", "status", "error_code"), + [ + ("plain_text", b"bad\x00text", "malformed", "invalid_control_character"), + ("plain_text", b"\xff", "malformed", "invalid_encoding"), + ("plain_text", b"a\xef\xbb\xbfb", "malformed", "invalid_encoding"), + ("markdown", b"\xef\xbb\xbf\xef\xbb\xbf", "malformed", "invalid_encoding"), + ("json", b'{"a":1,"a":2}', "malformed", "duplicate_json_key"), + ("json", b'{"a":NaN}', "malformed", "invalid_json_number"), + ("unsupported_binary", b"opaque", "unsupported", "unsupported_format"), + ], +) +def test_runner_returns_bounded_failures( + runner: GuideExtractionRunner, + tmp_path: Path, + detected_format: str, + payload: bytes, + status: str, + error_code: str, +) -> None: + result = runner.extract(BytesIO(payload), detected_format=detected_format, workspace=tmp_path) + + assert (result.status, result.error_code) == (status, error_code) + assert result.canonical_output is None + assert result.output_sha256 is None + + +def test_json_accepts_64_containers_and_rejects_65( + runner: GuideExtractionRunner, tmp_path: Path +) -> None: + accepted = ("[" * 64 + "0" + "]" * 64).encode() + rejected = ("[" * 65 + "0" + "]" * 65).encode() + + assert ( + runner.extract(BytesIO(accepted), detected_format="json", workspace=tmp_path).status + == "extracted" + ) + result = runner.extract(BytesIO(rejected), detected_format="json", workspace=tmp_path) + assert (result.status, result.error_code) == ( + "limit_exceeded", + "json_nesting_limit", + ) + + +def test_csv_cell_boundary_is_exact(runner: GuideExtractionRunner, tmp_path: Path) -> None: + assert ( + runner.extract( + BytesIO(("x" * 32_768).encode()), detected_format="csv", workspace=tmp_path + ).status + == "extracted" + ) + result = runner.extract( + BytesIO(("x" * 32_769).encode()), detected_format="csv", workspace=tmp_path + ) + assert (result.status, result.error_code) == ( + "limit_exceeded", + "csv_cell_size_limit", + ) + module_limit_over = runner.extract( + BytesIO(("x" * 200_000).encode()), detected_format="csv", workspace=tmp_path + ) + assert (module_limit_over.status, module_limit_over.error_code) == ( + "limit_exceeded", + "csv_cell_size_limit", + ) + + +def test_csv_row_and_total_cell_boundaries_are_exact( + runner: GuideExtractionRunner, tmp_path: Path +) -> None: + assert ( + runner.extract(BytesIO(b"\n" * 100_000), detected_format="csv", workspace=tmp_path).status + == "extracted" + ) + row_over = runner.extract(BytesIO(b"\n" * 100_001), detected_format="csv", workspace=tmp_path) + assert (row_over.status, row_over.error_code) == ( + "limit_exceeded", + "csv_row_limit", + ) + million_cells = (("," * 999) + "\n") * 1000 + assert ( + runner.extract( + BytesIO(million_cells.encode()), detected_format="csv", workspace=tmp_path + ).status + == "extracted" + ) + cell_over = runner.extract( + BytesIO((million_cells + ",\n").encode()), + detected_format="csv", + workspace=tmp_path, + ) + assert (cell_over.status, cell_over.error_code) == ( + "limit_exceeded", + "csv_cell_limit", + ) + + +def test_input_and_output_byte_boundaries_are_exact( + runner: GuideExtractionRunner, tmp_path: Path +) -> None: + assert ( + runner.extract( + BytesIO(b"x" * MAXIMUM_OUTPUT_BYTES), + detected_format="plain_text", + workspace=tmp_path, + ).status + == "extracted" + ) + output_over = runner.extract( + BytesIO(b"x" * (MAXIMUM_OUTPUT_BYTES + 1)), + detected_format="plain_text", + workspace=tmp_path, + ) + assert (output_over.status, output_over.error_code) == ( + "limit_exceeded", + "output_limit", + ) + input_over = runner.extract( + BytesIO(b"x" * (MAXIMUM_INPUT_BYTES + 1)), + detected_format="plain_text", + workspace=tmp_path, + ) + assert (input_over.status, input_over.error_code) == ( + "limit_exceeded", + "input_limit", + ) + + +def test_worker_kernel_isolation_denies_network_process_and_filesystem( + tmp_path: Path, +) -> None: + worker = Path(__file__).parent / "fixtures/guide_extraction_probe_worker.py" + completed = subprocess.run( + [sys.executable, "-I", str(worker), "__isolation_probe__"], + input=b"", + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + cwd=tmp_path, + check=True, + timeout=10, + ) + envelope = json.loads(completed.stdout) + assert envelope["status"] == "extracted" + assert json.loads(envelope["output"]) == { + "network": True, + "outside_read": True, + "outside_write": True, + "process": True, + "workspace_write": True, + } + assert list(tmp_path.iterdir()) == [] + + +@pytest.mark.parametrize( + ("probe", "status", "error_code"), + [ + ("__cpu_probe__", "limit_exceeded", "cpu_time_limit"), + ("__memory_probe__", "limit_exceeded", "memory_limit"), + ("__abnormal_probe__", "parser_failure", "executor_lost"), + ], +) +def test_real_executor_resource_and_loss_outcomes( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + probe: str, + status: str, + error_code: str, +) -> None: + monkeypatch.setattr(extraction_module, "_SUPPORTED", frozenset({probe})) + runner = GuideExtractionRunner() + runner._worker_path = Path(__file__).parent / "fixtures/guide_extraction_probe_worker.py" + result = runner.extract(BytesIO(b""), detected_format=probe, workspace=tmp_path) + assert (result.status, result.error_code) == (status, error_code) + + +def test_wall_timeout_kills_and_reaps_executor( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(extraction_module, "_SUPPORTED", frozenset({"__wall_probe__"})) + monkeypatch.setattr(extraction_module, "WALL_TIMEOUT_SECONDS", 0.05) + runner = GuideExtractionRunner() + runner._worker_path = Path(__file__).parent / "fixtures/guide_extraction_probe_worker.py" + result = runner.extract(BytesIO(b""), detected_format="__wall_probe__", workspace=tmp_path) + assert (result.status, result.error_code) == ( + "limit_exceeded", + "wall_time_limit", + ) + + +def test_runner_fails_closed_when_worker_cannot_start( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + def unavailable(*_args, **_kwargs): + raise OSError("unavailable") + + monkeypatch.setattr(subprocess, "Popen", unavailable) + runner = GuideExtractionRunner() + result = runner.extract(BytesIO(b"hello"), detected_format="plain_text", workspace=tmp_path) + + assert result.status == "parser_failure" + assert result.error_code == "executor_unavailable" + assert result.canonical_output is None + + +def test_runner_launch_is_secret_free_and_process_isolated( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + observed: dict[str, object] = {} + + class CompletedProcess: + returncode = 0 + pid = 123 + + def __init__(self, *_args, **kwargs) -> None: + observed.update(kwargs) + + def communicate(self, _payload, timeout): + observed["timeout"] = timeout + return b'{"status":"extracted","error_code":null,"output":"ok"}', b"" + + monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "must-not-leak") + monkeypatch.setenv("HTTPS_PROXY", "must-not-leak") + monkeypatch.setattr(subprocess, "Popen", CompletedProcess) + result = GuideExtractionRunner().extract( + BytesIO(b"ok"), detected_format="plain_text", workspace=tmp_path + ) + + assert result.status == "extracted" + assert observed["shell"] is False + assert observed["close_fds"] is True + assert observed["start_new_session"] is True + assert observed["cwd"] == tmp_path + environment = observed["env"] + assert isinstance(environment, dict) + assert set(environment) == {"LANG", "LC_ALL", "PATH"} + + +@pytest.mark.parametrize( + "worker_result", + [ + {"status": "extracted", "error_code": "conflict", "output": "ok"}, + {"status": "extracted", "error_code": None, "output": None}, + {"status": "extracted", "error_code": None, "output": 7}, + {"status": "malformed", "error_code": None, "output": None}, + {"status": "malformed", "error_code": 7, "output": None}, + {"status": "malformed", "error_code": "x" * 81, "output": None}, + {"status": "malformed", "error_code": "invalid", "output": "unexpected"}, + {"status": "malformed", "error_code": "invalid", "output": 7}, + ], +) +def test_runner_rejects_invalid_worker_result_shapes( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + worker_result: dict[str, object], +) -> None: + class CompletedProcess: + returncode = 0 + pid = 123 + + def __init__(self, *_args, **_kwargs) -> None: + pass + + def communicate(self, _payload, timeout): + del timeout + return json.dumps(worker_result).encode(), b"" + + monkeypatch.setattr(subprocess, "Popen", CompletedProcess) + + result = GuideExtractionRunner().extract( + BytesIO(b"guide"), detected_format="plain_text", workspace=tmp_path + ) + + assert (result.status, result.error_code, result.canonical_output) == ( + "parser_failure", + "invalid_executor_output", + None, + ) + + +@pytest.mark.asyncio +async def test_prepared_extraction_uses_and_cleans_scratch_workspace(tmp_path: Path) -> None: + manager = ArtifactScratchManager( + root=tmp_path / "scratch", + limits=ArtifactPreparationLimits( + aggregate_reserved_bytes=HARD_MAXIMUM_ARTIFACT_BYTES, + maximum_files=1, + maximum_concurrency=1, + minimum_free_bytes=0, + reservation_ttl_seconds=30, + total_deadline_seconds=10, + cleanup_margin_seconds=1, + stream_buffer_bytes=1024, + maximum_source_bytes=1024, + ), + ) + prepared = await ArtifactPreparationService(manager).prepare( + _bytes(b'{"b":2,"a":1}'), media_type="application/json" + ) + try: + result = await prepared.extract_guide( + BoundGuideExtractor(runner=GuideExtractionRunner(), detected_format="json") + ) + assert result.canonical_output == '{"a":1,"b":2}' + assert list((tmp_path / "scratch" / "workspaces").iterdir()) == [] + finally: + await prepared.close() + manager.close() + + +@pytest.mark.asyncio +async def test_cancelled_extraction_finishes_child_cleanup_before_returning( + tmp_path: Path, +) -> None: + started = threading.Event() + release = threading.Event() + + class BlockingRunner: + def extract(self, _reader, *, detected_format: str, workspace: Path): + del detected_format + assert workspace.is_dir() + started.set() + assert release.wait(timeout=5) + return GuideExtractionRunner._result("plain_text", "extracted", None, "ok") + + manager = ArtifactScratchManager( + root=tmp_path / "scratch", + limits=ArtifactPreparationLimits( + aggregate_reserved_bytes=HARD_MAXIMUM_ARTIFACT_BYTES, + maximum_files=1, + maximum_concurrency=1, + minimum_free_bytes=0, + reservation_ttl_seconds=30, + total_deadline_seconds=10, + cleanup_margin_seconds=1, + stream_buffer_bytes=1024, + maximum_source_bytes=1024, + ), + ) + prepared = await ArtifactPreparationService(manager).prepare( + _bytes(b"text"), media_type="text/plain" + ) + task = asyncio.create_task( + prepared.extract_guide( + BoundGuideExtractor( + runner=BlockingRunner(), # type: ignore[arg-type] + detected_format="plain_text", + ) + ) + ) + try: + assert await asyncio.to_thread(started.wait, 5) + task.cancel() + release.set() + with pytest.raises(asyncio.CancelledError): + await task + assert list((tmp_path / "scratch" / "workspaces").iterdir()) == [] + finally: + release.set() + await prepared.close() + manager.close() + + +@pytest.mark.asyncio +async def test_failed_extractor_residue_is_removed_before_workspace_release( + tmp_path: Path, +) -> None: + class DirtyRunner: + def extract(self, _reader, *, detected_format: str, workspace: Path): + del detected_format + nested = workspace / "nested" + nested.mkdir() + (nested / "partial").write_bytes(b"untrusted") + raise RuntimeError("parser failed") + + manager = ArtifactScratchManager( + root=tmp_path / "scratch", + limits=ArtifactPreparationLimits( + aggregate_reserved_bytes=HARD_MAXIMUM_ARTIFACT_BYTES, + maximum_files=2, + maximum_concurrency=1, + minimum_free_bytes=0, + reservation_ttl_seconds=30, + total_deadline_seconds=10, + cleanup_margin_seconds=1, + stream_buffer_bytes=1024, + maximum_source_bytes=1024, + ), + ) + prepared = await ArtifactPreparationService(manager).prepare( + _bytes(b"text"), media_type="text/plain" + ) + try: + with pytest.raises(RuntimeError, match="parser failed"): + await prepared.extract_guide( + BoundGuideExtractor( + runner=DirtyRunner(), # type: ignore[arg-type] + detected_format="plain_text", + ) + ) + assert list((tmp_path / "scratch" / "workspaces").iterdir()) == [] + assert manager.pending_cleanup_count == 0 + finally: + await prepared.close() + manager.close() + + +@pytest.mark.asyncio +async def test_stale_cleanup_reaps_crashed_extraction_workspace( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + manager = ArtifactScratchManager( + root=tmp_path / "scratch", + limits=ArtifactPreparationLimits( + aggregate_reserved_bytes=HARD_MAXIMUM_ARTIFACT_BYTES, + maximum_files=2, + maximum_concurrency=1, + minimum_free_bytes=0, + reservation_ttl_seconds=30, + total_deadline_seconds=10, + cleanup_margin_seconds=1, + stream_buffer_bytes=1024, + maximum_source_bytes=1024, + ), + ) + workspace_name = "extract_" + "a" * 32 + manager._reserve_workspace_sync(workspace_name) + workspace = tmp_path / "scratch" / "workspaces" / workspace_name + workspace.mkdir(mode=0o700) + (workspace / "partial").write_bytes(b"untrusted") + monkeypatch.setattr(manager, "_owner_process_matches", lambda *_: False) + + cleaned = await manager.cleanup_stale(now_unix_ns=2**63) + + assert cleaned == 1 + assert not workspace.exists() + assert manager._read_ledger()["workspaces"] == [] + manager.close() + + +@pytest.mark.asyncio +async def test_executor_failure_retries_once_with_fresh_authority_and_materialization() -> None: + request = GuideExtractionRequest( + project_id=uuid4(), + guide_id=uuid4(), + source_snapshot_id=uuid4(), + source_item_id=uuid4(), + project_setup_run_id=uuid4(), + setup_generation=1, + binding_id=uuid4(), + classification_id=uuid4(), + ) + prepared_sources: list[SimpleNamespace] = [] + + class Materializer: + async def materialize_with_fresh_authority(self, actual_request): + assert actual_request is request + + async def close() -> None: + source.closed = True + + source = SimpleNamespace(closed=False, close=close) + prepared_sources.append(source) + return source + + class Service: + calls = 0 + + async def claim_materialization_slot(self, actual_request): + assert actual_request is request + return None + + async def extract_prepared(self, actual_request, prepared): + assert actual_request is request + assert prepared is prepared_sources[-1] + self.calls += 1 + status = "parser_failure" if self.calls == 1 else "extracted" + return GuideExtractionPersistenceResult( + attempt_id=uuid4(), + status=status, + error_code="executor_lost" if status == "parser_failure" else None, + extracted_content_id=uuid4() if status == "extracted" else None, + usage_id=uuid4() if status == "extracted" else None, + replayed=False, + ) + + service = Service() + result = await GuideExtractionCoordinator( # type: ignore[arg-type] + service, + Materializer(), # type: ignore[arg-type] + ).extract(request) + + assert result.status == "extracted" + assert service.calls == 2 + assert len(prepared_sources) == 2 + assert all(source.closed for source in prepared_sources) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", ["malformed", "limit_exceeded", "unsupported"]) +async def test_terminal_extraction_replay_does_not_materialize_again(status: str) -> None: + request = GuideExtractionRequest( + project_id=uuid4(), + guide_id=uuid4(), + source_snapshot_id=uuid4(), + source_item_id=uuid4(), + project_setup_run_id=uuid4(), + setup_generation=1, + binding_id=uuid4(), + classification_id=uuid4(), + ) + terminal = GuideExtractionPersistenceResult( + attempt_id=uuid4(), + status=status, + error_code="terminal", + extracted_content_id=None, + usage_id=None, + replayed=False, + ) + + class Service: + async def claim_materialization_slot(self, actual_request): + assert actual_request is request + return terminal + + class Materializer: + async def materialize_with_fresh_authority(self, _request): + raise AssertionError("terminal extraction must not materialize again") + + result = await GuideExtractionCoordinator( # type: ignore[arg-type] + Service(), + Materializer(), # type: ignore[arg-type] + ).extract(request) + assert result is terminal diff --git a/docs/spec_artifact_storage_service.md b/docs/spec_artifact_storage_service.md index 782205e8..8000f7c6 100644 --- a/docs/spec_artifact_storage_service.md +++ b/docs/spec_artifact_storage_service.md @@ -1241,16 +1241,36 @@ structural facts; they contain no source filenames, provider references, raw document text, or parser exception strings. `artifact.guide_source.read` remains unavailable until AUTH-04B installs the fixed guide-reader adapter. -Typed extractors run in a bounded no-network isolation boundary. One immutable +Typed extractors run in a bounded no-network isolation boundary. Immutable +attempt evidence carries bounded status/error facts. A separate successful content-derived record identifies original content, format, extractor/version, -policy version, output digest, omission facts, status, and error code; a -separate immutable usage record binds it to the exact binding, item, setup run, -and generation. v0.1 persists these bounded records in PostgreSQL and does not +policy version, canonical output, output digest, and omission facts without an +error code; immutable usage binds that success to the exact binding, item, +setup run, and generation. v0.1 persists these bounded records in PostgreSQL and does not use guide-read authority to create a provider object. Only successful current- generation extraction reaches the agent, delimited as untrusted source data with no tool, secret, provider, or instruction authority. Unsupported, ambiguous, malformed, stale, or failed required material stops setup internally. +For 03B3A the isolation contract is descriptor-only parsing after trusted +imports, enforced by a default-deny Linux libseccomp profile with an explicit +syscall allowlist plus fixed resource limits: 32 MiB input, 4 MiB +canonical output/file size, 30 CPU seconds, 60 wall seconds, 512 MiB address +space, 32 descriptors, no children, no core dumps, and JSON depth 64. The child +receives a minimal secret-free environment, a scratch-owned working directory, +and no provider, object, or customer path; parsing becomes descriptor-only +after seccomp is installed. +An exact-lineage durable budget permits the initial materialization and at most +one fresh-authority retry for `parser_failure` or current-lineage cancellation; +deterministic terminal outcomes replay without another provider read. +Text/Markdown are normalized UTF-8 text; JSON is duplicate-free, finite, +sorted compact JSON; CSV is strict fixed-dialect row data serialized as compact +JSON. Failed outcomes may retain bounded status evidence but never canonical +output, successful usage, or a sufficiency report. +Numeric resource-limit termination maps to `limit_exceeded`; prohibited process +creation, unavailable isolation, or other abnormal child termination maps to +`parser_failure`. + Immediately before committing a sufficiency report, the setup transaction must lock, reload, and revalidate the exact project, draft guide, source snapshot, setup run and generation, guide-source binding, observed content-integrity @@ -1357,6 +1377,12 @@ Implementation is a clean cut: adds no Operator route or generic artifact-read API; future authorized operational visibility must project these bounded records without exposing provider references or document content. +- migration `0042_guide_extraction` adds bounded extraction attempts, + successful canonical content, exact usage provenance, and a durable + exact-lineage two-slot materialization budget. Composite constraints prevent + cross-binding/classification/content/generation usage and require usage to + reference an `extracted` attempt. Downgrade locks the four tables and refuses + while any extraction or retry-budget evidence exists. Every migration proves fresh upgrade, prior-head upgrade, populated-state preservation or explicit refusal, empty downgrade/re-upgrade, and no artifact