From 0910768165734b814cc30f47094f392bd421e75f Mon Sep 17 00:00:00 2001 From: Dylan Roy <249851642+d6g8k5htny-coder@users.noreply.github.com> Date: Thu, 8 Oct 2026 15:59:25 -0700 Subject: [PATCH 1/5] Specify private Artifact Index projection with external privacy controls --- .../public-artifact-index-bindings.json | 1 + .../public-artifact-index-policy.json | 18 + .../2026-10-08-architecture-interface.md | 99 ++ .../2026-10-08-artifact-index-contract.md | 47 + tests/test_architecture_artifact_index.py | 945 ++++++++++++++++++ 5 files changed, 1110 insertions(+) create mode 100644 architecture/public-artifact-index-bindings.json create mode 100644 architecture/public-artifact-index-policy.json create mode 100644 docs/plans/2026-10-08-architecture-interface.md create mode 100644 docs/plans/2026-10-08-artifact-index-contract.md create mode 100644 tests/test_architecture_artifact_index.py diff --git a/architecture/public-artifact-index-bindings.json b/architecture/public-artifact-index-bindings.json new file mode 100644 index 000000000..fe51488c7 --- /dev/null +++ b/architecture/public-artifact-index-bindings.json @@ -0,0 +1 @@ +[] diff --git a/architecture/public-artifact-index-policy.json b/architecture/public-artifact-index-policy.json new file mode 100644 index 000000000..3dcf1909c --- /dev/null +++ b/architecture/public-artifact-index-policy.json @@ -0,0 +1,18 @@ +{ + "schema_version": 1, + "artifact_ids": [ + "C151-MAIN285-LOG-BINDINGS-20261007-v1", + "C152-MAIN286-FACTUAL-REVIEW-20261007-v1", + "C215-GRAPH-ALIAS-AUDIT-20261008-v1" + ], + "columns": [ + "Artifact ID", + "Title", + "Org", + "Class", + "Status", + "Authority / canonical impact", + "Dependencies or supersession", + "Modified UTC" + ] +} diff --git a/docs/plans/2026-10-08-architecture-interface.md b/docs/plans/2026-10-08-architecture-interface.md new file mode 100644 index 000000000..aad7b7d2f --- /dev/null +++ b/docs/plans/2026-10-08-architecture-interface.md @@ -0,0 +1,99 @@ +# Architecture Interface and Canvas Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Publish an additive Astro reader of pinned graph and explicitly public evidence inputs, with source-bound cubic and finite elder teaching views and a verifiable immutable build identity. + +**Architecture:** A trusted capture/projection stage produces a public-only input bundle. A hosted offline static build consumes that bundle once; browser tests and publication consume the same assembled output artifact. Original graph and native evidence records retain their distinct scope and authority; the reader is a derived view. + +**Tech Stack:** Stdlib Python capture/projection controls; exact Astro/Node/npm candidates and lockfile after hosted verification; vanilla ES modules and Canvas; existing hosted Chrome tooling; GitHub Pages. + +**Spec:** [Research architecture design, sections 1/4 and 7](../superpowers/specs/2026-10-08-research-architecture-design.md); [global implementation plan, Tasks 3 and 7](../superpowers/plans/2026-10-08-research-architecture-plan.md); [evidence adapter contract](2026-10-08-architecture-evidence-adapter.md). + +## Global Constraints + +- Scientific effect is always `NONE`; scientific status authority is exact boolean `false`. Build success, recorded review, source-relative consistency and teaching output do not promote science. +- Preserve source, review, kernel, computation and alignment as five separate axes. Preserve original classifications, records, qualifiers and unknown/stale/unavailable states. The stdin evidence adapter's custody remains `unknown`; the reader cannot convert its `current` applicability into authenticated kernel verification or declaration existence. +- The public graph may display all 49 currently pinned nodes without any guessed Artifact Index or Lean join. Artifact/node/teaching joins require exact reviewed IDs and roles; absence remains unresolved. +- Keep the complete private Artifact Index/source capture outside the public repository and publication bundle. Publish only explicitly allowed rows and columns. No private original packet, raw table, credential or unlisted field is copied into generated assets, logs or failure artifacts. +- Root retained the actual R17 Artifact Index capture and froze its observed metadata, column names, record schema, size limits and public projection boundary in the first-unit contract. Synthetic TEST fixtures exercise that boundary; they do not substitute for the retained real capture or actual public projection delivery. +- Root owns the isolated interface branch and the precise first-unit and runtime-preparation paths recorded in the pickup and amendment. Tests precede implementation, with hosted RED required before product authoring. The earlier source-only checkpoint is superseded operationally by this scoped execution claim. Standing owner authorization creates no additional user-permission checkpoint. +- No project execution or installation on the laptop. Preparation, RED/GREEN, two clean builds and browser execution occur in disposable hosted environments; network is denied during the actual build/test stage. +- PR #317 owns shared site/brand styles, release-key changes and the existing browser harness/tests. Existing `docs/site` generators and release inventory stay owner-controlled until reconciliation after that PR lands. Root currently owns the architecture producer/required-check workflows. +- New public routes use `/main/reader/`. Preserve the complete existing `/docs` publication namespace and `/main/site/*.html` routes. Pages Actions migration and navigation integration require explicit technical coordination with those owners, not renewed owner permission. +- True 500-token retrieval, lease services, routing and broader execution migration remain separate increments. All eight original requirements remain active until their actual end-to-end completion evidence exists. + +## Review Focus + +- A public-column whitelist applied after serializing the full private capture still leaks data; Task 1 must prove private canaries never enter any output or retained failure artifact. +- Matching caller-provided digests do not establish archive custody; Task 1 must bind original ZIP bytes to the native producer outputs and reject contradictory/stale archive identities. +- Generated graph schema differs from the old source viewer schema; Task 2 must consume the exporter directly without inventing `terminal_classifications` or other missing semantics. +- A disjoint-set parent is not the mathematical elder, and neutral events are not deaths; Task 3 must independently verify those distinctions and interval endpoints. +- A green build rebuilt during deployment can publish different inputs; Task 4 must deploy and read back the exact tested output artifact while preserving legacy URLs. + +--- + +## Task 1: Private capture, public projection and native graph custody + +**Create:** `architecture/artifact_index.py`, `tools/artifact_index_export.py`, `tools/interface_artifact_readback.py`, `tools/interface_build_inputs.py`, `tests/test_architecture_artifact_index.py`, `tests/test_interface_artifact_readback.py`, `tests/test_interface_build_inputs.py`. + +**Interfaces:** `project_artifact_index(packet, expected) -> public_projection` follows the [frozen first-unit contract](2026-10-08-artifact-index-contract.md). The packet carries the native capture, explicit public policy, exact bindings and graph-node inventory; `expected` carries five independently supplied identities. The first CLI reads one packet from stdin and writes one admitted canonical public JSON object to stdout. It opens no caller-selected files, performs no network access, and uses the exact opaque refusal line `ARTIFACT_INDEX_REFUSED` on stderr. Policy selects explicit stable artifact IDs and the fixed eight public columns; bindings contain exactly `{node_id, artifact_id, roles}`. Roles are explicit source/review/kernel/computation/alignment/teaching-source associations, never acceptance. A later separately tested trusted collector reads the retained outside-repository private capture and writes the admitted stdout projection to `public-index.json`; its file and archive custody is outside this first CLI unit. + +`interface_artifact_readback.py` consumes a previously downloaded original ZIP plus expected native repository/checked-commit/run/attempt/artifact-ID/name/archive-digest/host-receipt-digest arguments. Expectations come from the current trusted producer job, not ZIP members. It produces a bounded regular-file staging directory and an original-byte readback receipt. `interface_build_inputs.py` consumes that staged graph, public projection, allowed evidence/teaching/brand inputs and renderer identities and emits the public bundle below. + +- [ ] Root freezes the real capture's metadata and parser ABI, exact public row/column allowlist, explicit ID joins, positive public fixture and refusal fixture. Preserve the full private capture in its existing private evidence location; record only its identity/digest publicly when allowed. +- [ ] Write external CLI controls: valid public projection; absent mapping leaves unresolved; unknown/duplicate IDs and contradictory joins refuse; unlisted columns/rows and private canaries remain absent from all public outputs and failure artifacts; mutable references stay labeled; raw capture/policy/binding hash changes refuse. Observe hosted RED before implementation. +- [ ] Write custody controls: native archive identity/digest mismatch; wrong attempt/checked commit/receipt; duplicate ZIP members; absolute/traversing paths; symlink/special members; expanded-size limits; changed graph member; failed or missing producer evidence. Preserve original failure evidence without producing an accepted input bundle. +- [ ] Implement format-specific projection and archive verification without fetching caller-selected paths/URLs or executing captured code. Validate generated graph's exact seven-key schema and source identity, and match its raw hash to the original host receipt. Keep its dated Math source commit separate from the current executing main commit. +- [ ] Run hosted normal/optimized Python controls, inspect public/failure artifacts for the private canaries, and obtain nonauthor source/custody review. Freeze this unit through root before the renderer consumes it. + +**Public bundle boundary:** `inputs.json` has schema version, exact non-scientific fields, checked repository/commit, graph producer binding and raw graph/receipt identities, public index capture/projection/policy/binding identities, public evidence projection identity, teaching/brand file identities, renderer source/lockfile/runtime identities. Each member is an explicitly listed regular file with byte count and SHA-256. The exact public-index record fields are frozen in the first-unit contract before test/code authoring. The full private capture and evidence adapter `original_packet` are never bundle members. + +## Task 2: Offline Astro proof pages and immutable build receipt + +**Create:** `interface/package.json`, `interface/package-lock.json`, `interface/astro.config.mjs`, `interface/src/lib/load-inputs.mjs`, `interface/src/layouts/Reader.astro`, `interface/src/components/EvidenceDimensions.astro`, `interface/src/components/DependencyTable.astro`, `interface/src/pages/index.astro`, `interface/src/pages/nodes/[id].astro`, `interface/tests/inputs.test.mjs`, `interface/tests/pages.test.mjs`, `tools/interface_build_receipt.py`, `tests/test_interface_build_receipt.py`, `.github/workflows/research-interface.yml`. + +**Interfaces:** `loadPublicInputs(inputDirectory, expectedManifestSha256) -> admittedInputs` reads only the already admitted public bundle; it performs no network fetch. `nodePageData(nodeId, admittedInputs) -> pageData` retains exact original node/edge data and explicitly joined evidence, including unresolved joins. `interface_build_receipt.py` binds the identity core and complete sorted output-file inventory. + +- [ ] In a root-owned trusted hosted preparation stage, verify candidates Astro `7.3.8`, Node `24.21.0` and npm `11.19.0`; retain exact package/runtime bytes and identities, install from a committed integrity-bearing lockfile, and test compatibility. These versions were observed during reconnaissance and are candidates until preparation verifies them. Use no new theme, external fonts, client framework or graph library in this bounded unit. +- [ ] Write controls for missing/tampered bundle members, duplicate/nonfinite/incorrectly typed JSON, mixed build inputs, escaped IDs/URLs, unknown roles, all graph nodes without joins, unavailable evidence and native metadata preservation. Observe hosted RED; a setup/import failure or zero discovery is not RED evidence. +- [ ] Implement static routes with `site: 'https://d6g8k5htny-coder.github.io'`, `base: '/main/reader'`, `output: 'static'`, and `build.inlineStylesheets: 'never'`. Keep scripts/styles as local external assets under the existing self-only CSP. Reuse an owner-approved, byte-pinned brand capture without modifying shared brand/release files. +- [ ] Render an accessible dependency table/view and one page per exact graph ID. Display original classification as a recorded source label; render all five evidence axes independently with custody and as-of identity. Escape retained text; never execute embedded HTML/code or synthesize formal mappings from titles/classifications. +- [ ] Define `build_id = SHA256(canonical identity_core)` binding checked commit, graph/projection/evidence/teaching/brand input hashes, source tree, lockfile and runtime identity. Exclude the build ID itself and fresh wall-clock timestamps. Emit `build-receipt.json` with that identity and sorted output path/byte/SHA inventory; external archive identity binds the receipt itself. Every page links to its build receipt. +- [ ] From two clean hosted work directories, with network denied and identical admitted inputs, run `npm ci --offline` then `npm run build` using the prepared dependency cache. Compare complete output membership and bytes; require identical build IDs and no private canary. Retain both manifests/logs and the single tested output archive. Review source and byte identities before freezing. + +## Task 3: Source-bound cubic and finite elder Canvas views + +**Create:** `interface/src/pages/teach/cubic.astro`, `interface/src/pages/teach/elder.astro`, `interface/src/scripts/cubic-model.mjs`, `interface/src/scripts/elder-model.mjs`, `interface/src/scripts/canvas-reader.mjs`, `interface/tests/cubic.test.mjs`, `interface/tests/elder.test.mjs`, `interface/tests/reader-controls.test.mjs`. Public teaching fixtures enter only through Task 1's explicit source policy. + +**Interfaces:** `cubicModel({r, k, b}) -> scales` implements the pinned educational `gap = k*r^3` definition. `finiteElderTrace({values, n, scale}) -> trace` returns finite-filtration threshold/component/elder/merge/neutral events and positive/zero/essential intervals. `renderTeachingView(host, admittedTeachingData)` connects the same data to Canvas and equivalent DOM tables/controls. Source-node links exist only through explicit admitted mappings. + +- [ ] Pin source bytes for the current cubic definition and `experiments/periodic_h0/{EXACT_H0.md,exact_h0.py,test_exact_h0.py}`. Keep their finite/educational scope and source dates visible. No synthetic source identity, continuum sample, Lean declaration or source-to-component certificate is invented. +- [ ] Write independent controls for positive cubic scales, halving `r` dividing gap by eight, normalization and nonrepresentable/invalid values. Write finite elder controls for `[5,1,4,2,3,0]` repeated six times: positive pairs `[[3,2],[4,1]]`, essential birth `5`, zero count `33`; plateau ties prefer the smaller row-major vertex ID. Include periodic seam and no-diagonal controls and a BFS connectivity oracle distinct from the sweep. +- [ ] Pin superlevel `(d/S,b/S]`, lifetime `(b-d)/S`, zero pairs excluded from the positive diagram, essential class separate, and no longest finite bar removal. Test that disjoint-set parents cannot overrule mathematical elders and same-class edges create neutral events rather than deaths. Missing mappings/certificates render unresolved, never a positive source claim. Observe hosted RED before implementing. +- [ ] Implement exact model/event data separately from approximate drawing coordinates. Label the cubic arithmetic as teaching; label the elder scene as a specified finite periodic filtration. Keep immediate mathematical merge relation, global class equality, essential bars and finite conventions distinct from data-structure pointers. +- [ ] Implement keyboard-operable ordinary DOM controls, event/value tables and source disclosures usable without Canvas. Respect reduced motion; avoid mandatory animation. Source absence/tamper leaves readable unavailable state with no inferred result. Run model and DOM controls hosted and obtain mathematical-scope/source review. + +## Task 4: Browser verification, existing namespace preservation and exact-artifact deployment + +**Create:** `tools/interface_browser_check.py`, `tests/test_interface_browser_contract.py`; extend only the new `research-interface.yml` under root's claim. **Later coordinated paths:** assembled Pages workflow/configuration and existing navigation/docs; exact modifications follow PR #317 integration and a refreshed owner claim. + +**Interfaces:** The browser harness serves Task 2/3's assembled artifact, not a fresh build. Publication consumes the same artifact ID/digest and build receipt accepted by tests. The assembled Pages root includes the complete verified existing `/docs` namespace plus the additive `reader/` subtree. + +- [ ] Test the actual built pages in hosted Chrome at desktop/mobile widths, light/dark schemes, keyboard-only interaction, reduced motion, Canvas unavailable, source/index refusal and no-JavaScript DOM fallback. Check graph-node routes, five-axis qualifiers, focus/overflow and source/build identities; retain exact runtime/report/screenshots and failures. Existing PR #317 browser findings remain separate work to reconcile. +- [ ] Freeze a preservation manifest for every legacy `/docs` file/route from the actual integrated source. Assemble that namespace unchanged with the new reader. Detect output collisions and missing legacy routes before publication; exercise existing reader links and release checks under their owner-controlled hosted lane. +- [ ] After PR #317 and current producer scopes are reconciled, root claims the precise navigation and Pages migration paths. Switch to GitHub Actions publication only through available authorized administration capability and retained readback. This is technical integration work, not another owner approval queue. +- [ ] Use reviewed, commit-pinned Pages upload/deploy actions and scoped publication credentials only in the trusted host stage. Deploy the verified assembled artifact; do not rerun Astro, regenerate graph/index data or fetch mutable cloud inputs during deployment. Keep a recoverable prior artifact and source/configuration record. +- [ ] Read back live build receipt, representative graph/source/teaching pages and asset bytes; verify their hashes/build ID and actual reader interactions. Retain native deployment identity and exact source-bound receipts in existing governance surfaces. Mark this increment deployed only after that evidence exists. + +## Handoff and Completion Boundary + +The initial source-only checkpoint is preserved in the root checkpoint below; its operational wait is superseded by root's retained capture, frozen contract and scoped execution claim. First-unit tests are now authored and under nonauthor review; hosted RED and product implementation remain pending. The build does not authenticate mathematical records, establish unknown declarations, repair missing mappings or complete semantic retrieval. Goals 1 and 7 require tested deployed pages/Canvas; the every-PR producer/SSG join and all remaining original goals retain their explicit incomplete scope until separately verified. + +Primary references: [Astro Pages deployment](https://docs.astro.build/en/guides/deploy/github/), [Astro configuration](https://docs.astro.build/en/reference/configuration-reference/), [Astro package metadata](https://registry.npmjs.org/astro/latest), [Node distribution metadata](https://nodejs.org/dist/index.json), [GitHub publishing-source configuration](https://docs.github.com/en/pages/getting-started-with-github-pages/configuring-a-publishing-source-for-your-github-pages-site). Package aliases/docs are mutable research inputs; exact preparation receipts and the committed lockfile govern execution. + +## Root capture and first-unit checkpoint + +The source-only handoff above is the earlier checkpoint. Root now owns the new isolated interface worktree/branch based main879fb4f5 under pickup6070431864, with exact initial path scope published on main307. The actual native formattedValue capture is9273 UTF-8 bytes, SHA25603baf67b92c7b31390fd497431abf120204a6ccbde7d8b00395d1d40288f6ebf, captured22:18:57UTC8October; full private bytes remain outside Git. The [frozen first-unit contract](2026-10-08-artifact-index-contract.md) records its observed12-column header, worksheet200004, three ranges, typed packet, five independent pins and exact public output. The explicit policy selects three dated rows and eight public columns; actual bindings are empty. No title-based graph/Lean join is inferred. + +The first CLI is stdout-only with stdin private bytes and fixed refusal codes. A later trusted collector writes its admitted public JSON to the bundle; that later wrapper's filesystem/archive custody is separate work. External tests use synthetic TEST capture data and private canaries and precede implementation. Hosted absent-CLI RED, implementation, normal/optimized GREEN and nonauthor privacy review remain pending. Existing namespace, Astro/model/browser and exact-artifact deployment work remains unchanged in scope and unfinished. diff --git a/docs/plans/2026-10-08-artifact-index-contract.md b/docs/plans/2026-10-08-artifact-index-contract.md new file mode 100644 index 000000000..c1e95087a --- /dev/null +++ b/docs/plans/2026-10-08-artifact-index-contract.md @@ -0,0 +1,47 @@ +# Artifact Index public projection v1 contract + +Root owns this new boundary under main307 pickup6070431864 through00:00UTC9October. Base main879fb4f5. This is the first unit of the architecture interface plan. Standing owner authorization applies; no additional permission checkpoint is introduced. Product execution, RED/GREEN and installation occur only in disposable hosted environments. Full private captures stay outside Git and publication. + +## Actual source and public policy + +The real R17 source is spreadsheet `1O6x8ivmaVUxYqKmCOXmToIHpMXqDI362ibqBl8HY8no`, worksheet ID200004, title `Artifact Index`, native12-column formatted-cell representation. The2026-10-08T22:18:57Z response capture is9273 UTF-8 bytes with SHA256 `03baf67b92c7b31390fd497431abf120204a6ccbde7d8b00395d1d40288f6ebf`. This is serialized native API formattedValue data, not raw Sheet-file bytes or an export. It has one sheet and three grid-data ranges: row0 header, rows935–936, and row1000, all starting at column0. Capture ranges are `'Artifact Index'!A1:L1`, `'Artifact Index'!A936:L937`, and `'Artifact Index'!A1001:L1001`, cell_fields `formattedValue`. No native Sheet write is part of this unit. + +Header exact order: Artifact ID; Title; Org; Class; Priority; Topics / object tags; Status; Authority / canonical impact; Dependencies or supersession; Modified UTC; Source; Notes. + +Public column policy exact order: Artifact ID; Title; Org; Class; Status; Authority / canonical impact; Dependencies or supersession; Modified UTC. Only these eight native values may appear in public row fields. Source and Notes are prohibited even under a caller-forged policy; Priority and Topics are excluded. Do not serialize the original capture into any output or failure artifact. The real policy selects only C151-MAIN285-LOG-BINDINGS-20261007-v1, C152-MAIN286-FACTUAL-REVIEW-20261007-v1, C215-GRAPH-ALIAS-AUDIT-20261008-v1. These are dated source records with their original negative/scope/independence labels, not current acceptance. Actual initial bindings are empty because there is no exact verified Artifact Index-to-graph mapping. All49 graph IDs remain unmapped. Synthetic tests use TEST IDs, synthetic spreadsheet IDs, and private canaries distinct from the real source; synthetic fixtures are not delivery evidence. + +## CLI and typed packet + +`python -B -S tools/artifact_index_export.py` reads one strict JSON document from stdin, emits exactly one public JSON document on stdout on success, and exits0. On every refusal, including CLI arguments, it exits nonzero, emits no stdout and exactly the public opaque ASCII constant `ARTIFACT_INDEX_REFUSED` plus one newline on stderr. Private values, base64, cells, paths and exception text cannot appear in diagnostics. It performs no network fetch, URL/path read, imports of captured code or filesystem writes. Module API is `project_artifact_index(packet, expected) -> projection`. The output is a projection, not an authenticated table or current status claim. + +All five independent flags are required exactly once, with lower-case64-hex values: `--expected-capture-sha256`, `--expected-metadata-sha256`, `--expected-policy-sha256`, `--expected-bindings-sha256`, `--expected-graph-nodes-sha256`. No abbreviation or duplicate arguments. Expectations are supplied from trusted root capture/policy inputs, not inferred from stdin. They prove consistency with declared pins only, not external custody. Canonical identity JSON uses sorted object keys, compact separators, ensure_ascii=true and allow_nan=false, preserving typed values and array order. Capture SHA covers exact decoded raw bytes; metadata SHA covers the capture object with only raw_base64 removed; remaining SHA values cover policy, bindings and graph_nodes respectively. + +Packet exact keys: +- schema_version: exact integer1 (bool refuses). +- capture: exactly {capture_id, captured_at, source, raw_base64}. +- policy: exactly {schema_version, artifact_ids, columns}. +- bindings: array of exactly {node_id, artifact_id, roles}. +- graph_nodes: array of explicit node IDs. + +Capture ID is a nonempty ASCII identifier matching `[A-Za-z0-9][A-Za-z0-9_.:-]{0,191}`. captured_at is a valid UTC second timestamp YYYY-MM-DDTHH:MM:SSZ. Source exact keys are {spreadsheet_id, worksheet_id, title, ranges, cell_fields}. spreadsheet_id is a nonempty ASCII identifier of at most192 characters; worksheet_id exact nonnegative integer; title nonempty string, at most256 UTF-8 bytes, no controls; ranges is a unique nonempty list of exact quoted title A-to-L ranges `'TITLE'!A:L`, nonreversed and ordered by starting row; title containing a quote is unsupported. cell_fields must be exactly formattedValue. Native metadata must agree with the declared spreadsheet ID, worksheet ID and title. Metadata is declared as-of, not immutable native revision authentication. + +Stdin limit16MiB, decoded capture limit4MiB, JSON nesting at most64 container levels (a scalar leaf does not add a level), at most4096 nonheader native data rows plus the one header and4096 graph IDs/bindings/policy IDs, each formatted cell at most8192 UTF-8 bytes. Artifact and graph IDs use the same ASCII identifier pattern and192-character bound as capture_id. Reject duplicate object keys, nonfinite numbers, multiple JSON documents, malformed UTF-8/base64/noncanonical base64, booleans used as integers, overflow, empty or unknown envelope fields, and mismatching pins. Unknown native annotations stay private and inert; no unknown native field is projected. + +Native response is an object containing spreadsheetId and one sheet. The sheet properties must identify the declared worksheet/title. Grid properties must declare rowCount positive <=10000000 and columnCount exactly12. Its data list must have exactly one grid block for each declared range in the same order. Missing startRow/startColumn means0; supplied offsets are exact nonnegative integers and must equal the range's0-based position/column0. Every block has exactly its declared number of rowData rows, every row exactly12 values. Each cell is an object; missing formattedValue means empty string, supplied formattedValue must be a string within the bound. Unknown annotations remain inert. Blocks may not overlap and must be inside declared grid bounds. Exactly one row0 contains the exact12-column header; all other rows have a nonempty valid artifact ID, with no duplicate ID even on unselected rows. Nonselected row values and private columns never enter output. Selected IDs must each exist exactly once; unknown/duplicate/invalid policy IDs refuse. + +Policy columns must equal the fixed eight public columns above, in that exact order. Policy schema_version exact1. artifact_ids is a nonempty unique list of valid IDs; its order is preserved for identity hashing but output rows sort by artifact ID. graph_nodes are valid unique IDs. Bindings require known graph ID and selected artifact ID. Roles are a nonempty unique list drawn from source, review, kernel, computation, alignment, teaching-source; no inferred acceptance or title join. Duplicate node/artifact pairs refuse, even with different roles. Explicit valid bindings sort by node_id, artifact_id; their roles sort lexically. Unknown mappings refuse; absent mappings remain unmapped. + +## Exact public output + +Projection exact root fields: +`{schema_version:1,scientific_effect:"NONE",scientific_status_authority:false,custody:"unknown",source,identities,columns,rows,bindings,unmapped_nodes}`. + +source exactly `{capture_id,captured_at,spreadsheet_id,worksheet_id,worksheet_title,ranges,cell_fields,representation:"native-api-formattedValue-json"}`; preserve declared metadata/range order only. It contains no source URL, original packet, native annotations or raw capture bytes. identities exactly `{capture_sha256,metadata_sha256,policy_sha256,bindings_sha256,graph_nodes_sha256}`, matching the independently supplied pins. columns equals the eight public column names. + +Every row exactly `{artifact_id,source_row,fields}`: source_row is original1-based native row, fields is an object with exactly the eight column names and original formatted strings. Do not interpret status, authority, review prose, URLs or scope into another evidence axis. Empty fields stay empty. Every binding exactly {node_id,artifact_id,roles} with explicit canonical sorting. unmapped_nodes is the sorted graph IDs with no binding. Success JSON is deterministic ASCII with sorted object keys and one terminal newline. No output timestamp, original private input, accepted marker or inferred formal join. + +## External controls and completion + +Tests launch the CLI as a subprocess with -B -S and propagate the active optimized mode. Before implementation, actual hosted RED must identify the missing CLI, not merely fail discovery/import/setup. Cover valid deterministic preservation and typing; multiple native ranges/default offsets/blank cells; every independently pinned identity; each required/duplicate/invalid flag; duplicate/nonfinite/deep/oversized JSON and canonical base64; native source/header/range/row/grid/type contradictions; unknown/duplicate artifact IDs and bindings; six explicit roles; unknown graph IDs; missing mappings; forged columns; private Source/Notes and unselected-row canaries absent from public stdout and refusal stderr; inert code/URL strings never fetched or executed. Positive selected public strings may themselves contain URLs/code as inert escaped data; omission is determined by explicit field/row policy. + +Keep fixtures synthetic and compact. Do not copy any real private capture into tests or the public repository. Complete this unit only after exact candidate hosted normal/optimized controls pass with full discovery, original failure/success archives and nonauthor source/privacy review. Real selected source projection/readback follows that tested parser; the capture SHA alone does not authenticate a Sheet revision. Native graph archive readback, evidence-public projection, Astro/build/browser/deployment and all other architecture outcomes remain separate work. diff --git a/tests/test_architecture_artifact_index.py b/tests/test_architecture_artifact_index.py new file mode 100644 index 000000000..46fc4f736 --- /dev/null +++ b/tests/test_architecture_artifact_index.py @@ -0,0 +1,945 @@ +"""External contract for the inert, explicitly pinned Artifact Index projection. + +Every fixture below is synthetic TEST data. It models formattedValue API JSON, +not a real Sheet capture, native revision authentication, or delivered evidence. +""" + +import base64 +import copy +import hashlib +import http.server +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import threading +import urllib.request +import unittest + + +ROOT = Path(__file__).resolve().parents[1] +CLI = ROOT / "tools" / "artifact_index_export.py" +HEADER = [ + "Artifact ID", "Title", "Org", "Class", "Priority", "Topics / object tags", + "Status", "Authority / canonical impact", "Dependencies or supersession", + "Modified UTC", "Source", "Notes", +] +PUBLIC_COLUMNS = [ + "Artifact ID", "Title", "Org", "Class", "Status", + "Authority / canonical impact", "Dependencies or supersession", "Modified UTC", +] +PIN_FLAGS = { + "capture_sha256": "--expected-capture-sha256", + "metadata_sha256": "--expected-metadata-sha256", + "policy_sha256": "--expected-policy-sha256", + "bindings_sha256": "--expected-bindings-sha256", + "graph_nodes_sha256": "--expected-graph-nodes-sha256", +} +ROLES = ["source", "review", "kernel", "computation", "alignment", "teaching-source"] +TITLE = "TEST Artifact Index" +RANGES = [ + "'TEST Artifact Index'!A1:L1", + "'TEST Artifact Index'!A936:L937", + "'TEST Artifact Index'!A1001:L1001", +] + +# Trusted test launcher, executed only in the external child. Its Python audit +# observer covers the fixture paths; it is not an operating-system sandbox. +AUDIT_LAUNCHER = r''' +import json +import os +import runpy +import sys + +cli_path, report_number, fixture_path, secret_path = sys.argv[1:5] +cli_arguments = sys.argv[5:] +report_fd = int(report_number) +fixture_root = os.path.normcase(os.path.abspath(fixture_path)) +labels = ("read", "write", "mkdir", "remove", "rename", "rmdir") +counts = dict.fromkeys(labels, 0) +calibration = dict(counts) +guard = False + +def fixture_target(path, dir_fd=None): + if isinstance(path, int): + # Standard streams and the trusted report descriptor are allowed. + # Other opaque descriptors conservatively count as fixture attempts. + return path not in (0, 1, 2, report_fd) + if path is None: + path = os.getcwd() + path = os.fsdecode(path) + if not os.path.isabs(path) and dir_fd not in (None, -1): + # A relative dir_fd target cannot safely be assigned to another path + # without extra I/O, so observe/guard it conservatively. + return True + path = os.path.normcase(os.path.abspath(path)) + return path == fixture_root or path.startswith(fixture_root + os.sep) + +def audit(event, arguments): + selected = [] + if event == "open" and fixture_target(arguments[0]): + flags = arguments[2] + if flags & (os.O_WRONLY | os.O_RDWR | os.O_APPEND | os.O_CREAT | os.O_TRUNC): + selected.append("write") + if not flags & os.O_WRONLY: + selected.append("read") + elif event in ("os.listdir", "os.scandir") and fixture_target(arguments[0]): + selected.append("read") + elif event == "os.mkdir" and fixture_target(arguments[0], arguments[2]): + selected.append("mkdir") + elif event in ("os.remove", "os.rmdir") and fixture_target(arguments[0], arguments[1]): + selected.append("remove" if event == "os.remove" else "rmdir") + elif event == "os.rename" and ( + fixture_target(arguments[0], arguments[2]) or fixture_target(arguments[1], arguments[3]) + ): + selected.append("rename") + for label in selected: + counts[label] += 1 + if selected and guard: + raise RuntimeError("TEST_FIXTURE_IO_GUARD") + +sys.addaudithook(audit) +try: + # Positive calibration uses trusted paths, never captured code or values. + with open(secret_path, "rb") as stream: + stream.read() + temporary = os.path.join(fixture_root, "TEST-audit-temporary") + renamed = os.path.join(fixture_root, "TEST-audit-renamed") + directory = os.path.join(fixture_root, "TEST-audit-directory") + with open(temporary, "wb") as stream: + stream.write(b"TEST calibration") + os.rename(temporary, renamed) + os.remove(renamed) + os.mkdir(directory) + os.rmdir(directory) + os.listdir(fixture_root) + calibration = dict(counts) + counts = dict.fromkeys(labels, 0) + guard = True + sys.argv = [cli_path, *cli_arguments] + sys.path[0] = os.path.dirname(cli_path) + runpy.run_path(cli_path, run_name="__main__") +finally: + report = {"calibration": calibration, "attempts": counts, "optimize": sys.flags.optimize} + encoded = (json.dumps(report, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + "\n").encode("ascii") + os.write(report_fd, encoded) + os.close(report_fd) +''' + + +def canonical(value): + return json.dumps(value, sort_keys=True, separators=(",", ":"), + ensure_ascii=True, allow_nan=False).encode("ascii") + + +def digest(raw): + return hashlib.sha256(raw).hexdigest() + + +def native_bytes(value): + return (json.dumps(value, separators=(",", ":"), ensure_ascii=False, + allow_nan=False) + "\n").encode("utf-8") + + +def row(values): + return {"values": [{"formattedValue": value} for value in values]} + + +def selected_values(artifact_id, label): + return [ + artifact_id, label + " — α; TEST.node.a is only title text", "TEST Org", + "dated review record", "PRIVATE_PRIORITY_" + artifact_id, + "PRIVATE_TOPICS_" + artifact_id, "HISTORICAL / NOT ACCEPTED", + "independence=0; scope=historical only; as of 2026-10-01", + "supersedes TEST-OLDER only within the stated scope", "2026-10-01T00:00:00Z", + "PRIVATE_SOURCE_" + artifact_id, "PRIVATE_NOTES_" + artifact_id, + ] + + +def synthetic_native(): + unselected = ["TEST-UNSELECTED"] + ["PRIVATE_UNSELECTED_COL_%02d" % i + for i in range(1, 12)] + return { + "spreadsheetId": "TEST-R17-SPREADSHEET", + "privateAnnotation": "PRIVATE_NATIVE_ANNOTATION", + "sheets": [{ + "properties": { + "sheetId": 4242, "title": TITLE, + "gridProperties": {"rowCount": 1002, "columnCount": 12}, + }, + "data": [ + {"startRow": 0, "startColumn": 0, "rowData": [row(HEADER)]}, + {"startRow": 935, "startColumn": 0, "rowData": [ + row(selected_values("TEST-ZETA", "Zeta title")), + row(selected_values("TEST-ALPHA", "Alpha title")), + ]}, + {"startRow": 1000, "startColumn": 0, "rowData": [row(unselected)]}, + ], + }], + } + + +def synthetic_packet(): + return { + "schema_version": 1, + "capture": { + "capture_id": "TEST-capture-1", "captured_at": "2026-10-01T12:34:56Z", + "source": { + "spreadsheet_id": "TEST-R17-SPREADSHEET", "worksheet_id": 4242, + "title": TITLE, "ranges": list(RANGES), "cell_fields": "formattedValue", + }, + "raw_base64": base64.b64encode(native_bytes(synthetic_native())).decode("ascii"), + }, + "policy": {"schema_version": 1, "artifact_ids": ["TEST-ZETA", "TEST-ALPHA"], + "columns": list(PUBLIC_COLUMNS)}, + "bindings": [], + "graph_nodes": ["TEST.node.z", "TEST.node.a", "TEST.node.m"], + } + + +def pins_for(packet): + capture = packet["capture"] + raw = base64.b64decode(capture["raw_base64"], validate=True) + metadata = {key: value for key, value in capture.items() if key != "raw_base64"} + return { + "capture_sha256": digest(raw), "metadata_sha256": digest(canonical(metadata)), + "policy_sha256": digest(canonical(packet["policy"])), + "bindings_sha256": digest(canonical(packet["bindings"])), + "graph_nodes_sha256": digest(canonical(packet["graph_nodes"])), + } + + +def replace_raw(packet, raw): + packet["capture"]["raw_base64"] = base64.b64encode(raw).decode("ascii") + + +def get_native(packet): + return json.loads(base64.b64decode(packet["capture"]["raw_base64"])) + + +def replace_native(packet, value): + replace_raw(packet, native_bytes(value)) + + +class ArtifactIndexProjectionContractTests(unittest.TestCase): + def setUp(self): + self.packet = synthetic_packet() + self.pins = pins_for(self.packet) + self.temp = tempfile.TemporaryDirectory(prefix="TEST-artifact-index-") + self.addCleanup(self.temp.cleanup) + self.workdir = Path(self.temp.name) + self.private_values = ["PRIVATE_NATIVE_ANNOTATION"] + for block in get_native(self.packet)["sheets"][0]["data"][1:]: + for native_row in block["rowData"]: + values = [cell["formattedValue"] for cell in native_row["values"]] + if values[0] == "TEST-UNSELECTED": + self.private_values.extend(values) + else: + self.private_values.extend(values[index] for index in (4, 5, 10, 11)) + + def flag_args(self, pins): + return [argument for name, flag in PIN_FLAGS.items() for argument in (flag, pins[name])] + + def invoke(self, packet=None, *, expected=None, raw=None, args=None, audit=None): + # Missing exporters must fail every control before launching a child. + self.assertTrue(CLI.is_file(), "missing required external CLI: tools/artifact_index_export.py") + packet = self.packet if packet is None else packet + expected = pins_for(packet) if expected is None else expected + raw = canonical(packet) + b"\n" if raw is None else raw + flags = ["-B", "-S"] + if sys.flags.optimize: + flags.append("-" + "O" * sys.flags.optimize) + env = dict(os.environ) + env.pop("PYTHONOPTIMIZE", None) + arguments = self.flag_args(expected) if args is None else args + + def launch(command, pass_fds=()): + return subprocess.run( + command, input=raw, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + cwd=self.workdir, env=env, timeout=20, check=False, pass_fds=pass_fds, + ) + + if audit is None: + return launch([sys.executable, *flags, str(CLI), *arguments]) + with tempfile.TemporaryDirectory(prefix="TEST-audit-report-") as report_directory: + report_root = Path(report_directory) + launcher = report_root / "trusted_launcher.py" + report_path = report_root / "counts.json" + launcher.write_text(AUDIT_LAUNCHER, encoding="utf-8") + report_fd = os.open(report_path, os.O_CREAT | os.O_RDWR | os.O_TRUNC, 0o600) + try: + result = launch([ + sys.executable, *flags, str(launcher), str(CLI), str(report_fd), + str(self.workdir), str(audit["secret"]), *arguments, + ], pass_fds=(report_fd,)) + finally: + os.close(report_fd) + result.audit_report = json.loads(report_path.read_bytes()) + return result + + def assert_private_absent(self, result): + combined = result.stdout + result.stderr + for value in self.private_values: + self.assertNotIn(value.encode("utf-8"), combined) + self.assertNotIn(self.packet["capture"]["raw_base64"].encode("ascii"), combined) + + def expected_projection(self, packet, expected): + source = packet["capture"]["source"] + selected = set(packet["policy"]["artifact_ids"]) + projected_rows = [] + for block in get_native(packet)["sheets"][0]["data"]: + start = block.get("startRow", 0) + for offset, native_row in enumerate(block["rowData"]): + values = [cell.get("formattedValue", "") for cell in native_row["values"]] + if values[0] in selected: + projected_rows.append({ + "artifact_id": values[0], "source_row": start + offset + 1, + "fields": {name: values[HEADER.index(name)] for name in PUBLIC_COLUMNS}, + }) + bindings = [dict(binding, roles=sorted(binding["roles"])) for binding in packet["bindings"]] + bindings.sort(key=lambda binding: (binding["node_id"], binding["artifact_id"])) + bound_nodes = {binding["node_id"] for binding in bindings} + return { + "schema_version": 1, "scientific_effect": "NONE", "scientific_status_authority": False, + "custody": "unknown", + "source": { + "capture_id": packet["capture"]["capture_id"], + "captured_at": packet["capture"]["captured_at"], + "spreadsheet_id": source["spreadsheet_id"], "worksheet_id": source["worksheet_id"], + "worksheet_title": source["title"], "ranges": source["ranges"], + "cell_fields": source["cell_fields"], "representation": "native-api-formattedValue-json", + }, + "identities": expected, "columns": list(PUBLIC_COLUMNS), + "rows": sorted(projected_rows, key=lambda value: value["artifact_id"]), + "bindings": bindings, "unmapped_nodes": sorted(set(packet["graph_nodes"]) - bound_nodes), + } + + def accept(self, packet=None, *, expected=None, raw=None, audit=None): + packet = self.packet if packet is None else packet + expected = pins_for(packet) if expected is None else expected + result = self.invoke(packet, expected=expected, raw=raw, audit=audit) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(result.stderr, b"") + self.assertTrue(result.stdout.isascii()) + projected = json.loads(result.stdout) + self.assertEqual(projected, self.expected_projection(packet, expected)) + self.assertIs(type(projected["schema_version"]), int) + self.assertIs(projected["scientific_status_authority"], False) + self.assertIs(type(projected["source"]["worksheet_id"]), int) + for value in projected["rows"]: + self.assertIs(type(value["source_row"]), int) + self.assertEqual(result.stdout, canonical(projected) + b"\n") + self.assert_private_absent(result) + return result + + def refuse(self, packet=None, *, expected=None, raw=None, args=None): + result = self.invoke(packet, expected=expected, raw=raw, args=args) + self.assertNotEqual(result.returncode, 0) + self.assertEqual(result.stdout, b"") + self.assertEqual(result.stderr, b"ARTIFACT_INDEX_REFUSED\n") + self.assert_private_absent(result) + return result + + def test_exact_public_projection_is_stable_and_retains_original_row_positions(self): + before = sorted(path.name for path in self.workdir.iterdir()) + first = self.accept() + self.assertEqual(self.accept().stdout, first.stdout) + projected = json.loads(first.stdout) + self.assertEqual([(item["artifact_id"], item["source_row"]) for item in projected["rows"]], + [("TEST-ALPHA", 937), ("TEST-ZETA", 936)]) + self.assertEqual(projected["unmapped_nodes"], ["TEST.node.a", "TEST.node.m", "TEST.node.z"]) + self.assertEqual(sorted(path.name for path in self.workdir.iterdir()), before) + + def test_all_49_explicit_graph_nodes_remain_unmapped_without_a_binding(self): + packet = copy.deepcopy(self.packet) + packet["graph_nodes"] = ["TEST.node.%02d" % index for index in reversed(range(49))] + projected = json.loads(self.accept(packet).stdout) + self.assertEqual(len(projected["unmapped_nodes"]), 49) + self.assertEqual(projected["bindings"], []) + + def test_public_unicode_empty_cells_and_historical_scope_are_preserved_without_promotion(self): + packet = copy.deepcopy(self.packet) + native = get_native(packet) + values = native["sheets"][0]["data"][1]["rowData"][1]["values"] + values[1]["formattedValue"] = "Unicode α / résumé / 雪; declared text only" + values[2] = {"unknownAnnotation": "PRIVATE_EMPTY_CELL_ANNOTATION"} + values[6]["formattedValue"] = "FAILED; skipped; as of 2026-09-30" + values[7]["formattedValue"] = "independence=0; historical scope only" + replace_native(packet, native) + result = self.accept(packet) + fields = json.loads(result.stdout)["rows"][0]["fields"] + self.assertEqual(fields["Org"], "") + self.assertEqual(fields["Status"], "FAILED; skipped; as of 2026-09-30") + self.assertNotIn(b"PRIVATE_EMPTY_CELL_ANNOTATION", result.stdout) + + def test_default_zero_offsets_and_unknown_native_annotations_are_private_and_inert(self): + packet = copy.deepcopy(self.packet) + native = get_native(packet) + header = native["sheets"][0]["data"][0] + del header["startRow"] + for block in native["sheets"][0]["data"]: + del block["startColumn"] + block["annotation"] = {"raw": "PRIVATE_BLOCK_ANNOTATION", "nested": [True, None, 1.5]} + native["sheets"][0]["properties"]["annotation"] = "PRIVATE_PROPERTIES_ANNOTATION" + replace_native(packet, native) + result = self.accept(packet) + self.assertNotIn(b"PRIVATE_BLOCK_ANNOTATION", result.stdout) + self.assertNotIn(b"PRIVATE_PROPERTIES_ANNOTATION", result.stdout) + + def test_explicit_six_roles_and_bindings_sort_without_inferred_acceptance(self): + packet = copy.deepcopy(self.packet) + packet["bindings"] = [ + {"node_id": "TEST.node.z", "artifact_id": "TEST-ZETA", "roles": list(ROLES)}, + {"node_id": "TEST.node.a", "artifact_id": "TEST-ZETA", "roles": ["teaching-source"]}, + {"node_id": "TEST.node.a", "artifact_id": "TEST-ALPHA", "roles": ["review", "source"]}, + ] + projected = json.loads(self.accept(packet).stdout) + self.assertEqual(projected["bindings"][2]["roles"], sorted(ROLES)) + self.assertEqual(projected["unmapped_nodes"], ["TEST.node.m"]) + self.assertEqual(projected["custody"], "unknown") + + def test_policy_array_order_is_pinned_even_when_public_rows_have_the_same_order(self): + packet = copy.deepcopy(self.packet) + packet["policy"]["artifact_ids"].reverse() + original = json.loads(self.accept().stdout) + reordered = json.loads(self.accept(packet).stdout) + self.assertEqual(original["rows"], reordered["rows"]) + self.assertNotEqual(original["identities"]["policy_sha256"], reordered["identities"]["policy_sha256"]) + + def test_each_of_the_five_independent_expected_pins_must_match(self): + for name in PIN_FLAGS: + with self.subTest(pin=name): + expected = dict(self.pins) + expected[name] = "0" * 64 if expected[name] != "0" * 64 else "1" * 64 + self.refuse(expected=expected) + + def test_changed_input_components_cannot_replace_independently_supplied_pins(self): + for component in ("capture", "metadata", "policy", "bindings", "graph_nodes"): + with self.subTest(component=component): + packet = copy.deepcopy(self.packet) + if component == "capture": + replace_raw(packet, base64.b64decode(packet["capture"]["raw_base64"]) + b" ") + elif component == "metadata": + packet["capture"]["captured_at"] = "2026-10-02T12:34:56Z" + elif component == "policy": + packet["policy"]["artifact_ids"].reverse() + elif component == "bindings": + packet["bindings"] = [{"node_id": "TEST.node.a", "artifact_id": "TEST-ALPHA", "roles": ["source"]}] + else: + packet["graph_nodes"].reverse() + self.refuse(packet, expected=self.pins) + + def test_all_cli_flags_are_required_once_and_only_full_names_are_accepted(self): + arguments = self.flag_args(self.pins) + for index, (name, flag) in enumerate(PIN_FLAGS.items()): + with self.subTest(flag=flag, fault="missing"): + self.refuse(args=arguments[:2 * index] + arguments[2 * index + 2:]) + with self.subTest(flag=flag, fault="duplicate"): + self.refuse(args=arguments + [flag, self.pins[name]]) + with self.subTest(flag=flag, fault="abbreviated"): + amended = list(arguments) + amended[2 * index] = flag[:-1] + self.refuse(args=amended) + + def test_invalid_cli_digests_and_private_argument_values_are_never_echoed(self): + for name in PIN_FLAGS: + for value in ("", "a" * 63, "a" * 65, "A" * 64, "g" * 64, "../PRIVATE_ARG_PATH"): + with self.subTest(pin=name, value=value): + expected = dict(self.pins) + expected[name] = value + result = self.refuse(expected=expected) + self.assertNotIn(b"PRIVATE_ARG_PATH", result.stderr) + for extra in (["--PRIVATE_UNKNOWN_FLAG", "PRIVATE_ARGUMENT_VALUE"], ["/PRIVATE_POSITIONAL_PATH"]): + with self.subTest(extra=extra): + result = self.refuse(args=self.flag_args(self.pins) + extra) + self.assertNotIn(b"PRIVATE_", result.stderr) + + def test_packet_wrappers_are_closed_and_schema_versions_have_exact_integer_types(self): + wrapped = copy.deepcopy(self.packet) + wrapped["bindings"] = [{"node_id": "TEST.node.a", "artifact_id": "TEST-ALPHA", "roles": ["source"]}] + + def remaining_pins(packet): + # Pin every remaining component, so an unrelated mismatch cannot + # mask the missing/extra-key control. A wholly absent component has + # no hash to supply and retains the positive fixture's expectation. + expected = pins_for(wrapped) + for name, key in (("policy_sha256", "policy"), ("bindings_sha256", "bindings"), + ("graph_nodes_sha256", "graph_nodes")): + if key in packet: + expected[name] = digest(canonical(packet[key])) + if "capture" in packet: + capture = packet["capture"] + expected["metadata_sha256"] = digest(canonical({ + key: value for key, value in capture.items() if key != "raw_base64" + })) + if "raw_base64" in capture: + expected["capture_sha256"] = digest(base64.b64decode(capture["raw_base64"], validate=True)) + return expected + + paths = [(), ("capture",), ("capture", "source"), ("policy",), ("bindings", 0)] + for path in paths: + target = wrapped + for key in path: + target = target[key] + for key in list(target): + with self.subTest(path=path, missing=key): + packet = copy.deepcopy(wrapped) + changed = packet + for part in path: + changed = changed[part] + del changed[key] + self.refuse(packet, expected=remaining_pins(packet)) + with self.subTest(path=path, extra=True): + packet = copy.deepcopy(wrapped) + changed = packet + for part in path: + changed = changed[part] + changed["PRIVATE_UNRECOGNIZED_WRAPPER"] = "PRIVATE_WRAPPER_VALUE" + result = self.refuse(packet, expected=remaining_pins(packet)) + self.assertNotIn(b"PRIVATE_UNRECOGNIZED_WRAPPER", result.stderr) + self.assertNotIn(b"PRIVATE_WRAPPER_VALUE", result.stderr) + for path in ((), ("policy",)): + for value in (True, 1.0, "1", 0, 2, None): + with self.subTest(path=path, schema=value): + packet = copy.deepcopy(self.packet) + target = packet if not path else packet["policy"] + target["schema_version"] = value + self.refuse(packet) + + def test_capture_identity_timestamp_and_declared_source_are_strictly_typed(self): + cases = { + "capture_id": ("", "x" * 193, " space", "TEST\nID", "é", True, 1, None), + "captured_at": ("2026-02-30T12:00:00Z", "2026-10-01T24:00:00Z", "2026-10-01T12:34:56+00:00", + "2026-10-01T12:34:56.000Z", "2026-10-01", True, None), + } + for key, values in cases.items(): + for value in values: + with self.subTest(field=key, value=value): + packet = copy.deepcopy(self.packet) + packet["capture"][key] = value + self.refuse(packet) + source_cases = { + "spreadsheet_id": ("", "x" * 193, "é", True, None), + "worksheet_id": (True, 4242.0, -1, "4242", None), + "title": ("", "λ" * 129, "TEST\nTitle", "TEST'Quote", True, None), + "cell_fields": ("", "formattedValue,userEnteredValue", ["formattedValue"], None), + } + for key, values in source_cases.items(): + for value in values: + with self.subTest(source_field=key, value=value): + packet = copy.deepcopy(self.packet) + packet["capture"]["source"][key] = value + self.refuse(packet) + + def test_native_spreadsheet_worksheet_and_title_must_match_declared_metadata(self): + for field, value in (("spreadsheetId", "TEST-OTHER-SHEET"), ("sheetId", 4243), + ("sheetId", True), ("sheetId", 4242.0), ("title", "TEST Other Title")): + with self.subTest(field=field, value=value): + packet = copy.deepcopy(self.packet) + native = get_native(packet) + target = native if field == "spreadsheetId" else native["sheets"][0]["properties"] + target[field] = value + replace_native(packet, native) + self.refuse(packet) + + def test_exact_header_order_names_types_and_single_header_are_required(self): + for fault in ("rename", "swap", "empty", "nonstring", "second_header"): + with self.subTest(fault=fault): + packet = copy.deepcopy(self.packet) + native = get_native(packet) + header = native["sheets"][0]["data"][0]["rowData"][0]["values"] + if fault == "rename": + header[10]["formattedValue"] = "Sources" + elif fault == "swap": + header[0], header[1] = header[1], header[0] + elif fault == "empty": + header[11] = {} + elif fault == "nonstring": + header[0]["formattedValue"] = 1 + else: + native["sheets"][0]["data"][2]["rowData"][0] = row(HEADER) + replace_native(packet, native) + self.refuse(packet) + + def test_ranges_require_exact_title_a_to_l_syntax_order_and_nonoverlap(self): + cases = [[], RANGES + [RANGES[1]], list(reversed(RANGES)), + [RANGES[0], "'TEST Artifact Index'!A936:L937", "'TEST Artifact Index'!A937:L1001"], + ["TEST Artifact Index!A1:L1", *RANGES[1:]], + ["'TEST Other Title'!A1:L1", *RANGES[1:]], + ["'TEST Artifact Index'!A0:L1", *RANGES[1:]], + ["'TEST Artifact Index'!A01:L1", *RANGES[1:]], + ["'TEST Artifact Index'!B1:L1", *RANGES[1:]], + ["'TEST Artifact Index'!A1:K1", *RANGES[1:]], + [RANGES[0], "'TEST Artifact Index'!A937:L936", RANGES[2]], + [RANGES[0], True, RANGES[2]], "PRIVATE_RANGES_STRING"] + for value in cases: + with self.subTest(ranges=value): + packet = copy.deepcopy(self.packet) + packet["capture"]["source"]["ranges"] = value + self.refuse(packet) + + def test_grid_dimensions_are_exact_integers_and_bound_all_declared_blocks(self): + for key, value in (("rowCount", 0), ("rowCount", -1), ("rowCount", True), + ("rowCount", 1002.0), ("rowCount", 1000), ("rowCount", 10000001), + ("columnCount", 11), ("columnCount", 13), ("columnCount", 12.0), + ("columnCount", True)): + with self.subTest(field=key, value=value): + packet = copy.deepcopy(self.packet) + native = get_native(packet) + native["sheets"][0]["properties"]["gridProperties"][key] = value + replace_native(packet, native) + self.refuse(packet) + + def test_native_offsets_match_range_positions_and_refuse_boolean_float_or_missing_nonzero(self): + for block_index, key, value in ((0, "startRow", 1), (1, "startRow", 934), + (1, "startRow", True), (1, "startRow", 935.0), + (1, "startRow", -1), (1, "startRow", "935"), + (1, "startRow", None), (2, "startColumn", 1), + (2, "startColumn", False), (2, "startColumn", 0.0)): + with self.subTest(block=block_index, offset=key, value=value): + packet = copy.deepcopy(self.packet) + native = get_native(packet) + block = native["sheets"][0]["data"][block_index] + if value is None: + del block[key] + else: + block[key] = value + replace_native(packet, native) + self.refuse(packet) + + def test_native_sheet_block_row_and_cell_shapes_and_cardinalities_are_enforced(self): + for fault in ("root_array", "no_sheet", "two_sheets", "no_block", "extra_block", "block_order", + "short_rows", "extra_rows", "short_cells", "extra_cells", "cell_not_object", + "row_not_object", "block_not_object", "missing_properties", "missing_grid"): + with self.subTest(fault=fault): + packet = copy.deepcopy(self.packet) + native = get_native(packet) + sheet = native["sheets"][0] + blocks = sheet["data"] + if fault == "root_array": + native = [native] + elif fault == "no_sheet": + native["sheets"] = [] + elif fault == "two_sheets": + native["sheets"].append(copy.deepcopy(sheet)) + elif fault == "no_block": + blocks.pop() + elif fault == "extra_block": + blocks.append(copy.deepcopy(blocks[2])) + elif fault == "block_order": + blocks[1], blocks[2] = blocks[2], blocks[1] + elif fault == "short_rows": + blocks[1]["rowData"].pop() + elif fault == "extra_rows": + blocks[1]["rowData"].append(copy.deepcopy(blocks[1]["rowData"][0])) + elif fault == "short_cells": + blocks[2]["rowData"][0]["values"].pop() + elif fault == "extra_cells": + blocks[2]["rowData"][0]["values"].append({}) + elif fault == "cell_not_object": + blocks[2]["rowData"][0]["values"][10] = "PRIVATE_BAD_CELL" + elif fault == "row_not_object": + blocks[2]["rowData"][0] = [] + elif fault == "block_not_object": + blocks[2] = [] + elif fault == "missing_properties": + del sheet["properties"] + else: + del sheet["properties"]["gridProperties"] + replace_native(packet, native) + result = self.refuse(packet) + self.assertNotIn(b"PRIVATE_BAD_CELL", result.stderr) + + def test_all_formatted_cells_are_strings_including_private_and_unselected_cells(self): + for block_index, row_index, column in ((1, 0, 1), (1, 1, 10), (1, 1, 11), (2, 0, 3)): + for value in (None, True, 3, 3.0, ["PRIVATE_ARRAY_CELL"], {"PRIVATE_OBJECT_CELL": "value"}): + with self.subTest(block=block_index, row=row_index, column=column, value=value): + packet = copy.deepcopy(self.packet) + native = get_native(packet) + native["sheets"][0]["data"][block_index]["rowData"][row_index]["values"][column]["formattedValue"] = value + replace_native(packet, native) + result = self.refuse(packet) + self.assertNotIn(b"PRIVATE_ARRAY_CELL", result.stderr) + self.assertNotIn(b"PRIVATE_OBJECT_CELL", result.stderr) + + def test_formatted_cell_limit_counts_utf8_bytes_in_public_private_and_unselected_rows(self): + packet = copy.deepcopy(self.packet) + native = get_native(packet) + native["sheets"][0]["data"][1]["rowData"][1]["values"][1]["formattedValue"] = "λ" * 4096 + replace_native(packet, native) + self.accept(packet) + for block_index, row_index, column in ((1, 1, 1), (1, 0, 10), (2, 0, 11)): + with self.subTest(block=block_index, row=row_index, column=column): + packet = copy.deepcopy(self.packet) + native = get_native(packet) + native["sheets"][0]["data"][block_index]["rowData"][row_index]["values"][column]["formattedValue"] = "λ" * 4096 + "x" + replace_native(packet, native) + self.refuse(packet) + + def test_forged_column_policies_cannot_publish_source_notes_priority_or_topics(self): + cases = [PUBLIC_COLUMNS + [name] for name in ("Source", "Notes", "Priority", "Topics / object tags")] + cases.extend([list(reversed(PUBLIC_COLUMNS)), PUBLIC_COLUMNS[:-1], + PUBLIC_COLUMNS[:-1] + ["Source"], PUBLIC_COLUMNS[:-1] + [PUBLIC_COLUMNS[0]], + PUBLIC_COLUMNS[:-1] + [True], "PRIVATE_COLUMNS_STRING"]) + for columns in cases: + with self.subTest(columns=columns): + packet = copy.deepcopy(self.packet) + packet["policy"]["columns"] = columns + self.refuse(packet) + + def test_policy_ids_are_nonempty_unique_valid_and_present_in_native_rows(self): + packet = copy.deepcopy(self.packet) + native = get_native(packet) + boundary_id = "T" + "x" * 191 + native["sheets"][0]["data"][1]["rowData"][1]["values"][0]["formattedValue"] = boundary_id + packet["policy"]["artifact_ids"] = ["TEST-ZETA", boundary_id] + replace_native(packet, native) + self.accept(packet) + for ids in ([], ["TEST-UNKNOWN"], ["TEST-ALPHA", "TEST-ALPHA"], [""], [True], [1], + ["TEST\nID"], [" TEST-ALPHA"], "TEST-ALPHA", None): + with self.subTest(ids=ids): + packet = copy.deepcopy(self.packet) + packet["policy"]["artifact_ids"] = ids + self.refuse(packet) + + def test_artifact_ids_are_globally_unique_and_valid_even_on_unselected_native_rows(self): + for value in ("TEST-ALPHA", "TEST-ZETA", "", True, 1, " TEST-UNSELECTED", "TEST\nUNSELECTED", + "T" + "x" * 192, "TEST-é", "TEST/UNSELECTED"): + with self.subTest(unselected_id=value): + packet = copy.deepcopy(self.packet) + native = get_native(packet) + native["sheets"][0]["data"][2]["rowData"][0]["values"][0]["formattedValue"] = value + replace_native(packet, native) + self.refuse(packet) + + def test_graph_ids_and_bindings_require_explicit_known_unique_identities(self): + packet = copy.deepcopy(self.packet) + packet["graph_nodes"] = ["T" + "x" * 191] + self.accept(packet) + for nodes in (["TEST.node.a", "TEST.node.a"], [""], [True], [1], ["TEST\nnode"], + [" TEST.node.a"], ["T" + "x" * 192], ["TEST.é"], ["TEST/node"], "TEST.node.a", None): + with self.subTest(nodes=nodes): + packet = copy.deepcopy(self.packet) + packet["graph_nodes"] = nodes + self.refuse(packet) + cases = [ + [{"node_id": "TEST.unknown", "artifact_id": "TEST-ALPHA", "roles": ["source"]}], + [{"node_id": "TEST.node.a", "artifact_id": "TEST-UNSELECTED", "roles": ["source"]}], + [{"node_id": "TEST.node.a", "artifact_id": "TEST-UNKNOWN", "roles": ["source"]}], + [{"node_id": True, "artifact_id": "TEST-ALPHA", "roles": ["source"]}], + [{"node_id": "TEST.node.a", "artifact_id": True, "roles": ["source"]}], + [{"node_id": "TEST.node.a", "artifact_id": "TEST-ALPHA", "roles": ["source"]}, + {"node_id": "TEST.node.a", "artifact_id": "TEST-ALPHA", "roles": ["review"]}], + [True], {}, None, + ] + for bindings in cases: + with self.subTest(bindings=bindings): + packet = copy.deepcopy(self.packet) + packet["bindings"] = bindings + self.refuse(packet) + + def test_binding_roles_are_nonempty_unique_strings_from_the_exact_six_role_set(self): + for roles in ([], ["accepted"], ["Source"], ["source", "source"], ["kernel", True], + [1], "source", None): + with self.subTest(roles=roles): + packet = copy.deepcopy(self.packet) + packet["bindings"] = [{"node_id": "TEST.node.a", "artifact_id": "TEST-ALPHA", "roles": roles}] + self.refuse(packet) + + def test_collection_limits_accept_4096_and_refuse_4097_without_duplicate_shortcuts(self): + packet = copy.deepcopy(self.packet) + packet["graph_nodes"] = ["TEST.node.%04d" % index for index in range(4096)] + packet["bindings"] = [{"node_id": node, "artifact_id": "TEST-ALPHA", "roles": ["source"]} + for node in packet["graph_nodes"]] + self.accept(packet) + too_many = copy.deepcopy(packet) + too_many["graph_nodes"].append("TEST.node.4096") + self.refuse(too_many) + too_many = copy.deepcopy(packet) + too_many["bindings"].append({"node_id": "TEST.node.0000", "artifact_id": "TEST-ZETA", "roles": ["review"]}) + self.refuse(too_many) + for count in (4096, 4097): + with self.subTest(native_rows_and_policy_ids=count): + many = copy.deepcopy(self.packet) + ids = ["TEST-ROW-%04d" % index for index in range(count)] + native = synthetic_native() + native["sheets"][0]["properties"]["gridProperties"]["rowCount"] = count + 1 + native["sheets"][0]["data"] = [ + {"rowData": [row(HEADER)]}, + {"startRow": 1, "rowData": [row([artifact_id] + ["TEST value"] * 9 + + ["PRIVATE_MANY_SOURCE", "PRIVATE_MANY_NOTES"]) + for artifact_id in ids]}, + ] + many["capture"]["source"]["ranges"] = [RANGES[0], "'TEST Artifact Index'!A2:L%d" % (count + 1)] + many["policy"]["artifact_ids"] = ids + replace_native(many, native) + result = self.accept(many) if count == 4096 else self.refuse(many) + self.assertNotIn(b"PRIVATE_MANY_", result.stdout + result.stderr) + + def test_stdin_limit_is_16_mib_of_bytes_including_legal_json_whitespace(self): + raw = canonical(self.packet) + bounded = raw + b" " * (16 * 1024 * 1024 - len(raw)) + self.accept(raw=bounded) + self.refuse(raw=bounded + b" ") + + def test_decoded_capture_limit_is_4_mib_of_original_bytes_not_base64_length(self): + packet = copy.deepcopy(self.packet) + raw = base64.b64decode(packet["capture"]["raw_base64"]) + bounded = raw + b" " * (4 * 1024 * 1024 - len(raw)) + replace_raw(packet, bounded) + self.accept(packet) + replace_raw(packet, bounded + b" ") + self.refuse(packet) + + def test_duplicate_keys_refuse_in_packet_wrappers_and_native_private_annotations(self): + raw = canonical(self.packet) + mutations = [ + raw[:-1] + b',"schema_version":1}', + raw.replace(b'"schema_version":1', b'"schema_version":1,"schema_version":1', 1), + raw.replace(b'"capture_id":"TEST-capture-1"', b'"capture_id":"TEST-capture-1","capture_id":"TEST-capture-1"', 1), + raw.replace(b'"artifact_ids":[', b'"artifact_ids":[],"artifact_ids":[', 1), + ] + for index, changed in enumerate(mutations): + with self.subTest(packet_duplicate=index): + self.refuse(raw=changed, expected=self.pins) + for marker, replacement in ( + (b'"spreadsheetId":"TEST-R17-SPREADSHEET"', b'"spreadsheetId":"TEST-R17-SPREADSHEET","spreadsheetId":"TEST-R17-SPREADSHEET"'), + (b'"formattedValue":"TEST-ZETA"', b'"formattedValue":"TEST-ZETA","formattedValue":"TEST-ZETA"'), + (b'"privateAnnotation":"PRIVATE_NATIVE_ANNOTATION"', b'"privateAnnotation":{"x":1,"x":1}'), + ): + with self.subTest(native_duplicate=marker): + packet = copy.deepcopy(self.packet) + native_raw = base64.b64decode(packet["capture"]["raw_base64"]) + self.assertIn(marker, native_raw) + replace_raw(packet, native_raw.replace(marker, replacement, 1)) + self.refuse(packet) + + def test_nonfinite_numbers_and_numeric_overflow_refuse_even_in_unknown_native_annotations(self): + for token in (b"NaN", b"Infinity", b"-Infinity", b"1e9999"): + with self.subTest(token=token, location="native_annotation"): + packet = copy.deepcopy(self.packet) + raw = base64.b64decode(packet["capture"]["raw_base64"]) + raw = raw.replace(b'"privateAnnotation":"PRIVATE_NATIVE_ANNOTATION"', b'"privateAnnotation":' + token, 1) + replace_raw(packet, raw) + self.refuse(packet) + with self.subTest(token=token, location="packet_schema"): + raw = canonical(self.packet).replace(b'"schema_version":1', b'"schema_version":' + token, 1) + self.refuse(raw=raw, expected=self.pins) + + def test_json_nesting_is_bounded_in_unknown_native_annotations_without_schema_masking(self): + packet = copy.deepcopy(self.packet) + raw = base64.b64decode(packet["capture"]["raw_base64"]) + marker = b'"privateAnnotation":"PRIVATE_NATIVE_ANNOTATION"' + self.assertIn(marker, raw) + # The native root object plus 63 nested arrays occupies 64 levels. + bounded = b"[" * 63 + b"0" + b"]" * 63 + replace_raw(packet, raw.replace(marker, b'"privateAnnotation":' + bounded, 1)) + self.accept(packet) + # One root object plus 64 arrays is exactly one level over the limit. + deep = b"[" * 64 + b"0" + b"]" * 64 + replace_raw(packet, raw.replace(marker, b'"privateAnnotation":' + deep, 1)) + self.refuse(packet) + + def test_malformed_utf8_json_and_multiple_documents_refuse_at_both_input_layers(self): + for raw in (b"", b"{", b"[]", b"null", canonical(self.packet) + b"{}", b"\xff", + canonical(self.packet).replace(b"TEST-capture-1", b"TEST-\xff-capture", 1)): + with self.subTest(packet_raw=raw[:20]): + self.refuse(raw=raw, expected=self.pins) + for raw in (b"", b"{", b"null", b"\xff", native_bytes(synthetic_native()) + b"{}", + native_bytes(synthetic_native()).replace(b"PRIVATE_NATIVE_ANNOTATION", b"PRIVATE_\xff_NATIVE", 1)): + with self.subTest(native_raw=raw[:20]): + packet = copy.deepcopy(self.packet) + replace_raw(packet, raw) + self.refuse(packet) + + def test_base64_is_strict_canonical_and_not_merely_decodable_to_the_expected_bytes(self): + encoded = self.packet["capture"]["raw_base64"] + for value in ("", "!!PRIVATE_BAD_BASE64!!", encoded + "=", encoded + "\n", " " + encoded, + encoded[:-1], True, None): + with self.subTest(base64_type=type(value).__name__, value_prefix=str(value)[:20]): + packet = copy.deepcopy(self.packet) + packet["capture"]["raw_base64"] = value + result = self.refuse(packet, expected=self.pins) + self.assertNotIn(b"PRIVATE_BAD_BASE64", result.stderr) + packet = copy.deepcopy(self.packet) + raw = base64.b64decode(encoded) + raw += b" " * ((1 - len(raw)) % 3) + replace_raw(packet, raw) + canonical_encoded = packet["capture"]["raw_base64"] + self.assertTrue(canonical_encoded.endswith("==")) + alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/" + index = alphabet.index(canonical_encoded[-3]) + alternate = canonical_encoded[:-3] + alphabet[index + 1] + "==" + self.assertEqual(base64.b64decode(alternate, validate=True), raw) + expected = pins_for(packet) + self.accept(packet) + packet["capture"]["raw_base64"] = alternate + self.refuse(packet, expected=expected) + + def test_urls_paths_and_code_are_inert_data_without_fetch_execution_or_filesystem_writes(self): + requests = [] + + class Handler(http.server.BaseHTTPRequestHandler): + def do_GET(self): + requests.append(self.path) + self.send_response(200) + self.end_headers() + self.wfile.write(b"PRIVATE_FETCH_RESULT") + + def log_message(self, *args): + pass + + server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + marker = self.workdir / "PRIVATE_EXECUTION_MARKER" + secret = self.workdir / "PRIVATE_LOCAL_SOURCE" + secret.write_text("PRIVATE_LOCAL_FILE_CONTENT", encoding="utf-8") + packet = copy.deepcopy(self.packet) + native = get_native(packet) + code = "__import__('pathlib').Path(%r).write_text('executed')" % str(marker) + url = "http://127.0.0.1:%d/PRIVATE_URL_PATH" % server.server_port + probe = "http://127.0.0.1:%d/TEST_OBSERVER_PROBE" % server.server_port + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + with opener.open(probe, timeout=5) as response: + self.assertEqual(response.read(), b"PRIVATE_FETCH_RESULT") + self.assertEqual(requests, ["/TEST_OBSERVER_PROBE"]) + requests.clear() + public = "inert URL %s; inert code %s; path %s" % (url, code, secret) + values = native["sheets"][0]["data"][1]["rowData"][1]["values"] + values[1]["formattedValue"] = public + values[10]["formattedValue"] = url + values[11]["formattedValue"] = code + native["privateAnnotation"] = {"file": str(secret), "code": code, "url": url} + replace_native(packet, native) + before = sorted(path.name for path in self.workdir.iterdir()) + result = self.accept(packet, audit={"secret": secret}) + report = result.audit_report + labels = {"read", "write", "mkdir", "remove", "rename", "rmdir"} + self.assertEqual(set(report), {"calibration", "attempts", "optimize"}) + self.assertIs(type(report["optimize"]), int) + self.assertEqual(report["optimize"], sys.flags.optimize) + self.assertEqual(set(report["calibration"]), labels) + for count in report["calibration"].values(): + self.assertIs(type(count), int) + self.assertGreater(count, 0) + self.assertEqual(report["attempts"], dict.fromkeys(labels, 0)) + for count in report["attempts"].values(): + self.assertIs(type(count), int) + projected = json.loads(result.stdout) + self.assertEqual(projected["rows"][0]["fields"]["Title"], public) + self.assertEqual(requests, []) + self.assertFalse(marker.exists()) + self.assertEqual(secret.read_text(encoding="utf-8"), "PRIVATE_LOCAL_FILE_CONTENT") + self.assertEqual(sorted(path.name for path in self.workdir.iterdir()), before) + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +if __name__ == "__main__": + unittest.main() From 9d623398c46791bdb0e98f93364c95f535d064ab Mon Sep 17 00:00:00 2001 From: Dylan Roy <249851642+d6g8k5htny-coder@users.noreply.github.com> Date: Thu, 8 Oct 2026 16:10:02 -0700 Subject: [PATCH 2/5] Implement inert public Artifact Index projection after hosted missing-CLI RED --- architecture/artifact_index.py | 325 +++++++++++++++++++++++++++++++++ tools/artifact_index_export.py | 56 ++++++ 2 files changed, 381 insertions(+) create mode 100644 architecture/artifact_index.py create mode 100644 tools/artifact_index_export.py diff --git a/architecture/artifact_index.py b/architecture/artifact_index.py new file mode 100644 index 000000000..7cb920cf6 --- /dev/null +++ b/architecture/artifact_index.py @@ -0,0 +1,325 @@ +"""Project a pinned native Artifact Index capture into fixed public fields. + +This module treats captures as inert JSON data. It performs no I/O, interprets +no cell content, and authenticates no external custody or scientific status. +""" +from __future__ import annotations + +import base64 +from datetime import datetime +import hashlib +import json +import math +import re + + +STDIN_LIMIT = 16 * 1024 * 1024 +CAPTURE_LIMIT = 4 * 1024 * 1024 +DEPTH_LIMIT = 64 +COLLECTION_LIMIT = 4096 +CELL_LIMIT = 8192 +GRID_ROW_LIMIT = 10_000_000 + +HEADER = ( + "Artifact ID", "Title", "Org", "Class", "Priority", "Topics / object tags", + "Status", "Authority / canonical impact", "Dependencies or supersession", + "Modified UTC", "Source", "Notes", +) +PUBLIC_COLUMNS = ( + "Artifact ID", "Title", "Org", "Class", "Status", + "Authority / canonical impact", "Dependencies or supersession", "Modified UTC", +) +PUBLIC_INDICES = tuple(HEADER.index(column) for column in PUBLIC_COLUMNS) +ROLES = frozenset(("source", "review", "kernel", "computation", "alignment", "teaching-source")) +PIN_KEYS = frozenset(( + "capture_sha256", "metadata_sha256", "policy_sha256", + "bindings_sha256", "graph_nodes_sha256", +)) +IDENTIFIER = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.:-]{0,191}") +SHA256 = re.compile(r"[0-9a-f]{64}") +UTC_SECOND = re.compile(r"([0-9]{4})-([0-9]{2})-([0-9]{2})T([0-9]{2}):([0-9]{2}):([0-9]{2})Z") + + +class ArtifactIndexRefused(ValueError): + """An opaque refusal whose message never includes caller data.""" + + def __init__(self): + super().__init__("ARTIFACT_INDEX_REFUSED") + + +def _require(condition): + if not condition: + raise ArtifactIndexRefused() + + +def _validate_json_value(value, depth=0): + """Validate JSON types, finite numbers, Unicode, and container depth.""" + kind = type(value) + if kind is dict: + _require(depth < DEPTH_LIMIT) + for key, child in value.items(): + _require(type(key) is str) + key.encode("utf-8", errors="strict") + _validate_json_value(child, depth + 1) + elif kind is list: + _require(depth < DEPTH_LIMIT) + for child in value: + _validate_json_value(child, depth + 1) + elif kind is str: + value.encode("utf-8", errors="strict") + elif kind is float: + _require(math.isfinite(value)) + else: + _require(kind in (int, bool, type(None))) + + +def _object_pairs(pairs): + result = {} + for key, value in pairs: + _require(key not in result) + result[key] = value + return result + + +def _nonfinite_constant(_token): + raise ArtifactIndexRefused() + + +def _finite_float(token): + value = float(token) + _require(math.isfinite(value)) + return value + + +def _strict_json(raw, limit): + try: + _require(type(raw) is bytes and 0 < len(raw) <= limit) + value = json.loads( + raw.decode("utf-8", errors="strict"), + object_pairs_hook=_object_pairs, + parse_constant=_nonfinite_constant, + parse_float=_finite_float, + ) + _validate_json_value(value) + return value + except Exception: + raise ArtifactIndexRefused() from None + + +def parse_packet(raw): + """Read one bounded strict JSON packet from caller-supplied bytes.""" + return _strict_json(raw, STDIN_LIMIT) + + +def canonical_json(value): + """Return deterministic ASCII JSON bytes, preserving array order/types.""" + try: + _validate_json_value(value) + return json.dumps( + value, sort_keys=True, separators=(",", ":"), + ensure_ascii=True, allow_nan=False, + ).encode("ascii") + except Exception: + raise ArtifactIndexRefused() from None + + +def _object(value, keys=None): + _require(type(value) is dict) + if keys is not None: + _require(value.keys() == keys) + return value + + +def _integer(value, minimum=0, maximum=None): + _require(type(value) is int and value >= minimum) + if maximum is not None: + _require(value <= maximum) + return value + + +def _identifier(value): + _require(type(value) is str and IDENTIFIER.fullmatch(value) is not None) + return value + + +def _bounded_list(value, minimum=0, maximum=COLLECTION_LIMIT): + _require(type(value) is list and minimum <= len(value) <= maximum) + return value + + +def _unique_identifiers(value, minimum=0): + values = _bounded_list(value, minimum) + seen = set() + for item in values: + _identifier(item) + _require(item not in seen) + seen.add(item) + return seen + + +def _expected_pins(expected): + _object(expected, PIN_KEYS) + for value in expected.values(): + _require(type(value) is str and SHA256.fullmatch(value) is not None) + + +def _capture_metadata(capture): + _object(capture, {"capture_id", "captured_at", "source", "raw_base64"}) + _identifier(capture["capture_id"]) + timestamp = capture["captured_at"] + _require(type(timestamp) is str) + match = UTC_SECOND.fullmatch(timestamp) + _require(match is not None) + # A numeric constructor checks calendar validity without parsing cell text. + datetime(*(int(part) for part in match.groups())) + source = _object(capture["source"], { + "spreadsheet_id", "worksheet_id", "title", "ranges", "cell_fields", + }) + _identifier(source["spreadsheet_id"]) + _integer(source["worksheet_id"]) + title = source["title"] + _require(type(title) is str and 0 < len(title.encode("utf-8")) <= 256) + _require("'" not in title and not any(ord(char) < 32 or 127 <= ord(char) <= 159 for char in title)) + _require(source["cell_fields"] == "formattedValue" and type(source["cell_fields"]) is str) + ranges = _bounded_list(source["ranges"], 1, COLLECTION_LIMIT + 1) + range_pattern = re.compile(r"'" + re.escape(title) + r"'!A([1-9][0-9]*):L([1-9][0-9]*)") + intervals = [] + prior_end = 0 + for declared in ranges: + _require(type(declared) is str) + match = range_pattern.fullmatch(declared) + _require(match is not None) + # Bounds precede integer conversion, including enormous decimal text. + _require(all(len(part) <= 8 for part in match.groups())) + first, last = (int(part) for part in match.groups()) + _require(prior_end < first <= last <= GRID_ROW_LIMIT) + intervals.append((first - 1, last)) + prior_end = last + _require(sum(stop - start for start, stop in intervals) <= COLLECTION_LIMIT + 1) + return source, intervals + + +def _capture_bytes(encoded): + _require(type(encoded) is str and 0 < len(encoded) <= 4 * ((CAPTURE_LIMIT + 2) // 3)) + raw = base64.b64decode(encoded.encode("ascii"), validate=True) + _require(0 < len(raw) <= CAPTURE_LIMIT) + _require(base64.b64encode(raw).decode("ascii") == encoded) + return raw + + +def _native_rows(native, source, intervals): + _object(native) + _require(type(native.get("spreadsheetId")) is str and native["spreadsheetId"] == source["spreadsheet_id"]) + sheets = _bounded_list(native.get("sheets"), 1, 1) + sheet = _object(sheets[0]) + properties = _object(sheet.get("properties")) + _integer(properties.get("sheetId")) + _require(properties["sheetId"] == source["worksheet_id"]) + _require(type(properties.get("title")) is str and properties["title"] == source["title"]) + grid = _object(properties.get("gridProperties")) + row_count = _integer(grid.get("rowCount"), 1, GRID_ROW_LIMIT) + _require(_integer(grid.get("columnCount"), 1) == len(HEADER)) + blocks = _bounded_list(sheet.get("data"), len(intervals), len(intervals)) + rows_by_id = {} + header_count = 0 + for block, (start, stop) in zip(blocks, intervals): + _object(block) + _require(stop <= row_count) + _require(_integer(block.get("startRow", 0)) == start) + _require(_integer(block.get("startColumn", 0)) == 0) + rows = _bounded_list(block.get("rowData"), stop - start, stop - start) + for offset, row in enumerate(rows): + _object(row) + cells = _bounded_list(row.get("values"), len(HEADER), len(HEADER)) + values = [] + for cell in cells: + _object(cell) + value = cell.get("formattedValue", "") + _require(type(value) is str and len(value.encode("utf-8")) <= CELL_LIMIT) + values.append(value) + position = start + offset + if position == 0: + _require(values == list(HEADER)) + header_count += 1 + else: + artifact_id = _identifier(values[0]) + _require(artifact_id not in rows_by_id) + rows_by_id[artifact_id] = (position + 1, values) + _require(len(rows_by_id) <= COLLECTION_LIMIT) + _require(header_count == 1) + return rows_by_id + + +def _project(packet, expected): + _validate_json_value(packet) + _expected_pins(expected) + _object(packet, {"schema_version", "capture", "policy", "bindings", "graph_nodes"}) + _require(_integer(packet["schema_version"]) == 1) + capture = packet["capture"] + source, intervals = _capture_metadata(capture) + raw = _capture_bytes(capture["raw_base64"]) + metadata = {key: value for key, value in capture.items() if key != "raw_base64"} + identities = { + "capture_sha256": hashlib.sha256(raw).hexdigest(), + "metadata_sha256": hashlib.sha256(canonical_json(metadata)).hexdigest(), + "policy_sha256": hashlib.sha256(canonical_json(packet["policy"])).hexdigest(), + "bindings_sha256": hashlib.sha256(canonical_json(packet["bindings"])).hexdigest(), + "graph_nodes_sha256": hashlib.sha256(canonical_json(packet["graph_nodes"])).hexdigest(), + } + _require(identities == expected) + policy = _object(packet["policy"], {"schema_version", "artifact_ids", "columns"}) + _require(_integer(policy["schema_version"]) == 1) + _require(type(policy["columns"]) is list and policy["columns"] == list(PUBLIC_COLUMNS)) + selected = _unique_identifiers(policy["artifact_ids"], 1) + graph_nodes = _unique_identifiers(packet["graph_nodes"]) + bindings = [] + pairs = set() + bound_nodes = set() + for binding in _bounded_list(packet["bindings"]): + _object(binding, {"node_id", "artifact_id", "roles"}) + node = _identifier(binding["node_id"]) + artifact = _identifier(binding["artifact_id"]) + _require(node in graph_nodes and artifact in selected) + roles = _bounded_list(binding["roles"], 1, len(ROLES)) + _require(all(type(role) is str and role in ROLES for role in roles)) + _require(len(set(roles)) == len(roles)) + pair = (node, artifact) + _require(pair not in pairs) + pairs.add(pair) + bound_nodes.add(node) + bindings.append({"node_id": node, "artifact_id": artifact, "roles": sorted(roles)}) + native = _strict_json(raw, CAPTURE_LIMIT) + native_rows = _native_rows(native, source, intervals) + _require(selected <= native_rows.keys()) + rows = [] + for artifact in sorted(selected): + source_row, values = native_rows[artifact] + rows.append({ + "artifact_id": artifact, "source_row": source_row, + "fields": {column: values[index] for column, index in zip(PUBLIC_COLUMNS, PUBLIC_INDICES)}, + }) + return { + "schema_version": 1, "scientific_effect": "NONE", + "scientific_status_authority": False, "custody": "unknown", + "source": { + "capture_id": capture["capture_id"], "captured_at": capture["captured_at"], + "spreadsheet_id": source["spreadsheet_id"], "worksheet_id": source["worksheet_id"], + "worksheet_title": source["title"], "ranges": list(source["ranges"]), + "cell_fields": source["cell_fields"], "representation": "native-api-formattedValue-json", + }, + "identities": identities, "columns": list(PUBLIC_COLUMNS), "rows": rows, + "bindings": sorted(bindings, key=lambda item: (item["node_id"], item["artifact_id"])), + "unmapped_nodes": sorted(graph_nodes - bound_nodes), + } + + +def project_artifact_index(packet, expected): + """Return the exact public projection, or raise an opaque refusal. + + Independent expected pins are mandatory; no expectation is learned from + the packet. Input objects are not mutated and private data is not returned. + """ + try: + return _project(packet, expected) + except Exception: + raise ArtifactIndexRefused() from None diff --git a/tools/artifact_index_export.py b/tools/artifact_index_export.py new file mode 100644 index 000000000..2ce248406 --- /dev/null +++ b/tools/artifact_index_export.py @@ -0,0 +1,56 @@ +#!/usr/bin/env python3 +"""Emit a pinned Artifact Index public projection from stdin only.""" +from __future__ import annotations + +import os +import sys + +# Keep the CLI's own imports read-only even if a caller omits Python's -B flag. +sys.dont_write_bytecode = True +# Resolve only the trusted source location lexically, never a caller path. +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from architecture.artifact_index import ( + ArtifactIndexRefused, SHA256, STDIN_LIMIT, canonical_json, parse_packet, + project_artifact_index, +) + + +FLAGS = { + "--expected-capture-sha256": "capture_sha256", + "--expected-metadata-sha256": "metadata_sha256", + "--expected-policy-sha256": "policy_sha256", + "--expected-bindings-sha256": "bindings_sha256", + "--expected-graph-nodes-sha256": "graph_nodes_sha256", +} + + +def _arguments(arguments): + if len(arguments) != 2 * len(FLAGS): + raise ArtifactIndexRefused() + expected = {} + for index in range(0, len(arguments), 2): + flag, value = arguments[index:index + 2] + name = FLAGS.get(flag) + if name is None or name in expected or SHA256.fullmatch(value) is None: + raise ArtifactIndexRefused() + expected[name] = value + return expected + + +def main(): + try: + expected = _arguments(sys.argv[1:]) + raw = sys.stdin.buffer.read(STDIN_LIMIT + 1) + packet = parse_packet(raw) + projection = project_artifact_index(packet, expected) + output = canonical_json(projection) + b"\n" + except Exception: + sys.stderr.buffer.write(b"ARTIFACT_INDEX_REFUSED\n") + return 1 + sys.stdout.buffer.write(output) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) From 21cb96777acf71c5327fff33da5feb69c8c4a0b8 Mon Sep 17 00:00:00 2001 From: Dylan Roy <249851642+d6g8k5htny-coder@users.noreply.github.com> Date: Thu, 15 Oct 2026 20:12:20 -0500 Subject: [PATCH 3/5] Reproduce CLI import and stdout boundaries before product fixes Add four external CLI regression methods with calibrated synthetic sibling-package canaries and real closed-reader pipes, including the graph reader's buffered-flush boundary. Preserve both affected product CLIs unchanged. Run the focused diagnostic in both Python modes on the exact experimental branch, with original context/results retained. This diagnostic does not replace the required full suite. Scientific effect: NONE. --- .../architecture-cli-boundary-regression.yml | 135 +++++++++++ tests/test_architecture_cli_boundaries.py | 225 ++++++++++++++++++ 2 files changed, 360 insertions(+) create mode 100644 .github/workflows/architecture-cli-boundary-regression.yml create mode 100644 tests/test_architecture_cli_boundaries.py diff --git a/.github/workflows/architecture-cli-boundary-regression.yml b/.github/workflows/architecture-cli-boundary-regression.yml new file mode 100644 index 000000000..784fb27dd --- /dev/null +++ b/.github/workflows/architecture-cli-boundary-regression.yml @@ -0,0 +1,135 @@ +name: Architecture CLI boundary diagnostic + +'on': + push: + branches: + - codex/architecture-cli-boundary-red-20261016 + pull_request: + paths: + - architecture/artifact_index.py + - architecture/graph_artifact_readback.py + - tools/artifact_index_export.py + - tools/architecture_graph_artifact_readback.py + - tests/test_architecture_artifact_index.py + - tests/test_architecture_graph_artifact_readback.py + - tests/test_architecture_cli_boundaries.py + - .github/workflows/architecture-cli-boundary-regression.yml + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: architecture-cli-boundary-${{ github.run_id }}-${{ github.run_attempt }} + cancel-in-progress: false + +jobs: + cli-boundaries: + name: architecture-cli-boundary-${{ matrix.mode }} + runs-on: ubuntu-24.04 + timeout-minutes: 6 + strategy: + fail-fast: false + matrix: + mode: [normal, optimized] + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + timeout-minutes: 1 + with: + persist-credentials: false + fetch-depth: 1 + + - name: Record diagnostic native context before running controls + timeout-minutes: 1 + env: + DIAGNOSTIC_MODE: ${{ matrix.mode }} + run: | + set -euo pipefail + evidence="$RUNNER_TEMP/architecture-cli-boundary-$GITHUB_RUN_ID-$GITHUB_RUN_ATTEMPT-$DIAGNOSTIC_MODE" + mkdir -m 700 "$evidence" + git rev-parse HEAD > "$evidence/checked-commit.txt" + python3 -I -B -S - "$evidence" <<'PY' + from datetime import datetime, timezone + import hashlib + import json + import os + from pathlib import Path + import platform + import sys + root = Path(os.environ['GITHUB_WORKSPACE']) + destination = Path(sys.argv[1]) + paths = ( + '.github/workflows/architecture-cli-boundary-regression.yml', + 'tests/test_architecture_cli_boundaries.py', + 'tests/test_architecture_artifact_index.py', + 'tests/test_architecture_graph_artifact_readback.py', + 'tools/artifact_index_export.py', + 'tools/architecture_graph_artifact_readback.py', + 'architecture/artifact_index.py', + 'architecture/graph_artifact_readback.py', + 'docs/site/dependency-source/GRAPH.json', + ) + sources = {} + for name in paths: + raw = (root / name).read_bytes() + sources[name] = {'bytes': len(raw), 'sha256': hashlib.sha256(raw).hexdigest()} + context = { + 'schema': 'architecture-cli-boundary-diagnostic/v1', + 'recorded_at': datetime.now(timezone.utc).isoformat(), + 'repository': os.environ['GITHUB_REPOSITORY'], + 'event_name': os.environ['GITHUB_EVENT_NAME'], + 'workflow_ref': os.environ['GITHUB_WORKFLOW_REF'], + 'workflow_sha': os.environ['GITHUB_WORKFLOW_SHA'], + 'event_sha': os.environ['GITHUB_SHA'], + 'checked_commit': (destination / 'checked-commit.txt').read_text().strip(), + 'run_id': os.environ['GITHUB_RUN_ID'], + 'run_attempt': os.environ['GITHUB_RUN_ATTEMPT'], + 'mode': os.environ['DIAGNOSTIC_MODE'], + 'python': sys.version, + 'platform': platform.platform(), + 'source_identities': sources, + 'scope': 'four CLI boundary methods only; required full suite remains separate', + 'scientific_effect': 'NONE', + 'scientific_status_authority': False, + } + (destination / 'native-context.json').write_text( + json.dumps(context, sort_keys=True, indent=2) + '\n', encoding='utf-8') + PY + + - name: Run only the four external CLI boundary controls + timeout-minutes: 3 + env: + DIAGNOSTIC_MODE: ${{ matrix.mode }} + ARCHITECTURE_CLI_BOUNDARY_REPORT: ${{ runner.temp }}/architecture-cli-boundary-${{ github.run_id }}-${{ github.run_attempt }}-${{ matrix.mode }}/child-results.jsonl + run: | + set -euo pipefail + evidence="$RUNNER_TEMP/architecture-cli-boundary-$GITHUB_RUN_ID-$GITHUB_RUN_ATTEMPT-$DIAGNOSTIC_MODE" + flags=(-I -B -S) + if [ "$DIAGNOSTIC_MODE" = optimized ]; then + flags+=(-O) + fi + printf '%q ' timeout --signal=TERM --kill-after=5s 150s \ + python3 "${flags[@]}" -m unittest discover -s tests -p test_architecture_cli_boundaries.py -v \ + > "$evidence/test-command.txt" + printf '\n' >> "$evidence/test-command.txt" + set +e + timeout --signal=TERM --kill-after=5s 150s \ + python3 "${flags[@]}" -m unittest discover -s tests -p test_architecture_cli_boundaries.py -v \ + > "$evidence/tests.stdout" 2> "$evidence/tests.stderr" + status=$? + set -e + printf '%s\n' "$status" > "$evidence/test-exit-status.txt" + cat "$evidence/tests.stdout" + cat "$evidence/tests.stderr" >&2 + exit "$status" + + - name: Retain separate diagnostic originals on every conclusion + if: always() + timeout-minutes: 1 + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 + with: + name: architecture-cli-boundary-${{ matrix.mode }}-${{ github.run_id }}-${{ github.run_attempt }} + path: ${{ runner.temp }}/architecture-cli-boundary-${{ github.run_id }}-${{ github.run_attempt }}-${{ matrix.mode }}/ + if-no-files-found: error + retention-days: 90 + compression-level: 6 diff --git a/tests/test_architecture_cli_boundaries.py b/tests/test_architecture_cli_boundaries.py new file mode 100644 index 000000000..9b526cae5 --- /dev/null +++ b/tests/test_architecture_cli_boundaries.py @@ -0,0 +1,225 @@ +"""External CLI boundary regressions; execute only in hosted isolation. + +The sibling package and its private-input canary are trusted, synthetic TEST +instruments. They never refer to an actual private capture or user file. Existing +fixture modules are imported as modules, so their TestCases are not rediscovered +or inherited by these four controls. A passing diagnostic is not the full suite. +""" + +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest + +import test_architecture_artifact_index as artifact_fixtures +import test_architecture_graph_artifact_readback as graph_fixtures + + +ROOT = Path(__file__).resolve().parents[1] +ARTIFACT_CLI = ROOT / "tools/artifact_index_export.py" +GRAPH_CLI = ROOT / "tools/architecture_graph_artifact_readback.py" +CHILD_DEADLINE_SECONDS = 20 +PRIVATE_CANARY = b"TEST_ONLY_SYNTHETIC_PRIVATE_INPUT_CANARY" +# Trusted test instrument: the real stdout descriptor receives a real Python +# buffered stream. No exception, captured value or product result is fabricated. +BUFFERED_STDOUT_LAUNCHER = r''' +import io +import runpy +import sys +capacity, cli = int(sys.argv[1]), sys.argv[2] +sys.stdout = io.TextIOWrapper( + io.BufferedWriter(io.FileIO(1, mode="wb", closefd=False), buffer_size=capacity), + encoding="utf-8", newline="\n") +sys.argv = [cli, *sys.argv[3:]] +runpy.run_path(cli, run_name="__main__") +''' + + +class ArchitectureCliBoundaryContract(unittest.TestCase): + def setUp(self): + temporary = tempfile.TemporaryDirectory(prefix="TEST-cli-boundary-") + self.addCleanup(temporary.cleanup) + self.cwd = Path(temporary.name) + self.environment = dict(os.environ) + for name in ("PYTHONPATH", "PYTHONHOME", "PYTHONOPTIMIZE", "PYTHONUNBUFFERED"): + self.environment.pop(name, None) + + def python_flags(self): + flags = ["-I", "-B", "-S"] + if sys.flags.optimize: + flags.append("-" + "O" * sys.flags.optimize) + return flags + + def invoke(self, command, raw=b"", *, closed_stdout=False): + """Launch the real child, optionally with a pipe that has no reader. + + The reader is closed before Popen, so no child write or shutdown flush + can race a parent read/close. Parent and child each close their writer. + """ + writer = None + if closed_stdout: + reader, writer = os.pipe() + os.close(reader) + try: + process = subprocess.Popen( + command, stdin=subprocess.PIPE, + stdout=writer if closed_stdout else subprocess.PIPE, + stderr=subprocess.PIPE, cwd=self.cwd, env=self.environment, + ) + finally: + if writer is not None: + os.close(writer) + try: + stdout, stderr = process.communicate(raw, timeout=CHILD_DEADLINE_SECONDS) + except subprocess.TimeoutExpired: + process.kill() + # Killing the child closes its pipes; keep cleanup bounded as well. + process.communicate(timeout=5) + self.fail("CLI child exceeded its 20-second deadline") + return subprocess.CompletedProcess(command, process.returncode, stdout or b"", stderr) + + def artifact_case(self): + packet = artifact_fixtures.synthetic_packet() + expected = artifact_fixtures.pins_for(packet) + arguments = [part for name, flag in artifact_fixtures.PIN_FLAGS.items() + for part in (flag, expected[name])] + return ARTIFACT_CLI, arguments, artifact_fixtures.canonical(packet) + b"\n" + + def graph_case(self): + # Reuse the existing fixture builder and its source-pin assertions, + # without inheriting or executing any of its test methods. + fixture = graph_fixtures.ArchitectureGraphArtifactReadbackContract() + self.addCleanup(fixture.doCleanups) + fixture.setUp() + raw = graph_fixtures.classic_zip(fixture.entries()) + archive = fixture.cwd / "TEST-cli-boundary-input.zip" + archive.write_bytes(raw) + return GRAPH_CLI, fixture.arguments(archive, fixture.expected(raw)), b"" + + def command(self, cli, arguments): + self.assertTrue(cli.is_file(), "required external CLI is absent") + return [sys.executable, *self.python_flags(), str(cli), *arguments] + + def accepted_baseline(self, case): + cli, arguments, raw = case + result = self.invoke(self.command(cli, arguments), raw) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(result.stderr, b"") + document = json.loads(result.stdout) + self.assertEqual(document["scientific_effect"], "NONE") + self.assertIs(document["scientific_status_authority"], False) + self.assertEqual(document["custody"], "unknown") + return result.stdout + + def symlink_does_not_select_sibling_package(self, case): + cli, arguments, raw = case + baseline = self.accepted_baseline(case) + attack_root = self.cwd / "TEST-attacker-source-root" + tools = attack_root / "tools" + package = attack_root / "architecture" + tools.mkdir(parents=True) + package.mkdir() + secret = self.cwd / "TEST-synthetic-private-input" + marker = self.cwd / "TEST-sibling-package-read-marker" + secret.write_bytes(PRIVATE_CANARY) + observed = PRIVATE_CANARY + b"\nOPTIMIZE=" + str(sys.flags.optimize).encode("ascii") + package.joinpath("__init__.py").write_text( + "from pathlib import Path\n" + "import sys\n" + "Path(" + repr(str(marker)) + ").write_bytes(\n" + " Path(" + repr(str(secret)) + ").read_bytes()\n" + " + b'\\nOPTIMIZE=' + str(sys.flags.optimize).encode('ascii'))\n" + "raise RuntimeError('TEST_SYNTHETIC_SIBLING_PACKAGE_EXECUTED')\n", + encoding="utf-8", + ) + linked_cli = tools / cli.name + linked_cli.symlink_to(cli) + + # Positive calibration deliberately imports this trusted TEST package. + # It proves the marker detects a synthetic private read in this mode. + calibration = self.invoke([ + sys.executable, *self.python_flags(), "-c", + "import sys; sys.path.insert(0, sys.argv[1]); import architecture", + str(attack_root), + ]) + self.record_diagnostic_child("synthetic-sibling-package-calibration", calibration) + self.assertNotEqual(calibration.returncode, 0) + self.assertIn(b"TEST_SYNTHETIC_SIBLING_PACKAGE_EXECUTED", calibration.stderr) + self.assertEqual(marker.read_bytes(), observed, "sibling-package observer was not live") + marker.unlink() + + result = self.invoke(self.command(linked_cli, arguments), raw) + self.record_diagnostic_child("symlinked-cli-with-synthetic-sibling", result) + self.assertEqual(secret.read_bytes(), PRIVATE_CANARY) + self.assertFalse(marker.exists(), "symlink-selected sibling package read the synthetic private input") + self.assertNotIn(PRIVATE_CANARY, result.stdout + result.stderr) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(result.stderr, b"") + self.assertEqual(result.stdout, baseline, "symlink invocation changed the validated public unit") + + def record_diagnostic_child(self, variant, result): + # Only the focused hosted workflow sets this path. Do not print between + # unittest's progress and result tokens in the required full-suite log. + report = self.environment.get("ARCHITECTURE_CLI_BOUNDARY_REPORT") + if report is not None: + record = { + "test_id": self.id(), "variant": variant, + "optimize": sys.flags.optimize, "returncode": result.returncode, + "stdout_bytes": len(result.stdout), + "stderr": result.stderr.decode("utf-8", errors="backslashreplace"), + } + with open(report, "ab") as stream: + stream.write((json.dumps(record, sort_keys=True) + "\n").encode("utf-8")) + + def assert_opaque_refusal(self, result, refusal): + self.assertEqual(result.returncode, 1, result.stderr) + self.assertEqual(result.stdout, b"") + self.assertEqual(result.stderr, refusal) + + def broken_stdout_has_opaque_refusal(self, case, refusal, *, buffered_flush=False): + # This positive run ensures an unrelated argument/fixture refusal cannot + # satisfy the closed-reader protocol assertion below. + baseline = self.accepted_baseline(case) + cli, arguments, raw = case + with self.subTest(stdout_boundary="direct CLI"): + result = self.invoke(self.command(cli, arguments), raw, closed_stdout=True) + self.record_diagnostic_child("direct-closed-reader", result) + self.assert_opaque_refusal(result, refusal) + if buffered_flush: + # The graph unit can exceed Python's default buffer and fail during + # direct write, leaving no bytes for shutdown to flush. A larger + # real buffer exercises flush-time failure instead. Its capacity is + # derived from the already accepted bounded synthetic unit only. + capacity = len(baseline) + 1 + self.assertLessEqual(capacity, 16 * 1024 * 1024 + 1) + command = [sys.executable, *self.python_flags(), "-c", BUFFERED_STDOUT_LAUNCHER, + str(capacity), str(cli), *arguments] + calibration = self.invoke(command, raw) + self.record_diagnostic_child("buffered-open-reader-calibration", calibration) + self.assertEqual(calibration.returncode, 0, calibration.stderr) + self.assertEqual(calibration.stderr, b"") + self.assertEqual(calibration.stdout, baseline) + with self.subTest(stdout_boundary="real buffered flush"): + result = self.invoke(command, raw, closed_stdout=True) + self.record_diagnostic_child("buffered-closed-reader", result) + self.assert_opaque_refusal(result, refusal) + + def test_artifact_index_symlink_cannot_read_synthetic_private_input(self): + self.symlink_does_not_select_sibling_package(self.artifact_case()) + + def test_graph_readback_symlink_cannot_read_synthetic_private_input(self): + self.symlink_does_not_select_sibling_package(self.graph_case()) + + def test_artifact_index_broken_stdout_has_one_opaque_refusal_and_exit_one(self): + self.broken_stdout_has_opaque_refusal(self.artifact_case(), b"ARTIFACT_INDEX_REFUSED\n") + + def test_graph_readback_broken_stdout_has_one_opaque_refusal_and_exit_one(self): + self.broken_stdout_has_opaque_refusal( + self.graph_case(), b"ARCHITECTURE_GRAPH_READBACK_REFUSED\n", buffered_flush=True) + + +if __name__ == "__main__": + unittest.main() From 580a5c22eb41ff571e96011590a7bacda5fec715 Mon Sep 17 00:00:00 2001 From: Dylan Roy <249851642+d6g8k5htny-coder@users.noreply.github.com> Date: Thu, 15 Oct 2026 20:15:47 -0500 Subject: [PATCH 4/5] Fix CLI trust roots and broken-pipe refusals after hosted RED Resolve each CLI's actual source path before importing trusted modules. Keep terminal stdout errors to the existing fixed refusal and status1 without a residual interpreter shutdown flush. Preserve accepted output bytes and all source/admission limits. Retain the four regression methods and diagnostic workflow from the genuine normal/optimized RED at run38833849114. Increase only the combined architecture command/step/job envelope to900s/16m/49m after the retained600s run stopped during optimized execution; physical resources, security controls and custody gates remain unchanged. Scientific effect: NONE. --- .github/workflows/research-architecture.yml | 8 ++++---- tools/architecture_graph_artifact_readback.py | 9 ++++++--- tools/artifact_index_export.py | 16 +++++++++++++--- 3 files changed, 23 insertions(+), 10 deletions(-) diff --git a/.github/workflows/research-architecture.yml b/.github/workflows/research-architecture.yml index d59679046..8739e26fd 100644 --- a/.github/workflows/research-architecture.yml +++ b/.github/workflows/research-architecture.yml @@ -36,8 +36,8 @@ jobs: graph: name: research-architecture runs-on: ubuntu-24.04 - # Complete stage ceiling: 19 preparation + 11 tests + fourteen 1-minute stages. - timeout-minutes: 44 + # Complete stage ceiling: 19 preparation + 16 tests + fourteen 1-minute stages. + timeout-minutes: 49 outputs: checked_commit: ${{ steps.accepted.outputs.checked_commit }} repository: ${{ steps.accepted.outputs.repository }} @@ -311,13 +311,13 @@ jobs: printf '%s\n' "$GITHUB_RUN_ATTEMPT" > "$RUNNER_TEMP/architecture-publish/run-attempt.txt" - name: Run controls and export in ephemeral restricted container id: architecture_controls - timeout-minutes: 11 + timeout-minutes: 16 env: IMAGE: python:3.11.14-slim-bookworm@sha256:83f339c1be6340ae1096010fdccf6552ac932d8f410d45d206014916bdf37e48 run: | set -euo pipefail set +e - timeout --signal=TERM --kill-after=10s 600s docker run --pull=never --rm --log-driver none --name architecture-controls \ + timeout --signal=TERM --kill-after=10s 900s docker run --pull=never --rm --log-driver none --name architecture-controls \ --network none --read-only --user 65532:65532 --cpus 2 --memory 1g --pids-limit 128 \ --cap-drop ALL --security-opt no-new-privileges \ --env ARCHITECTURE_SANDBOX=1 \ diff --git a/tools/architecture_graph_artifact_readback.py b/tools/architecture_graph_artifact_readback.py index 9a8a0b822..5df3c59b6 100644 --- a/tools/architecture_graph_artifact_readback.py +++ b/tools/architecture_graph_artifact_readback.py @@ -9,7 +9,7 @@ import sys sys.dont_write_bytecode = True -ROOT = Path(__file__).parent.parent +ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) from architecture.graph_artifact_readback import ( # noqa: E402 RAW_LIMIT, _encoded, read_graph_archive, validate_expected, @@ -93,8 +93,11 @@ def main(argv: list[str] | None = None) -> int: raise OSError(REFUSAL) sys.stdout.buffer.flush() except OSError: - sys.stderr.write(REFUSAL) - return 1 + # This terminal output failure must not retry buffered stdout at exit. + try: + os.write(2, REFUSAL.encode("ascii")) + finally: + os._exit(1) return 0 diff --git a/tools/artifact_index_export.py b/tools/artifact_index_export.py index 2ce248406..a3f86692a 100644 --- a/tools/artifact_index_export.py +++ b/tools/artifact_index_export.py @@ -7,8 +7,8 @@ # Keep the CLI's own imports read-only even if a caller omits Python's -B flag. sys.dont_write_bytecode = True -# Resolve only the trusted source location lexically, never a caller path. -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +# Anchor imports to the actual script target, including symlink invocation. +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.realpath(__file__)))) from architecture.artifact_index import ( ArtifactIndexRefused, SHA256, STDIN_LIMIT, canonical_json, parse_packet, @@ -48,7 +48,17 @@ def main(): except Exception: sys.stderr.buffer.write(b"ARTIFACT_INDEX_REFUSED\n") return 1 - sys.stdout.buffer.write(output) + try: + written = sys.stdout.buffer.write(output) + if written != len(output): + raise OSError + sys.stdout.buffer.flush() + except OSError: + # This terminal output failure must not retry buffered stdout at exit. + try: + os.write(2, b"ARTIFACT_INDEX_REFUSED\n") + finally: + os._exit(1) return 0 From 22bbffb99f6fa0280ce6f90e1bfd8b5c12250384 Mon Sep 17 00:00:00 2001 From: Dylan Roy <249851642+d6g8k5htny-coder@users.noreply.github.com> Date: Thu, 15 Oct 2026 21:00:55 -0500 Subject: [PATCH 5/5] Give complete serial navigation suites a measured finite job budget Change only Navigation's outer timeout from 10 to 15 minutes. Preserve both full discovery modes, positive-count guards, checks, action pins, triggers, permissions and original-result retention. Two required runs exhausted 600 seconds. Original scoped profile38836910226 completes all832 optimized methods in505.669s; earlier normal210.162s is a different attempt. Their715.831s sum informs capacity, not an observed combined runtime. Required current-head CI still must pass; the diagnostic supplies no merge or science clearance. Scientific effect: NONE. --- .github/workflows/navigation.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/navigation.yml b/.github/workflows/navigation.yml index 586d2a722..c4e157293 100644 --- a/.github/workflows/navigation.yml +++ b/.github/workflows/navigation.yml @@ -10,7 +10,7 @@ permissions: jobs: navigation: runs-on: ubuntu-latest - timeout-minutes: 10 + timeout-minutes: 15 steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97