diff --git a/.agents/skills/README.md b/.agents/skills/README.md index a009e69..387f291 100644 --- a/.agents/skills/README.md +++ b/.agents/skills/README.md @@ -3,7 +3,7 @@ This directory contains repo-local copies of canonical skills from `The-Interdependency/skill-lib`. -Source commit: `c14ee9d500579a4b5d6821f62c9d82ca96e73608` +Source commit: `8de4f12d0f31ff94f41e4a0196c447c0cbe20faf` Repo-local copies are not the source of truth. Edit `skill-lib` first, then propagate from the canonical source. diff --git a/.agents/skills/msdmd/SKILL.md b/.agents/skills/msdmd/SKILL.md index ffa10c4..1a6f0be 100644 --- a/.agents/skills/msdmd/SKILL.md +++ b/.agents/skills/msdmd/SKILL.md @@ -63,7 +63,9 @@ claims to prove those obligations. See referenced from external tooling). - **Field lines**: indented one level beneath the id (two spaces of visible indent inside the comment). Field names are lowercase - snake_case followed by `:` and a value. + snake_case followed by `:` and a value. Digits are allowed after the first + character, so `evidence_sha256` is valid; the first character must be a + lowercase letter or underscore. - **Multiple blocks per file**: a module may declare more than one block, of the same or different types. The parser concatenates entries. diff --git a/.agents/skills/msdmd/parsers/universal.py b/.agents/skills/msdmd/parsers/universal.py index 204f533..e6cb88a 100644 --- a/.agents/skills/msdmd/parsers/universal.py +++ b/.agents/skills/msdmd/parsers/universal.py @@ -100,7 +100,7 @@ def parse_text(text: str, block_name: str, marker: str = "#") -> list[dict]: block_re = _block_regex(block_name, marker) m = re.escape(marker) id_re = re.compile(rf"^\s*{m}\s*id:\s*(?P\S+)\s*$") - field_re = re.compile(rf"^\s*{m}\s+(?P[a-z_]+):\s*(?P.+?)\s*$") + field_re = re.compile(rf"^\s*{m}\s+(?P[a-z_][a-z0-9_]*):\s*(?P.+?)\s*$") entries: list[dict] = [] for block in block_re.finditer(text): @@ -189,7 +189,7 @@ def iter_source_files(path: Path) -> Iterable[Path]: # boundary to literal line 2: # ratios: loc_comments=N:M imports_exports=N:M calls_definitions=N:M RATIO_IDS = ("loc_comments", "imports_exports", "calls_definitions") -_RATIOS_TOKEN_RE = re.compile(r"(?P[a-z_]+)=(?P\S+)") +_RATIOS_TOKEN_RE = re.compile(r"(?P[a-z_][a-z0-9_]*)=(?P\S+)") def _ratios_line_re(marker: str) -> re.Pattern[str]: @@ -253,4 +253,4 @@ def ratios_placement(text: str, marker: str = "#") -> tuple[bool, bool]: last_ok = bool(line_re.match(raw.rstrip())) break return (opening_ok, last_ok) -# ratios: loc_comments=161:57 imports_exports=4:7 calls_definitions=55:10 +# ratios: loc_comments=161:57 imports_exports=4:7 calls_definitions=55:10 \ No newline at end of file diff --git a/.agents/skills/msdmd/parsers/universal.ts b/.agents/skills/msdmd/parsers/universal.ts index 141f46c..67b7b21 100644 --- a/.agents/skills/msdmd/parsers/universal.ts +++ b/.agents/skills/msdmd/parsers/universal.ts @@ -86,7 +86,7 @@ export function parseText( "gm", ); const idRe = new RegExp(`^\\s*${m}\\s*id:\\s*(\\S+)\\s*$`); - const fieldRe = new RegExp(`^\\s*${m}\\s+([a-z_]+):\\s*(.+?)\\s*$`); + const fieldRe = new RegExp(`^\\s*${m}\\s+([a-z_][a-z0-9_]*):\\s*(.+?)\\s*$`); const entries: Entry[] = []; let match: RegExpExecArray | null; @@ -185,7 +185,7 @@ function ratiosLineRe(marker: string): RegExp { export function parseRatios(text: string, marker: string = "#"): Entry[] { const lineRe = ratiosLineRe(marker); - const tokenRe = /([a-z_]+)=(\S+)/g; + const tokenRe = /([a-z_][a-z0-9_]*)=(\S+)/g; const out: Entry[] = []; for (const raw of text.split("\n")) { const lm = lineRe.exec(raw.replace(/\s+$/, "")); diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..cac6e40 --- /dev/null +++ b/.env.example @@ -0,0 +1,7 @@ +# This file intentionally exists as an example only. Real credentials belong in +# your local .env (ignored) or process environment. +OPENAI_API_KEY= +OPENAI_MODEL= +ANTHROPIC_API_KEY= +ANTHROPIC_MODEL= +# Base URL overrides are process-environment-only and are not read from .env. diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 76792dd..5ad6459 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1,16 +1,29 @@ name: ci on: [push, pull_request] +permissions: + contents: read jobs: test: runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ['3.11', '3.12'] steps: - - uses: actions/checkout@v4 - - uses: actions/setup-python@v5 + - uses: actions/checkout@v7 with: - python-version: '3.12' + persist-credentials: false + - uses: actions/setup-python@v7 + with: + python-version: ${{ matrix.python-version }} - run: | python -m venv .venv . .venv/bin/activate python -m pip install -e . python -m unittest discover -s tests python -m pubskill_lib.audit examples/neglected-repo --out /tmp/findings.json + python -m pip wheel . --no-deps -w /tmp/pubskill-wheel + python -m venv /tmp/pubskill-wheel-venv + /tmp/pubskill-wheel-venv/bin/python -m pip install --no-deps /tmp/pubskill-wheel/pubskill_lib-*.whl + cd /tmp + /tmp/pubskill-wheel-venv/bin/python -c "from pubskill_lib import evidence; assert evidence._comment_markers()['.py'] == '#'" diff --git a/.gitignore b/.gitignore index f03a7a0..78d3764 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,8 @@ __pycache__/ *.pyc *.egg-info/ +.env +.env.* +!.env.example + +.examiner-originals-*/ diff --git a/HANDOFF.md b/HANDOFF.md index 304c993..ae61d01 100644 --- a/HANDOFF.md +++ b/HANDOFF.md @@ -15,12 +15,18 @@ VM contract: `HANDOFF.vm.md` A stranger clones this repository, runs the commands in README.md, and gets a findings file for `examples/neglected-repo`. -That is `v0.2.0`. Nothing else is the first tag. A VM receipt is not a tag. +That is the `v0.2.0` release gate. A VM receipt is not a tag, and an implementation on `main` is not a published release. + +## Current closure + +- The clean-checkout gate runs in GitHub CI on Python 3.11 and 3.12. +- `v0.2.0` remains unpublished until the repaired head is merged and the tag is explicitly created. +- Provider base-URL overrides are operator configuration: they may come from the process environment, never from a repository `.env` file. ## Non-goals for this handoff - Do not port the full skill-lib catalog. -- Do not implement `--fix-one` until inspect works (that is `v0.3.0`). +- Do not add `--fix-one` to v0.2; remote execution and repair require later versioned work. - Do not host a SaaS. - Do not rewrite msdmd. - Do not add Way / UCNS / energy text to README. @@ -123,11 +129,12 @@ python -m pubskill_lib.audit PATH --out findings.json v0.2 inspect only: -- read README / pyproject / package.json / lockfiles / `.github/workflows/*` +- read README, `pyproject.toml`, `package.json`, and `.github/workflows/*` - record identity if `.git` exists, else `hmmm` -- flag README links to missing local files -- flag workflows that claim tests but only `echo` -- flag missing advertised scripts +- flag README links to missing local files or paths that escape the repository +- flag workflows that claim tests but only `echo`/no-op +- flag Python console scripts whose modules are missing +- flag direct local `package.json` script targets invoked by node/python/bash/sh when the referenced file is missing or escapes the repository - do not install target deps - do not run target tests @@ -135,7 +142,7 @@ Exit 0 if the tool ran. Do not exit nonzero just because the target repo is sick ## Step D — fixture -`examples/neglected-repo` must contain at least three evidenced defects the CLI will see without `--run`: +`examples/neglected-repo` must contain at least three evidenced defects the CLI will see without execution: 1. README references a file that does not exist 2. CI workflow named like tests that does not invoke a test runner @@ -155,7 +162,7 @@ Compare on `id`, `class`, and `surface`. Optional if time or disk is scarce. Prefer A–D first. -From skill-lib at the SOURCE.md SHA, copy only `msdmd` and `repo-audit-repair` into `.agents/skills/` and write `.agents/skills/README.md` with the SHA. Do not copy the rest of skill-lib. +From skill-lib at the SOURCE.md SHA, copy only `msdmd` and `repo-audit-repair` into `.agents/skills/` and write `.agents/skills/README.md` with the same SHA. Do not copy the rest of skill-lib. ## Step F — tests in this repo @@ -165,20 +172,24 @@ python -m unittest discover -s tests python -m pubskill_lib.audit examples/neglected-repo --out /tmp/out.json ``` -A `.github/workflows/ci.yml` may be added. The VM does not push it unless `PUSH=1`. +CI runs this gate on Python 3.11 and 3.12. The VM does not push unless `PUSH=1`. ## Step G — close the public door Not a VM default. Requires `PUSH=1`. -1. Rewrite README status table: inspect ships. -2. Tag `v0.2.0`. -3. Add this repo to skill-lib consumer list only with `WRITE_CANON=1`. +1. Merge only a head that passes the full gate and required review. +2. Create tag `v0.2.0` explicitly from the accepted release commit. +3. Claim the release as shipped only after the tag exists. +4. Add this repo to skill-lib consumer list only with `WRITE_CANON=1`. ## Done / not done -Done: clean clone → install → unittest → audit fixture → findings.json. +Done at a verified release head: clean clone → install → unittest → audit fixture → findings.json. + +Not done: stars, SaaS, full catalog, remote execution/repair, architectural-drift theater, selling VERIFIED on the zip. -Not done: stars, SaaS, full catalog, architectural-drift theater, selling VERIFIED on the zip. +## hmmm -hmmm — if the fixture expected file was authored by hand and never produced by the CLI, the utility is still fake. +- Remote URL inspection, remote commit selection, target execution, and repair semantics remain outside v0.2 until separately specified. +- The fixture is continuously checked against live CLI output on `id`, `class`, and `surface`; provenance of its original byte-for-byte generation is not retained. diff --git a/README.md b/README.md index 4ff731a..8cea28e 100644 --- a/README.md +++ b/README.md @@ -2,17 +2,17 @@ Public distribution of [skill-lib](https://github.com/The-Interdependency/skill-lib). -Clone this repo when you want a command that inspects a repository and writes findings. The full catalog, org doctrine, and unfinished skills stay in skill-lib. This repo is the subset a stranger can run. +Clone this repo when you want a command that inspects a local repository and writes findings. The full catalog, org doctrine, and unfinished skills stay in skill-lib. This repo is the subset a stranger can run. ## Status -The inspect CLI ships. Clone, run, get findings. +The inspect CLI implementation passes the repository gate; the `v0.2.0` release tag is not published yet. | Claim | State | |---|---| | Canon | `The-Interdependency/skill-lib` | | This repo | distribution + public CLI + fixtures | -| Clone / run / findings | **shipped** — `v0.2` inspect; see `HANDOFF.md` | +| Clone / run / findings | **implementation ready** — release pending | | VM populate | `HANDOFF.vm.md` | | Source pin | `SOURCE.md` | @@ -29,55 +29,59 @@ python -m unittest discover -s tests python -m pubskill_lib.audit examples/neglected-repo --out /tmp/findings.json ``` -Those commands are the definition of done for the first utility tag (`v0.2.0`). They run on a clean clone. +Those commands are the definition of done for the first utility tag (`v0.2.0`). They run in GitHub CI from a clean checkout; publish the tag only after the release gate is explicitly completed. -## What this will do +## Inspect CLI -Inspect one repository path or URL at a named commit and write: +`v0.2` inspects one **local repository path** without executing the target repository: -- identity (remote, commit, dirty state, declared instructions) -- claimed gates vs files that exist -- obvious dependency and docs drift -- findings classified as `defect`, `environment`, `external`, `policy`, or `hmmm` +```bash +python -m pubskill_lib.audit PATH --out findings.json +``` -`--run` is opt-in. `verified` is stamped only on a finding whose gate was re-run. +It writes: -## What this will not do +- identity when the target contains `.git` (remote, commit, dirty state) +- README links to missing local files or paths that escape the repository +- obvious test-workflow no-ops +- Python `pyproject.toml` console scripts whose modules are missing +- direct local `package.json` script targets whose referenced files are missing or escape the repository +- findings classified as `defect`, `environment`, `external`, `policy`, or `hmmm` + +The inspector does **not** yet clone URLs, select remote commits, execute target tests, or repair the target. Those are later capabilities and must not be inferred from the schema. -- audit the whole internet -- execute private CI secrets by default -- rewrite a repo unless `--fix-one` is explicit and bounded -- carry The Interdependent Way, UCNS, or org liturgy on the first screen +Direct script inspection handles interpreter flags and their arguments, such as +`python -W ignore app.py`, `node --require preload.js app.js`, and +`bash -o errexit build.sh`. Shell expansion and indirect launcher commands remain +outside this static inspection contract. ## Repository examiner (BYOK) -The inspect CLI is the first consumer of a repository evidence engine. A -documentation generator builds on that same evidence substrate: +A separate documentation examiner builds on the repository evidence substrate: ```bash -python -m pubskill_lib.examine --repo /path/to/repo --json # dry run -python -m pubskill_lib.examine --repo /path/to/repo --apply --narrate # write + assemble +python -m pubskill_lib.examine --repo /path/to/repo --json +python -m pubskill_lib.examine --repo /path/to/repo --apply --narrate ``` -With `--apply`, the examiner inventories actual code, writes a descriptive -`NARRATIVE` msdmd block into each supported source file (never a `CONTRACT`, -`CHECK`, `CAPABILITY`, or other normative declaration), maintains -shebang-first RATIOS placement, and assembles `docs/examiner/EXAMINER.md` -from the discovered module graph. Narratives are evidence-bound to the source -hash that produced them; changed source without a re-narrate is marked stale. -The tool never leaves the repository boundary it was pointed at. +With `--apply`, the examiner inventories actual code, writes descriptive `NARRATIVE` msdmd blocks into supported source files, maintains source-boundary RATIOS placement, and assembles `docs/examiner/EXAMINER.md`. Narratives are evidence-bound to the source hash that produced them; changed source without a re-narrate is marked stale. The tool never writes outside the repository boundary it was pointed at. -BYOK credentials come from the environment or a `.env` file and are never -printed: +BYOK credentials may come from the process environment or a `.env` file. `.env` files are ignored by this repository. To prevent a target repository from redirecting an operator credential, provider base-URL overrides are accepted only from the process environment, not from `.env`: ```text -OPENAI_API_KEY / OPENAI_BASE_URL / OPENAI_MODEL -ANTHROPIC_API_KEY / ANTHROPIC_BASE_URL / ANTHROPIC_MODEL +OPENAI_API_KEY / OPENAI_MODEL +ANTHROPIC_API_KEY / ANTHROPIC_MODEL +OPENAI_BASE_URL / ANTHROPIC_BASE_URL # process environment only ``` -Multiple providers are attempted sequentially (fallback). Languages without -a shipped ratio computer keep `hmmm` values; unsupported languages are -skipped and reported as `hmmm`, never guessed. +Multiple configured providers are attempted sequentially as fallback. Unsupported or not-faithfully-computable metrics remain `hmmm`; they are not guessed. + +UTF-8 byte-order marks are preserved during source mutation. Files whose protected +coding cookies conflict with the pinned canonical RATIOS placement are left intact +and reported as `hmmm`; this consumer cannot expand canonical placement rules. +If generated prose cannot be encoded in the source encoding, the file is also +left intact with `hmmm`. Existing narratives remain +available in assembled documentation even when a file has no safe mutation adapter. ## License @@ -85,4 +89,26 @@ MPL-2.0, same as skill-lib. Changes to MPL-covered files must be published. ## Canon -Do not add skills here first. Add them in skill-lib, mark them `runnable`, pin the SHA in `SOURCE.md`, then propagate. +Do not add skills here first. Add them in skill-lib, mark them appropriately, pin the SHA in `SOURCE.md`, then propagate the public slice. + +Source updates preserve the original inode in a private `.examiner-originals-*` +directory beside the file, recorded under `preserved_sources` in the apply report. +These recovery directories are excluded from examiner inventory and should not be +committed. The required hard-link operations are probed before source is moved. +Publication briefly withdraws the old name, then creates the updated +name only if it remains absent; it never replaces a competing live file. A +collision or observed write to the retained original records `hmmm`. Already-open +writers can still change the retained original after the operation; stop editors +and generators before applying, then inspect recovery files before removing them. +This protocol preserves bytes; it does not claim a transactional edit shared with +uncooperative writers or uninterrupted availability to concurrent readers. + +Source publication currently requires Linux inode metadata support. Ownership, +permission bits, ACL/xattr/security-label bytes are copied and compared before +publication; an unavailable operation leaves the source intact with `hmmm`. +The generated volume uses a new source inventory after application, so preserved +concurrent edits can mark their older narratives stale. +Direct-script audit is intentionally bounded: unsupported commands, malformed +quoting, shell expansions/control syntax, and working-directory transitions remain +visible as `hmmm`. Later commands after an unresolved shell context inherit that +uncertainty; the tool does not guess their working directory or entrypoint. diff --git a/SOURCE.md b/SOURCE.md index 3c88a77..ed6f935 100644 --- a/SOURCE.md +++ b/SOURCE.md @@ -4,10 +4,10 @@ Canon: https://github.com/The-Interdependency/skill-lib | Field | Value | |---|---| -| Pinned SHA | `be72da66a112d0632fd25480c1f51b6e69db4976` | -| Pinned date | 2026-09-04 | -| Pin meaning | last observed skill-lib `main` when pubskill-lib was created | -| Runnable subset | `msdmd` (runnable), `repo-audit-repair` (contract) — set in skill-lib `skills.json` after the pin | +| Pinned SHA | `8de4f12d0f31ff94f41e4a0196c447c0cbe20faf` | +| Pinned date | 2026-09-11 | +| Pin meaning | exact canonical source used for the vendored public skill slice | +| Runnable subset | `msdmd` (runnable), `repo-audit-repair` (contract) | Update this file in the same commit that propagates vendored skills. diff --git a/examples/neglected-repo/expected-findings.json b/examples/neglected-repo/expected-findings.json index 0a4ca74..40df805 100644 --- a/examples/neglected-repo/expected-findings.json +++ b/examples/neglected-repo/expected-findings.json @@ -1,19 +1,14 @@ { "schema_version": 1, "tool": "pubskill-lib", - "source_pin": "be72da66a112d0632fd25480c1f51b6e69db4976", + "source_pin": "8de4f12d0f31ff94f41e4a0196c447c0cbe20faf", "target": { "path": "examples/neglected-repo", "commit": "hmmm", "remote": "hmmm", "dirty": false }, - "surfaces": [ - "identity", - "docs", - "ci", - "deps" - ], + "surfaces": ["identity", "docs", "ci", "deps"], "findings": [ { "id": "F001", @@ -43,7 +38,5 @@ "verified": false } ], - "hmmm": [ - "target has no .git directory; identity unresolved" - ] + "hmmm": ["target has no .git directory; identity unresolved"] } diff --git a/src/pubskill_lib.egg-info/PKG-INFO b/src/pubskill_lib.egg-info/PKG-INFO deleted file mode 100644 index 4700b39..0000000 --- a/src/pubskill_lib.egg-info/PKG-INFO +++ /dev/null @@ -1,68 +0,0 @@ -Metadata-Version: 2.4 -Name: pubskill-lib -Version: 0.2.0 -Summary: Public inspect CLI for evidence-led repository findings -License: MPL-2.0 -Requires-Python: >=3.11 -Description-Content-Type: text/markdown -License-File: LICENSE -Dynamic: license-file - -# pubskill-lib - -Public distribution of [skill-lib](https://github.com/The-Interdependency/skill-lib). - -Clone this repo when you want a command that inspects a repository and writes findings. The full catalog, org doctrine, and unfinished skills stay in skill-lib. This repo is the subset a stranger can run. - -## Status - -The repository exists. The runnable utility is not in the tree yet. - -| Claim | State | -|---|---| -| Canon | `The-Interdependency/skill-lib` | -| This repo | distribution + public CLI + fixtures | -| Clone / run / findings | **not shipped** — see `HANDOFF.md` | -| VM populate | `HANDOFF.vm.md` | -| Source pin | `SOURCE.md` | - -If a command is not in this README, it is not a public promise. - -## Intended quickstart (target, not current) - -```bash -git clone https://github.com/The-Interdependency/pubskill-lib -cd pubskill-lib -python -m venv .venv && source .venv/bin/activate -python -m pip install -e . -python -m unittest discover -s tests -python -m pubskill_lib.audit examples/neglected-repo --out /tmp/findings.json -``` - -Those commands are the definition of done for the first utility tag (`v0.2.0`). Until they work, do not treat this README as a product page. - -## What this will do - -Inspect one repository path or URL at a named commit and write: - -- identity (remote, commit, dirty state, declared instructions) -- claimed gates vs files that exist -- obvious dependency and docs drift -- findings classified as `defect`, `environment`, `external`, `policy`, or `hmmm` - -`--run` is opt-in. `verified` is stamped only on a finding whose gate was re-run. - -## What this will not do - -- audit the whole internet -- execute private CI secrets by default -- rewrite a repo unless `--fix-one` is explicit and bounded -- carry The Interdependent Way, UCNS, or org liturgy on the first screen - -## License - -MPL-2.0, same as skill-lib. Changes to MPL-covered files must be published. - -## Canon - -Do not add skills here first. Add them in skill-lib, mark them `runnable`, pin the SHA in `SOURCE.md`, then propagate. diff --git a/src/pubskill_lib.egg-info/SOURCES.txt b/src/pubskill_lib.egg-info/SOURCES.txt deleted file mode 100644 index 9f36f7f..0000000 --- a/src/pubskill_lib.egg-info/SOURCES.txt +++ /dev/null @@ -1,13 +0,0 @@ -LICENSE -README.md -pyproject.toml -src/pubskill_lib/__init__.py -src/pubskill_lib/audit.py -src/pubskill_lib/schema.py -src/pubskill_lib.egg-info/PKG-INFO -src/pubskill_lib.egg-info/SOURCES.txt -src/pubskill_lib.egg-info/dependency_links.txt -src/pubskill_lib.egg-info/entry_points.txt -src/pubskill_lib.egg-info/top_level.txt -tests/test_audit_fixture.py -tests/test_schema.py \ No newline at end of file diff --git a/src/pubskill_lib.egg-info/dependency_links.txt b/src/pubskill_lib.egg-info/dependency_links.txt deleted file mode 100644 index 8b13789..0000000 --- a/src/pubskill_lib.egg-info/dependency_links.txt +++ /dev/null @@ -1 +0,0 @@ - diff --git a/src/pubskill_lib.egg-info/entry_points.txt b/src/pubskill_lib.egg-info/entry_points.txt deleted file mode 100644 index 4c26d9f..0000000 --- a/src/pubskill_lib.egg-info/entry_points.txt +++ /dev/null @@ -1,2 +0,0 @@ -[console_scripts] -pubskill-audit = pubskill_lib.audit:main diff --git a/src/pubskill_lib.egg-info/top_level.txt b/src/pubskill_lib.egg-info/top_level.txt deleted file mode 100644 index cc9b8de..0000000 --- a/src/pubskill_lib.egg-info/top_level.txt +++ /dev/null @@ -1 +0,0 @@ -pubskill_lib diff --git a/src/pubskill_lib/_msdmd_universal.py b/src/pubskill_lib/_msdmd_universal.py new file mode 100644 index 0000000..e6cb88a --- /dev/null +++ b/src/pubskill_lib/_msdmd_universal.py @@ -0,0 +1,256 @@ +# ratios: loc_comments=161:57 imports_exports=4:7 calls_definitions=55:10 +"""Universal msdmd parser — pure stdlib. + +Implements the parser contract from ``msdmd/SKILL.md``: extracts every +``# === ===`` … ``# === END ===`` block from +a source file and returns its entries as flat dicts. + +Comment marker is auto-detected by file extension. The block syntax +itself is identical across languages; only the per-line marker changes. +``COMMENT_MARKERS`` is public so runners can distinguish parser support +from their narrower language-specific execution or metric coverage. + +Public API: + + parse_text(text, block_name, marker="#") -> list[dict] + parse_file(path, block_name) -> list[dict] + walk_tree(root, block_name, *, skip=None, extensions=None) -> tuple[annotated, untested] + +RATIOS is the one msdmd declaration that is *not* a fenced block — it is a +single comment line carried on a file's opening and closing source boundaries. +A valid interpreter shebang owns literal line 1, so the opening RATIOS line is +literal line 2 in that case. The reader lives here as a sanctioned extension: + + parse_ratios(text, marker="#") -> list[dict] + parse_ratios_file(path) -> list[dict] + ratios_placement(text, marker="#") -> tuple[opening_ok, closing_ok] + +This module has zero non-stdlib dependencies and is safe to copy +verbatim into any project that wants msdmd support. +""" +from __future__ import annotations +import re +from pathlib import Path +from typing import Iterable + +# extension → line-comment marker. Keep this registry entry-for-entry equivalent +# to universal.ts; tests fail if either parser gains or loses an extension alone. +COMMENT_MARKERS: dict[str, str] = { + ".py": "#", ".pyw": "#", ".pyi": "#", + ".rb": "#", ".rake": "#", ".gemspec": "#", + ".ex": "#", ".exs": "#", + ".sh": "#", ".bash": "#", ".zsh": "#", ".fish": "#", + ".pl": "#", ".pm": "#", ".t": "#", + ".r": "#", ".jl": "#", + ".ps1": "#", ".psm1": "#", ".tcl": "#", + ".raku": "#", ".rakumod": "#", + ".ts": "//", ".tsx": "//", ".mts": "//", ".cts": "//", + ".js": "//", ".jsx": "//", ".mjs": "//", ".cjs": "//", + ".rs": "//", ".go": "//", ".java": "//", + ".c": "//", ".cc": "//", ".cp": "//", ".cpp": "//", + ".cxx": "//", ".c+": "//", ".c++": "//", + ".h": "//", ".hh": "//", ".hp": "//", ".hpp": "//", + ".hxx": "//", ".h+": "//", ".h++": "//", + ".tcc": "//", ".ipp": "//", ".inl": "//", + ".swift": "//", ".kt": "//", ".kts": "//", ".cs": "//", + ".mm": "//", ".scala": "//", ".dart": "//", ".zig": "//", + ".groovy": "//", ".gradle": "//", ".php": "//", + ".sql": "--", ".lua": "--", ".hs": "--", + ".adb": "--", ".ads": "--", ".vhd": "--", ".vhdl": "--", + ".lean": "--", + ".erl": "%", ".hrl": "%", ".prolog": "%", + ".clj": ";", ".cljs": ";", ".cljc": ";", ".bb": ";", + ".lisp": ";", ".lsp": ";", ".cl": ";", + ".scm": ";", ".ss": ";", ".rkt": ";", + ".f": "!", ".for": "!", ".f90": "!", ".f95": "!", + ".f03": "!", ".f08": "!", + ".vb": "'", ".vbs": "'", + ".cob": "*>", ".cbl": "*>", +} + +_DEFAULT_SKIP = ( + "__pycache__", "node_modules", ".git", ".venv", "venv", + "dist", "build", ".next", ".nuxt", "target", ".pytest_cache", + ".mypy_cache", ".tox", +) + + +def marker_for(path: Path) -> str | None: + """Return the comment marker for a file path, or None if unsupported.""" + return COMMENT_MARKERS.get(path.suffix.lower()) + + +def _block_regex(block_name: str, marker: str) -> re.Pattern[str]: + m = re.escape(marker) + name = re.escape(block_name) + return re.compile( + rf"^{m} === {name} ===\s*$(?P.*?)^{m} === END {name} ===\s*$", + re.MULTILINE | re.DOTALL, + ) + + +def parse_text(text: str, block_name: str, marker: str = "#") -> list[dict]: + """Extract every entry from every matching block in ``text``. + + Entries are flat ``dict[str, str]`` keyed by field name. The first + line of an entry must be ``id: ``; subsequent lines until + the next ``id:`` (or block end) carry indented ``: `` + pairs. + """ + block_re = _block_regex(block_name, marker) + m = re.escape(marker) + id_re = re.compile(rf"^\s*{m}\s*id:\s*(?P\S+)\s*$") + field_re = re.compile(rf"^\s*{m}\s+(?P[a-z_][a-z0-9_]*):\s*(?P.+?)\s*$") + + entries: list[dict] = [] + for block in block_re.finditer(text): + current: dict[str, str] | None = None + for line in block.group("body").splitlines(): + line = line.rstrip() + mid = id_re.match(line) + if mid: + if current is not None: + entries.append(current) + current = {"id": mid.group("id")} + continue + if current is None: + continue + mf = field_re.match(line) + if mf: + current[mf.group("key")] = mf.group("val") + if current is not None: + entries.append(current) + return entries + + +def parse_file(path: Path, block_name: str) -> list[dict]: + """Parse a single file. Returns [] if the file's extension has no + known comment marker or if the file can't be read.""" + marker = marker_for(path) + if marker is None: + return [] + try: + return parse_text(path.read_text(encoding="utf-8"), block_name, marker) + except (OSError, UnicodeDecodeError): + return [] + + +def walk_tree( + root: Path, + block_name: str, + *, + skip: Iterable[str] | None = None, + extensions: Iterable[str] | None = None, +) -> tuple[list[tuple[Path, list[dict]]], list[Path]]: + """Walk ``root`` and partition source files into (annotated, untested). + + ``annotated`` is a list of ``(path, entries)`` for every file that + contains at least one entry of ``block_name``. ``untested`` is every + other source file (still filtered by extension and skip-dirs) so + coverage gaps remain observable. + """ + skip_set = set(skip) if skip is not None else set(_DEFAULT_SKIP) + ext_set = ( + set(e.lower() if e.startswith(".") else "." + e.lower() for e in extensions) + if extensions is not None + else set(COMMENT_MARKERS.keys()) + ) + + def iter_source_files(path: Path) -> Iterable[Path]: + if path.name in skip_set: + return + try: + children = sorted(path.iterdir()) + except OSError: + return + for child in children: + if child.is_dir(): + if child.name in skip_set: + continue + yield from iter_source_files(child) + elif child.is_file() and child.suffix.lower() in ext_set: + yield child + + annotated: list[tuple[Path, list[dict]]] = [] + untested: list[Path] = [] + for path in iter_source_files(root): + entries = parse_file(path, block_name) + if entries: + annotated.append((path, entries)) + else: + untested.append(path) + return annotated, untested + + +# --- RATIOS single-line declaration (msdmd extension) -------------------- +# Unlike every other declaration, RATIOS is not fenced. It is a single +# comment line carrying the three canonical ratios at the opening source +# boundary and last non-blank line. A valid line-1 shebang moves the opening +# boundary to literal line 2: +# ratios: loc_comments=N:M imports_exports=N:M calls_definitions=N:M +RATIO_IDS = ("loc_comments", "imports_exports", "calls_definitions") +_RATIOS_TOKEN_RE = re.compile(r"(?P[a-z_][a-z0-9_]*)=(?P\S+)") + + +def _ratios_line_re(marker: str) -> re.Pattern[str]: + return re.compile(rf"^{re.escape(marker)}\s*ratios:\s*(?P.+?)\s*$") + + +def parse_ratios(text: str, marker: str = "#") -> list[dict]: + """Read single-line RATIOS declarations from ``text``. + + RATIOS is not a fenced block: it is one comment line of the form + `` ratios: loc_comments=N:M imports_exports=N:M calls_definitions=N:M`` + placed at the file's opening and closing source boundaries. Returns one + flat ``{"id", "value"}`` dict per (declaration line x ratio token) so a + drift gate can verify every occurrence. + """ + line_re = _ratios_line_re(marker) + out: list[dict] = [] + for raw in text.splitlines(): + lm = line_re.match(raw.rstrip()) + if not lm: + continue + for tm in _RATIOS_TOKEN_RE.finditer(lm.group("body")): + out.append({"id": tm.group("key"), "value": tm.group("val")}) + return out + + +def parse_ratios_file(path: Path) -> list[dict]: + """``parse_ratios`` for a file path (marker auto-detected); [] on error.""" + marker = marker_for(path) + if marker is None: + return [] + try: + return parse_ratios(path.read_text(encoding="utf-8"), marker) + except (OSError, UnicodeDecodeError): + return [] + + +def ratios_placement(text: str, marker: str = "#") -> tuple[bool, bool]: + """Return ``(opening_ratios_ok, closing_ratios_ok)``. + + A non-empty ``#!`` interpreter directive may occupy literal line 1. It is + the only accepted preamble and RATIOS must immediately follow it. + """ + line_re = _ratios_line_re(marker) + lines = text.splitlines() + if not lines: + return (False, False) + has_shebang = lines[0].startswith("#!") and bool(lines[0][2:].strip()) + opening_index = 1 if has_shebang else 0 + opening_ok = ( + len(lines) > opening_index + and bool(line_re.match(lines[opening_index].rstrip())) + ) + if opening_index == 0 and len(lines) > 1: + displaced = lines[1].startswith("#!") and bool(lines[1][2:].strip()) + opening_ok = opening_ok and not displaced + last_ok = False + for raw in reversed(lines): + if raw.strip() == "": + continue + last_ok = bool(line_re.match(raw.rstrip())) + break + return (opening_ok, last_ok) +# ratios: loc_comments=161:57 imports_exports=4:7 calls_definitions=55:10 \ No newline at end of file diff --git a/src/pubskill_lib/assemble.py b/src/pubskill_lib/assemble.py index 7b82d73..3fe2e2a 100644 --- a/src/pubskill_lib/assemble.py +++ b/src/pubskill_lib/assemble.py @@ -1,16 +1,16 @@ -"""Documentation assembly from the discovered module graph. +"""Documentation assembly from discovered repository structure. -Structure follows the architecture actually discovered, not directory depth -mechanically mirrored: +The assembler organizes inventoried files by their actual source paths: volume -> one markdown document per repository - parts -> top-level directories that contain modules - chapters-> nested directories with at least one module - sections-> individual module files + parts -> top-level directories that contain inventoried files + chapters-> nested directories with at least one inventoried file + sections-> individual files lists -> narrative summaries and gap/hmmm roll-ups -Files that exist but produced no narrative appear as lists with their hmmm -state, so gaps remain visible instead of being invented. +This is a filesystem-derived documentation hierarchy, not a dependency graph. +Dependency discovery may inform a future renderer, but this module does not +claim graph semantics it does not consume. """ from __future__ import annotations @@ -48,22 +48,32 @@ def _heading(path: str) -> str: return path.replace("_", " ").replace("-", " ") -def build_structure(evidence_list: list[FileEvidence], narratives: dict[str, dict[str, str]]) -> dict[str, Part]: +def build_structure( + evidence_list: list[FileEvidence], narratives: dict[str, dict[str, str]] +) -> dict[str, Part]: parts: dict[str, Part] = {} for ev in evidence_list: entry = narratives.get(ev.path) summary = entry.get("summary", "") if entry else "" stale = bool(entry) and is_stale(entry, ev.sha256) - hmmm = [h for h in ev.hmmm] + hmmm = list(ev.hmmm) if ev.marker is not None and not entry: hmmm.append("no narrative generated") - section = Section(path=ev.path, heading=_heading(ev.path), summary=summary, stale=stale, hmmm=hmmm) + section = Section( + path=ev.path, + heading=_heading(ev.path), + summary=summary, + stale=stale, + hmmm=hmmm, + ) rel = Path(ev.path) top = rel.parts[0] if len(rel.parts) > 1 else "(root)" part = parts.setdefault(top, Part(title=_heading(top))) chapter_key = "/".join(rel.parts[1:-1]) or "(root)" - chapter = part.chapters.setdefault(chapter_key, Chapter(title=_heading(chapter_key))) + chapter = part.chapters.setdefault( + chapter_key, Chapter(title=_heading(chapter_key)) + ) chapter.sections.append(section) return parts @@ -85,7 +95,11 @@ def _render_section(section: Section) -> list[str]: return lines -def render_markdown(root: Path, evidence_list: list[FileEvidence], narratives: dict[str, dict[str, str]]) -> str: +def render_markdown( + root: Path, + evidence_list: list[FileEvidence], + narratives: dict[str, dict[str, str]], +) -> str: parts = build_structure(evidence_list, narratives) lines = [ "# Repository examination", @@ -112,11 +126,18 @@ def render_markdown(root: Path, evidence_list: list[FileEvidence], narratives: d return "\n".join(lines) -def assemble_docs(root: Path, evidence_list: list[FileEvidence], narratives: dict[str, dict[str, str]], out_dir: Path) -> Path: +def assemble_docs( + root: Path, + evidence_list: list[FileEvidence], + narratives: dict[str, dict[str, str]], + out_dir: Path, +) -> Path: """Write the assembled volume and return its path.""" root = boundary.assert_inside(root, root) out = boundary.assert_inside(root, out_dir) out.mkdir(parents=True, exist_ok=True) volume = out / "EXAMINER.md" - volume.write_text(render_markdown(root, evidence_list, narratives), encoding="utf-8") + volume.write_text( + render_markdown(root, evidence_list, narratives), encoding="utf-8" + ) return volume diff --git a/src/pubskill_lib/audit.py b/src/pubskill_lib/audit.py index 6a1c22d..e0ad99b 100644 --- a/src/pubskill_lib/audit.py +++ b/src/pubskill_lib/audit.py @@ -3,14 +3,15 @@ Usage: python -m pubskill_lib.audit PATH --out findings.json -v0.2 inspect only: read declared files, record identity, flag evidenced -repository defects. Never install target deps, never run target tests. -Exit 0 when the tool ran; exit 3 on tool/schema failures. +v0.2 inspect only: read declared files, record identity, and flag evidenced +repository defects. It never installs target dependencies or runs target tests. """ import argparse import json import re +import shlex +from urllib.parse import unquote, urlsplit import subprocess import sys from pathlib import Path @@ -27,6 +28,52 @@ ECHO_OR_NOOP_PATTERN = re.compile(r"\b(echo|true|exit\s+0|printf)\b", re.IGNORECASE) MARKDOWN_LINK_PATTERN = re.compile(r"\[[^\]]*\]\(([^)]+)\)") PIN_PATTERN = re.compile(r"`([0-9a-f]{40})`") +LOCAL_SCRIPT_INTERPRETERS = {"node", "python", "python3", "bash", "sh"} +NON_FILE_MODES = { + "node": {"-e", "--eval", "-p", "--print", "--run", "-h", "--help", "-v", "--version", "--v8-options", "--completion-bash"}, + "python": {"-c", "-m", "-h", "-?", "--help", "-V", "--version", "--help-env", "--help-xoptions", "--help-all"}, + "python3": {"-c", "-m", "-h", "-?", "--help", "-V", "--version", "--help-env", "--help-xoptions", "--help-all"}, + "bash": {"-c", "--help", "--version"}, + "sh": {"-c", "--help", "--version"}, +} +BOOLEAN_OPTIONS = { + "node": {"--trace-warnings", "--inspect", "--inspect-brk", "--inspect-wait", "--watch", "--test", "--no-warnings", "--enable-source-maps", "--experimental-strip-types", "--experimental-transform-types", "--abort-on-uncaught-exception", "--check", "--interactive", "-c", "-i"}, + "python": {"-" + character for character in "bBdEiIOPqRsSuvx"}, + "python3": {"-" + character for character in "bBdEiIOPqRsSuvx"}, + "bash": {"-" + character for character in "abefhkmnptuvxBCEHPTlirs"} | {"+" + character for character in "abefhkmnptuvxBCEHPTlirs"} | {"--debugger", "--dump-po-strings", "--dump-strings", "--noprofile", "--norc", "--posix", "--restricted", "--verbose", "--login"}, + "sh": {"-" + character for character in "aefnuvxCImps"} | {"+" + character for character in "aefnuvxCImps"}, +} + +VALUE_OPTIONS = { + "python": {"-W", "-X", "--check-hash-based-pycs"}, + "python3": {"-W", "-X", "--check-hash-based-pycs"}, + "bash": {"-o", "+o", "-O", "+O", "--rcfile", "--init-file"}, + "sh": {"-o", "+o"}, + "node": { + "--allow-fs-read", "--allow-fs-write", "--build-snapshot-config", "--conditions", + "--cpu-prof-dir", "--cpu-prof-interval", "--cpu-prof-name", "--debug-port", + "--diagnostic-dir", "--disable-proto", "--disable-warning", "--dns-result-order", + "--env-file", "--env-file-if-exists", "--experimental-config-file", + "--experimental-default-type", "--experimental-loader", "--experimental-sea-config", + "--experimental-package-map", "--experimental-test-tag-filter", "--experimental-test-isolation", "--heap-prof-dir", "--heap-prof-interval", + "--heap-prof-name", "--heapsnapshot-near-heap-limit", "--heapsnapshot-signal", + "--icu-data-dir", "--import", "--input-type", "--inspect-port", + "--inspect-publish-uid", "--loader", "--localstorage-file", "--max-http-header-size", + "--max-old-space-size", "--max-old-space-size-percentage", "--max-semi-space-size", + "--network-family-autoselection-attempt-timeout", "--openssl-config", + "--redirect-warnings", "--report-dir", "--report-directory", "--report-filename", + "--report-signal", "--require", "--secure-heap", "--secure-heap-min", + "--snapshot-blob", "--stack-trace-limit", "--test-concurrency", + "--test-coverage-branches", "--test-coverage-exclude", "--test-coverage-functions", + "--test-coverage-include", "--test-coverage-lines", "--test-name-pattern", + "--test-global-setup", "--test-isolation", "--test-random-seed", "--test-rerun-failures", + "--test-reporter", "--test-reporter-destination", "--test-shard", + "--test-skip-pattern", "--test-timeout", "--title", "--tls-cipher-list", + "--tls-keylog", "--trace-event-categories", "--trace-event-file-pattern", + "--trace-require-module", "--unhandled-rejections", "--use-largepages", + "--v8-pool-size", "--watch-kill-signal", "--watch-path", "-C", "-r", + }, +} class _Sink: @@ -65,12 +112,13 @@ def _git_identity(target): def run(args): try: - return subprocess.run( + result = subprocess.run( ["git", "-C", str(target), *args], capture_output=True, text=True, timeout=10, - ).stdout.strip() + ) + return result.stdout.strip() if result.returncode == 0 else None except (OSError, subprocess.SubprocessError): return None @@ -95,12 +143,13 @@ def _check_readme_links(target, sink): if local.startswith("/"): continue resolved = (readme.parent / local).resolve() + try: + resolved.relative_to(target.resolve()) + except ValueError: + sink.add("docs", f"README link escapes repository: {dest}", f"{name}:{lineno}") + continue if not resolved.exists(): - sink.add( - "docs", - f"README links to {dest}", - f"{name}:{lineno}", - ) + sink.add("docs", f"README links to {dest}", f"{name}:{lineno}") def _check_ci_workflows(target, sink): @@ -135,30 +184,310 @@ def _module_exists(target, module): return any(candidate.exists() for candidate in candidates) -def _check_declared_scripts(target, sink): +def _check_pyproject_scripts(target, sink): pyproject = target / "pyproject.toml" text = _read_text(pyproject) if text is None: return try: import tomllib - except ImportError: # pragma: no cover - requires Python 3.11+ - return - try: data = tomllib.loads(text) - except Exception: + except (ImportError, ValueError): return scripts = (data.get("project") or {}).get("scripts") or {} for name in sorted(scripts): entry = str(scripts[name]) module = entry.split(":", 1)[0].strip() - if not module or _module_exists(target, module): + if module and not _module_exists(target, module): + sink.add( + "deps", + f"declared script {name} points to missing module {module}", + "pyproject.toml [project.scripts]", + ) + + +def _shell_segments(command, separators=";&|\n"): + """Split direct shell commands while retaining quoted/escaped separators.""" + start, quote, escaped, comment = 0, None, False, False + for index, character in enumerate(command): + if comment: + if character == "\n": + comment = False + start = index + 1 continue - sink.add( - "deps", - f"declared script {name} points to missing module {module}", - "pyproject.toml [project.scripts]", - ) + if escaped: + escaped = False + elif character == "\\" and quote != "'": + escaped = True + elif quote: + if character == quote: + quote = None + elif character in {"'", '"'}: + quote = character + elif character == "#" and (index == 0 or command[index - 1] in " \t\r\n;&|()"): + yield command[start:index] + comment = True + elif character in separators: + yield command[start:index] + start = index + 1 + if not comment: + yield command[start:] + + +def _shell_context_gap(segment, *, context_only=False): + """Identify unsupported syntax, separating word expansion from shell structure.""" + quote, escaped, word_start = None, False, 0 + for index, character in enumerate(segment): + if escaped: + if character == "\n": + return "shell line continuation is outside literal-path audit scope" + escaped = False + continue + if quote is None and character.isspace(): + word_start = index + 1 + continue + if character == "\\" and quote != "'": + escaped = True + elif quote == "'": + if character == "'": + quote = None + elif quote is None and (character in "`{}()<>" or segment[index:index + 2] == "$("): + return "shell control syntax is outside literal-path audit scope" + elif not context_only and (character in "$`" or (quote is None and + (character in "*?[]" or (character == "~" and (index == word_start or + (re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*=", segment[word_start:index]) + or (segment[index - 1] == ":" and re.match(r"[A-Za-z_][A-Za-z0-9_]*=", segment[word_start:index])))))))): + return "shell word expansion is outside literal-path audit scope" + elif quote: + if character == quote: + quote = None + elif character in {"'", '"'}: + quote = character + return None + + +def _fixed_word_arity(raw): + """Bounded proof that a supported option value remains one shell argument.""" + quote, escaped = None, False + for index, character in enumerate(raw): + if escaped: + escaped = False + continue + if character == "\\" and quote != "'": + escaped = True + elif quote == "'": + if character == "'": + quote = None + elif quote == '"': + if character == '"': + quote = None + elif character == "$": + # Ordinary quoted scalar expansions have fixed arity. Positional + # arrays and complex parameter/substitution forms stay unresolved. + tail = raw[index:] + if not re.match(r"\$(?:[A-Za-z_][A-Za-z0-9_]*|[0-9*#?$!]|\{[A-Za-z_][A-Za-z0-9_]*\})", tail): + return False + elif character == "`": + return False + elif character in {"'", '"'}: + quote = character + elif character in "$`*?[]{}()<>" or character.isspace(): + return False + return quote is None and not escaped + + +def _attached_option_value(token, interpreter): + if token.startswith("--"): + return "=" in token and token.split("=", 1)[0] in VALUE_OPTIONS[interpreter] + if not token.startswith(("-", "+")): + return False + for position, character in enumerate(token[1:], start=1): + option = token[0] + character + if option in VALUE_OPTIONS[interpreter]: + return position < len(token) - 1 + if option not in BOOLEAN_OPTIONS[interpreter]: + return False + return False + + +def _entrypoint_target(token, entry_url, unresolved=None): + try: + target = token + if entry_url: + parsed = urlsplit(token) + if parsed.scheme == "file" and parsed.netloc not in {"", "localhost"}: + raise ValueError("unsupported file URL authority") + if parsed.scheme not in {"", "file"}: + return None + if re.search(r"%(?![0-9a-fA-F]{2})|%(?:2[fF]|5[cC])", parsed.path): + raise ValueError("invalid or unsupported encoded URL path separator") + target = unquote(parsed.path, errors="strict") + if not target or "\0" in target: + raise ValueError("empty or NUL-containing path") + return target + except (ValueError, UnicodeError) as error: + if unresolved is not None: + unresolved.append(f"unresolved entrypoint {token!r}: {error}") + return None + + +def _local_script_targets(command, unresolved=None): + """Yield direct file operands after documented interpreter options. + + This is a static audit of direct invocations, not a shell evaluator. + Python -W/-X, Bash -o/-O and startup files, and common Node value options + consume their arguments; attached values and -- delimiters are supported. + """ + cwd_unknown = False + for segment in _shell_segments(command): + if not segment.strip(): + continue + gap = _shell_context_gap(segment) + if gap and unresolved is not None: + unresolved.append(gap) + prior_cwd_unknown = cwd_unknown + cwd_unknown = cwd_unknown or bool(_shell_context_gap(segment, context_only=True)) + try: + tokens = shlex.split(segment) + except ValueError as error: + cwd_unknown = True + if unresolved is not None: + unresolved.append(f"unparseable package script: {error}") + continue + raw_words = [word for word in _shell_segments(segment, " \t\r") if word] + while tokens and raw_words and re.match(r"^[A-Za-z_][A-Za-z0-9_]*=", raw_words[0]): + tokens.pop(0) + raw_words.pop(0) + if not tokens: + continue + if _shell_context_gap(raw_words[0]): + cwd_unknown = True # A dynamic command could resolve to a shell builtin. + continue + interpreter = Path(tokens[0]).name + if interpreter in {"cd", "pushd", "popd"}: + cwd_unknown = True + if unresolved is not None: + unresolved.append("working-directory change is outside direct-script audit scope") + continue + if interpreter not in LOCAL_SCRIPT_INTERPRETERS: + cwd_unknown = True + if unresolved is not None: + unresolved.append(f"command is outside direct interpreter audit scope: {tokens[0]!r}") + continue + if prior_cwd_unknown: + if unresolved is not None: + unresolved.append(f"script target after working-directory change is unresolved: {segment.strip()!r}") + continue + non_file_modes = NON_FILE_MODES[interpreter] + entry_url = False + inspecting = False + index = 1 + while index < len(tokens): + token = tokens[index] + if _shell_context_gap(raw_words[index]): + if not _attached_option_value(token, interpreter) or not _fixed_word_arity(raw_words[index]): + break + if interpreter == "node" and token == "inspect" and not inspecting: + inspecting = True + index += 1 + continue + if inspecting and re.fullmatch(r"[^:]+:\d+", token): + break # Remote debugger attachment. + if interpreter == "node" and token in {"--entry-url", "--experimental-entry-url"}: + entry_url = True + index += 1 + continue + if token == "--": + if (index + 1 < len(tokens) and tokens[index + 1] != "-" + and not _shell_context_gap(raw_words[index + 1])): + target = _entrypoint_target(tokens[index + 1], entry_url, unresolved) + if target is not None: + yield target + break + if token == "-" or token.split("=", 1)[0] in non_file_modes: + break + if token in VALUE_OPTIONS[interpreter]: + if index + 1 < len(raw_words) and not _fixed_word_arity(raw_words[index + 1]): + break + index += 2 + continue + if token.startswith("-") or (interpreter in {"bash", "sh"} and token.startswith("+")): + if token.startswith("--"): + option = token.split("=", 1)[0] + if option not in BOOLEAN_OPTIONS[interpreter] and option not in VALUE_OPTIONS[interpreter] and not (inspecting and re.fullmatch(r"--port=\d+", token)): + if unresolved is not None: + unresolved.append(f"interpreter option arity is unresolved: {token!r}") + break + # Short options may be clustered or carry an attached argument. + non_file = False + if not token.startswith("--"): + modes = {mode[1:] for mode in non_file_modes if len(mode) == 2} + if interpreter in {"bash", "sh"}: + modes.add("s") # Read commands from stdin. + for position, option in enumerate(token[1:], start=1): + if (token[0] == "-" and option in modes) or (interpreter == "bash" and option == "s"): + non_file = True + break + if token[0] + option in VALUE_OPTIONS[interpreter]: + if position == len(token) - 1: + if index + 1 < len(raw_words) and not _fixed_word_arity(raw_words[index + 1]): + non_file = True + index += 1 + break + if token[0] + option not in BOOLEAN_OPTIONS[interpreter]: + if unresolved is not None: + unresolved.append(f"interpreter option arity is unresolved: {token!r}") + non_file = True + break + if non_file: + break + index += 1 + continue + target = _entrypoint_target(token, entry_url, unresolved) + if target is not None: + yield target + break + + +def _check_package_scripts(target, sink, unresolved): + package = target / "package.json" + text = _read_text(package) + if text is None: + return + try: + data = json.loads(text) + except json.JSONDecodeError: + sink.add("deps", "package.json is not valid JSON", "package.json") + return + if not isinstance(data, dict): + sink.add("deps", "package.json top level is not an object", "package.json") + return + scripts = data.get("scripts") or {} + if not isinstance(scripts, dict): + return + for name, command in sorted(scripts.items()): + if not isinstance(command, str): + continue + script_unresolved = [] + for raw_path in _local_script_targets(command, script_unresolved): + try: + candidate = Path(raw_path) + local = candidate.resolve() if candidate.is_absolute() else (target / candidate).resolve() + except (ValueError, OSError) as error: + script_unresolved.append(f"unresolved local path {raw_path!r}: {error}") + continue + try: + local.relative_to(target.resolve()) + except ValueError: + sink.add("deps", f"package script {name} escapes repository via {raw_path}", "package.json [scripts]") + continue + if not local.exists(): + sink.add( + "deps", + f"package script {name} points to missing local file {raw_path}", + "package.json [scripts]", + ) + unresolved.extend(f"package script {name}: {item}" for item in script_unresolved) def _read_source_pin(): @@ -169,7 +498,6 @@ def _read_source_pin(): def audit_path(target_path, source_pin=None): - """Inspect one repository path and return a schema-valid document.""" target = Path(target_path) source_pin = source_pin or _read_source_pin() document = new_document(source_pin, target_path) @@ -191,9 +519,14 @@ def audit_path(target_path, source_pin=None): surfaces.append("ci") _check_ci_workflows(target, sink) - if (target / "pyproject.toml").exists() or (target / "package.json").exists(): + has_pyproject = (target / "pyproject.toml").exists() + has_package = (target / "package.json").exists() + if has_pyproject or has_package: surfaces.append("deps") - _check_declared_scripts(target, sink) + if has_pyproject: + _check_pyproject_scripts(target, sink) + if has_package: + _check_package_scripts(target, sink, document["hmmm"]) document["surfaces"] = surfaces document["findings"] = sink.finalize() @@ -203,7 +536,7 @@ def audit_path(target_path, source_pin=None): def main(argv=None): parser = argparse.ArgumentParser(prog="python -m pubskill_lib.audit") - parser.add_argument("path", help="repository path to inspect") + parser.add_argument("path", help="local repository path to inspect") parser.add_argument("--out", required=True, help="findings.json output path") args = parser.parse_args(argv) @@ -214,7 +547,7 @@ def main(argv=None): try: document = audit_path(target) - except Exception as exc: # tool/schema failure, never the target's fault + except Exception as exc: print(f"pubskill_lib.audit: tool failure: {exc}", file=sys.stderr) return 3 diff --git a/src/pubskill_lib/boundary.py b/src/pubskill_lib/boundary.py index 3335390..74be335 100644 --- a/src/pubskill_lib/boundary.py +++ b/src/pubskill_lib/boundary.py @@ -62,7 +62,7 @@ def iter_files(root: Path, skip: set[str] | None = None) -> list[Path]: skip = set(skip if skip is not None else DEFAULT_SKIP) found: list[Path] = [] for dirpath, dirnames, filenames in os.walk(root): - dirnames[:] = sorted(d for d in dirnames if d not in skip) + dirnames[:] = sorted(d for d in dirnames if d not in skip and not d.startswith(".examiner-originals-")) for name in sorted(filenames): path = Path(dirpath) / name try: diff --git a/src/pubskill_lib/evidence.py b/src/pubskill_lib/evidence.py index b6a4f2c..afac379 100644 --- a/src/pubskill_lib/evidence.py +++ b/src/pubskill_lib/evidence.py @@ -1,34 +1,59 @@ """Evidence engine: inventory actual code before describing it. -This layer never infers behavior from names, layout, or convention. It reads -files and records what is actually there: language marker, shebang, existing -msdmd blocks, existing RATIOS lines, content hash, size, and executable bit. +Language comment markers and entry grammar are loaded from the packaged copy of +the pinned canonical msdmd parser; this module does not maintain a second +dialect. ``sha256`` is the stable source evidence hash: generated examiner +NARRATIVE blocks and RATIOS seals are excluded when the source can be decoded. +``raw_sha256`` is always the literal file-byte hash. """ from __future__ import annotations import hashlib +import io import re +import tokenize from dataclasses import dataclass, field from pathlib import Path +from . import _msdmd_universal as _canonical_msdmd from . import boundary - -# extension -> comment marker (same table as canon msdmd) -MARKERS: dict[str, str] = { - ".py": "#", ".rb": "#", ".ex": "#", ".exs": "#", ".sh": "#", - ".ts": "//", ".tsx": "//", ".js": "//", ".jsx": "//", ".mjs": "//", - ".rs": "//", ".go": "//", ".java": "//", ".c": "//", ".cpp": "//", - ".cc": "//", ".h": "//", ".hpp": "//", ".swift": "//", ".kt": "//", - ".sql": "--", ".lua": "--", ".hs": "--", -} +from . import ratios, source_boundaries SHEBANG_RE = re.compile(r"^#!.*$") -_RATIOS_LINE_RE = re.compile(r"^(?:#|//|--)\s*ratios:\s*(.+?)\s*$") -_NARRATIVE_FENCE_RE = re.compile( - r"^(?:#|//|--) === NARRATIVE ===\s*$.*?^(?:#|//|--) === END NARRATIVE ===\s*$", - re.MULTILINE | re.DOTALL, -) +_RATIOS_LINE_RE = re.compile(r"^(?:#|//|--|%|;|!|'|\*>)\s*ratios:\s*(.+?)\s*$") +_PYTHON_SUFFIXES = {".py", ".pyw", ".pyi"} + + +def _msdmd_parser(): + """Return the packaged, source-pinned canonical msdmd parser module.""" + return _canonical_msdmd + + +def _comment_markers() -> dict[str, str]: + """Load COMMENT_MARKERS from the packaged canonical msdmd parser.""" + markers = getattr(_msdmd_parser(), "COMMENT_MARKERS", {}) + return dict(markers) if isinstance(markers, dict) else {} + + +def source_text(text: str, marker: str | None, path: Path | None = None) -> str: + """Return source text with complete generated NARRATIVE/RATIOS metadata removed. + + Trailing blank lines are normalized because RATIOS placement already removes + them. Incomplete NARRATIVE fences are preserved rather than guessed away. + """ + if marker is None: + return text + + lines = text.splitlines() + adapter = ratios.default_adapter_for(path) if path is not None else None + bookends, narrative = source_boundaries.metadata_indices(lines, marker, adapter) + excluded = bookends | narrative + kept = [line for index, line in enumerate(lines) if index not in excluded] + + while kept and not kept[-1].strip(): + kept.pop() + return "\n".join(kept) + ("\n" if kept else "") @dataclass @@ -41,82 +66,92 @@ class FileEvidence: msdmd_blocks: dict[str, list[dict]] = field(default_factory=dict) narrative_entries: list[dict] = field(default_factory=list) sha256: str = "" + raw_sha256: str = "" size: int = 0 executable: bool = False + encoding: str | None = "utf-8" hmmm: list[str] = field(default_factory=list) -def _block_name_re(marker: str) -> re.Pattern[str]: +def _block_names(text: str, marker: str) -> list[str]: + """Return distinct declared block names; canonical parser owns entry grammar.""" m = re.escape(marker) - return re.compile(rf"^{m} === ([A-Z_]+) ===\s*$(?P.*?)^{m} === END \1 ===\s*$", re.MULTILINE | re.DOTALL) - - -def _parse_block_entries(marker: str, body: str) -> list[dict]: - m = re.escape(marker) - id_re = re.compile(rf"^\s*{m}\s*id:\s*(?P\S+)\s*$") - field_re = re.compile(rf"^\s*{m}\s+(?P[a-z_]+):\s*(?P.+?)\s*$") - entries: list[dict] = [] - current: dict[str, str] | None = None - for line in body.splitlines(): - line = line.rstrip() - mid = id_re.match(line) - if mid: - if current is not None: - entries.append(current) - current = {"id": mid.group("id")} - continue - if current is None: - continue - mf = field_re.match(line) - if mf: - current[mf.group("key")] = mf.group("val") - if current is not None: - entries.append(current) - return entries + start_re = re.compile(rf"^{m} === (?P[A-Z_]+) ===\s*$", re.MULTILINE) + return list(dict.fromkeys(match.group("name") for match in start_re.finditer(text))) + + +def _decode_source(path: Path, raw: bytes) -> tuple[str | None, str | None, str | None]: + """Decode source without changing byte identity; honor Python coding cookies.""" + if path.suffix.lower() in _PYTHON_SUFFIXES: + try: + encoding, _ = tokenize.detect_encoding(io.BytesIO(raw).readline) + return raw.decode(encoding), encoding, None + except (LookupError, SyntaxError, UnicodeDecodeError) as exc: + return None, None, f"source encoding unresolved: {exc}" + try: + encoding = "utf-8-sig" if raw.startswith(b"\xef\xbb\xbf") else "utf-8" + return raw.decode(encoding), encoding, None + except UnicodeDecodeError as exc: + return None, None, f"source encoding unresolved: {exc}" def read_evidence(root: Path, path: Path) -> FileEvidence: - """Read one file into evidence. Never raises for unsupported files.""" root = Path(root).resolve() rel = str(path.relative_to(root)) - marker = MARKERS.get(path.suffix.lower()) + marker = _comment_markers().get(path.suffix.lower()) language = path.suffix.lower().lstrip(".") or "unknown" - evidence = FileEvidence(path=rel, language=language, marker=marker) + item = FileEvidence(path=rel, language=language, marker=marker) try: - text = path.read_text(encoding="utf-8", errors="replace") + raw = path.read_bytes() except OSError: - evidence.hmmm.append("unreadable file") - return evidence - evidence.sha256 = hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest() - evidence.size = len(text.encode("utf-8", errors="replace")) + item.hmmm.append("unreadable file") + return item + + item.raw_sha256 = hashlib.sha256(raw).hexdigest() + item.size = len(raw) try: - evidence.executable = bool(path.stat().st_mode & 0o111) + item.executable = bool(path.stat().st_mode & 0o111) except OSError: pass + text, item.encoding, decode_hmmm = _decode_source(path, raw) + if text is None: + item.sha256 = item.raw_sha256 + item.marker = None + item.hmmm.append(decode_hmmm or "source encoding unresolved") + item.hmmm.append("metadata-excluding source hash unavailable; mutation disabled") + return item + + try: + stable_encoded = source_text(text, marker, path).encode("utf-8") + except UnicodeError as error: + item.sha256 = item.raw_sha256 + item.marker = None + item.encoding = None + item.hmmm.append(f"source encoding unresolved while hashing: {error}") + item.hmmm.append("metadata-excluding source hash unavailable; mutation disabled") + return item + item.sha256 = hashlib.sha256(stable_encoded).hexdigest() + first_line = text.splitlines()[0].rstrip() if text.splitlines() else "" if SHEBANG_RE.match(first_line): - evidence.shebang = first_line + item.shebang = first_line if marker is not None: - for raw in text.splitlines(): - if _RATIOS_LINE_RE.match(raw.rstrip()): - evidence.ratios_lines.append(raw.rstrip()) - block_re = _block_name_re(marker) - for match in block_re.finditer(text): - name = match.group(1) - entries = _parse_block_entries(marker, match.group("body")) - evidence.msdmd_blocks.setdefault(name, []).extend(entries) - evidence.narrative_entries = evidence.msdmd_blocks.get("NARRATIVE", []) + lines = text.splitlines() + bookends, narrative_indices = source_boundaries.metadata_indices(lines, marker, ratios.default_adapter_for(path)) + item.ratios_lines = [lines[index].rstrip() for index in sorted(bookends)] + parser = _msdmd_parser() + for name in _block_names(text, marker): + block_text = "\n".join(lines[index] for index in sorted(narrative_indices)) if name == "NARRATIVE" else text + entries = parser.parse_text(block_text, name, marker) + item.msdmd_blocks[name] = entries + item.narrative_entries = item.msdmd_blocks.get("NARRATIVE", []) else: - evidence.hmmm.append(f"unsupported language for msdmd: .{language}") - return evidence + item.hmmm.append(f"unsupported language for msdmd: .{language}") + return item def inventory(root: Path) -> list[FileEvidence]: - """Inventory every regular file inside the boundary.""" root = boundary.assert_inside(root, root) - out: list[FileEvidence] = [] - for path in boundary.iter_files(root): - out.append(read_evidence(root, path)) - return out + return [read_evidence(root, path) for path in boundary.iter_files(root)] diff --git a/src/pubskill_lib/examine.py b/src/pubskill_lib/examine.py index ddbd40b..b7c2926 100644 --- a/src/pubskill_lib/examine.py +++ b/src/pubskill_lib/examine.py @@ -4,19 +4,16 @@ python -m pubskill_lib.examine [--repo PATH] [--apply] [--narrate] [--out DIR] [--env FILE] -Layers are separable: evidence (inventory), model reasoning (narrate), -source mutation (msdmd writer + RATIOS), documentation assembly, and -provider access each live in their own module and can be used directly. - -Default is a dry run: report what would change without writing anything. +Default is a dry run. Evidence, model reasoning, source mutation, +documentation assembly, and provider access remain separate layers. """ from __future__ import annotations import argparse +import hashlib import json import sys -from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone from pathlib import Path @@ -29,71 +26,143 @@ from . import ratios -def _read_text(path: Path) -> str: +def _canonical_artifact(root: Path, path: Path) -> bool: + candidate = path if path.is_absolute() else root / path try: - return path.read_text(encoding="utf-8", errors="replace") - except OSError: - return "" + return (candidate.relative_to(root).as_posix() == "src/pubskill_lib/_msdmd_universal.py" + and not candidate.is_symlink() + and candidate.read_bytes() == Path(evidence._canonical_msdmd.__file__).read_bytes()) + except (OSError, ValueError): + return False def _plan(root: Path, evidence_list: list[evidence.FileEvidence]) -> dict: - supported = [ev for ev in evidence_list if ev.marker is not None] + engine = ratios.RatiosEngine() + supported = [] + unsupported = [] + for ev in evidence_list: + if _canonical_artifact(root, Path(ev.path)): + continue + reason = "; ".join(ev.hmmm) + if ev.marker is None or ev.encoding is None or engine.adapter_for(Path(ev.path)) is None: + reason = reason or "no safe metrics/write adapter" + else: + try: + path = boundary.assert_inside(root, root / ev.path) + engine.place(path.read_bytes().decode(ev.encoding), ev.marker, {}, path) + except (OSError, UnicodeError, ratios.UnsupportedPlacementError) as error: + reason = str(error) + else: + supported.append(ev) + continue + unsupported.append({"path": ev.path, "hmmm": [reason]}) return { "root": str(root), "files": len(evidence_list), "supported_files": len(supported), - "unsupported": [ - {"path": ev.path, "hmmm": ev.hmmm} for ev in evidence_list if ev.marker is None - ], + "preserved_authority": [ev.path for ev in evidence_list if _canonical_artifact(root, Path(ev.path))], + "unsupported": unsupported, "ratios_missing": [ev.path for ev in supported if not ev.ratios_lines], "narrative_present": [ev.path for ev in supported if ev.narrative_entries], } -def _apply(root: Path, evidence_list: list[evidence.FileEvidence], provider_list: list[providers.Provider], narrate: bool) -> tuple[dict, dict]: +def _apply( + root: Path, + evidence_list: list[evidence.FileEvidence], + provider_list: list[providers.Provider], + narrate: bool, +) -> tuple[dict, dict]: narratives: dict[str, dict[str, str]] = {} changed: list[str] = [] + unresolved: dict[str, str] = {} + preserved_authority: list[str] = [] + preserved_sources: dict[str, str] = {} now = datetime.now(timezone.utc).isoformat() + engine = ratios.RatiosEngine() for ev in evidence_list: path = boundary.assert_inside(root, root / ev.path) - text = _read_text(path) - if ev.marker is None: + if ev.narrative_entries: + narratives[ev.path] = ev.narrative_entries[0] + if _canonical_artifact(root, path): + preserved_authority.append(ev.path) + continue + adapter = engine.adapter_for(path) + if ev.marker is None or adapter is None or ev.encoding is None: + unresolved[ev.path] = "; ".join(ev.hmmm) or "no safe metrics/write adapter; mutation skipped" + continue + try: + raw = path.read_bytes() + if hashlib.sha256(raw).hexdigest() != ev.raw_sha256: + unresolved[ev.path] = "source changed after inventory; mutation skipped" + continue + original_text = raw.decode(ev.encoding) + except (OSError, UnicodeError) as exc: + unresolved[ev.path] = f"source unavailable; mutation skipped: {exc}" continue - new_text = text - engine = ratios.RatiosEngine() - values = engine.compute(path, text) - new_text, ratio_changed = engine.place(new_text, ev.marker, values, path) - if ratio_changed: - changed.append(f"{ev.path}:ratios") - + try: + engine.place(original_text, ev.marker, {}, path) + except ratios.UnsupportedPlacementError as error: + unresolved[ev.path] = str(error) + continue + new_text = original_text + file_changes: list[str] = [] + entry = narratives.get(ev.path) if narrate: - result = narrative.narrate_file(ev, text, provider_list, now) - narratives[ev.path] = result.entry + result = narrative.narrate_file( + ev, + evidence.source_text(original_text, ev.marker, path), + provider_list, + now, + ) + entry = result.entry if result.hmmm: result.entry["summary"] = result.entry["summary"] or "hmmm" - else: - existing = ev.narrative_entries[0] if ev.narrative_entries else None - if existing: - narratives[ev.path] = existing - - entry = narratives.get(ev.path) if entry: - new_text, block_changed = msdmd_writer.upsert_narrative(new_text, ev.marker, entry) + new_text, block_changed = msdmd_writer.upsert_narrative( + new_text, ev.marker, entry, path + ) if block_changed: - changed.append(f"{ev.path}:narrative") + file_changes.append(f"{ev.path}:narrative") - if new_text != text: - msdmd_writer.write_text_safely(path, new_text) + values = engine.compute(path, evidence.source_text(new_text, ev.marker, path)) + try: + new_text, ratio_changed = engine.place(new_text, ev.marker, values, path) + except ratios.UnsupportedPlacementError as error: + unresolved[ev.path] = str(error) + continue + if ratio_changed: + file_changes.append(f"{ev.path}:ratios") + + if new_text != original_text: + try: + if path.is_symlink() or boundary.assert_inside(root, path) != path or path.read_bytes() != raw: + unresolved[ev.path] = "source changed during examination; mutation skipped" + continue + original = msdmd_writer.write_text_safely(path, new_text, ev.encoding, expected_raw=raw) + preserved_sources[ev.path] = original.relative_to(root).as_posix() + except msdmd_writer.SourceChangedError as exc: + unresolved[ev.path] = str(exc) + continue + except OSError as exc: + unresolved[ev.path] = f"source unavailable before write; mutation skipped: {exc}" + continue + except UnicodeEncodeError: + unresolved[ev.path] = f"generated text cannot use {ev.encoding}; mutation skipped" + continue + if entry: + narratives[ev.path] = entry + changed.extend(file_changes) - return narratives, {"changed": changed, "narrated": len(narratives)} + return narratives, {"changed": changed, "narrated": len(narratives), "hmmm": unresolved, "preserved_authority": preserved_authority, "preserved_sources": preserved_sources} def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog="python -m pubskill_lib.examine") parser.add_argument("--repo", help="repository path (default: current directory)") - parser.add_argument("--apply", action="store_true", help="write msdmd + RATIOS + docs (default: dry run)") + parser.add_argument("--apply", action="store_true", help="write msdmd + RATIOS + docs") parser.add_argument("--narrate", action="store_true", help="call BYOK providers for narratives") parser.add_argument("--env", default=".env", help=".env file for BYOK credentials") parser.add_argument("--out", default="docs/examiner", help="documentation output directory") @@ -127,15 +196,24 @@ def main(argv: list[str] | None = None) -> int: return 3 narratives, report = _apply(root, evidence_list, provider_list, args.narrate) + # Rendering observes live source after every write/skip, so an old summary + # cannot retain a current marker after a concurrent edit was preserved. + evidence_list = evidence.inventory(root) + narratives = {ev.path: ev.narrative_entries[0] for ev in evidence_list if ev.narrative_entries} + report["narrated"] = len(narratives) out_dir = boundary.assert_inside(root, root / args.out) volume = assemble.assemble_docs(root, evidence_list, narratives, out_dir) if args.json: - print(json.dumps({"changed": report["changed"], "volume": str(volume)}, indent=2)) + print(json.dumps({**report, "volume": str(volume)}, indent=2)) else: print(f"applied: {len(report['changed'])} writes") for change in report["changed"]: print(f" {change}") + for path, original in report["preserved_sources"].items(): + print(f" preserved source: {path}: {original}") + for path, reason in report["hmmm"].items(): + print(f" hmmm: {path}: {reason}") print(f"assembled: {volume}") return 0 diff --git a/src/pubskill_lib/msdmd_writer.py b/src/pubskill_lib/msdmd_writer.py index b29480d..3c4464a 100644 --- a/src/pubskill_lib/msdmd_writer.py +++ b/src/pubskill_lib/msdmd_writer.py @@ -1,25 +1,21 @@ """msdmd NARRATIVE writer: generated explanation is descriptive evidence. -The NARRATIVE block is a normal msdmd fenced block, never a CONTRACT, CHECK, -CAPABILITY, OWNERS, DOCS, or other normative declaration. It carries the -content hash that produced it so stale narrative can be detected. +NARRATIVE placement shares the language opening-boundary rules used by the +RATIOS engine so metadata never displaces an interpreter or protected source +prologue. """ from __future__ import annotations -import hashlib -import re from pathlib import Path +import os +import sys +import tempfile -NARRATIVE_BLOCK = "NARRATIVE" - +from . import ratios +from . import source_boundaries -def _fence_re(marker: str) -> re.Pattern[str]: - m = re.escape(marker) - return re.compile( - rf"^{m} === {NARRATIVE_BLOCK} ===\s*$.*?^{m} === END {NARRATIVE_BLOCK} ===\s*$", - re.MULTILINE | re.DOTALL, - ) +NARRATIVE_BLOCK = "NARRATIVE" def _block_lines(marker: str, entry: dict[str, str]) -> list[str]: @@ -30,39 +26,120 @@ def _block_lines(marker: str, entry: dict[str, str]) -> list[str]: return lines +def _without_narrative_lines(text: str, marker: str, adapter=None) -> list[str]: + """Remove complete NARRATIVE blocks without creating phantom blank lines.""" + lines = text.splitlines() + _, indices = source_boundaries.metadata_indices(lines, marker, adapter) + kept = [line for index, line in enumerate(lines) if index not in indices] + + while kept and not kept[-1].strip(): + kept.pop() + return kept + + def narrative_id(sha256: str) -> str: - """Stable, refactor-safe entry id derived from the evidence hash.""" return f"examiner_{sha256[:16]}" -def upsert_narrative(text: str, marker: str, entry: dict[str, str]) -> tuple[str, bool]: - """Replace any existing NARRATIVE blocks with ``entry``, keeping the - shebang first and the ratios bookends where they were.""" - fence = _fence_re(marker) - body = fence.sub("", text).rstrip("\n") +def upsert_narrative( + text: str, + marker: str, + entry: dict[str, str], + path: Path | None = None, +) -> tuple[str, bool]: + """Replace NARRATIVE blocks without crossing the protected opening boundary.""" + adapter = ratios.RatiosEngine().adapter_for(path) if path is not None else None + lines = _without_narrative_lines(text, marker, adapter) block = "\n".join(_block_lines(marker, entry)) - lines = body.splitlines() + insert_at = ratios.opening_index(lines, adapter) ratios_prefix = f"{marker} ratios:" - insert_at = 0 - if lines and lines[0].startswith("#!"): - insert_at = 1 - if len(lines) > 1 and lines[1].lstrip().startswith(ratios_prefix): - insert_at = 2 - elif lines and lines[0].lstrip().startswith(ratios_prefix): - insert_at = 1 + if insert_at < len(lines) and lines[insert_at].lstrip().startswith(ratios_prefix): + insert_at += 1 + lines.insert(insert_at, block) - new_text = "\n".join(lines) + "\n" + new_text = "\n".join(lines) + ("\n" if lines else "") return new_text, new_text != text -def write_text_safely(path: Path, new_text: str) -> None: - """Write text without changing the file's executable bit.""" - mode = None +class SourceChangedError(RuntimeError): + """The live source no longer matches the inventoried bytes.""" + + +def _inode_metadata(path: Path) -> tuple: + if not sys.platform.startswith("linux") or not hasattr(os, "listxattr"): + raise OSError("source inode metadata verification is unsupported on this platform") + info = path.stat(follow_symlinks=False) + attributes = {name: os.getxattr(path, name, follow_symlinks=False) + for name in os.listxattr(path, follow_symlinks=False)} + return info.st_uid, info.st_gid, info.st_mode & 0o7777, attributes + + +def _copy_inode_metadata(path: Path, metadata: tuple) -> None: + uid, gid, mode, attributes = metadata + os.chown(path, uid, gid, follow_symlinks=False) + path.chmod(mode) + for name in set(os.listxattr(path, follow_symlinks=False)) - attributes.keys(): + os.removexattr(path, name, follow_symlinks=False) + for name, value in attributes.items(): + os.setxattr(path, name, value, follow_symlinks=False) + if _inode_metadata(path) != metadata: + raise OSError("source inode metadata cannot be preserved exactly") + + +def write_text_safely(path: Path, new_text: str, encoding: str = "utf-8", *, expected_raw: bytes | None = None) -> Path: + """Publish without replacing a live name; retain the original inode. + + There is a short absent-name interval. Publication uses link's atomic + no-replace guarantee. Open writers keep their original inode in recovery + storage, which is deliberately never deleted by this operation. + """ + encoded = new_text.encode(encoding) + if path.is_symlink(): + raise SourceChangedError("source became a symlink; mutation skipped") + raw = path.read_bytes() if expected_raw is None else expected_raw + metadata = _inode_metadata(path) + # A fresh private directory prevents a preexisting recovery path from + # redirecting writes. The caller reports its path; inventory skips it. + recovery = Path(tempfile.mkdtemp(prefix=".examiner-originals-", dir=path.parent)) + original = recovery / "original" + candidate = recovery / "candidate" + moved = False try: - mode = path.stat().st_mode & 0o777 - except OSError: - pass - path.write_text(new_text, encoding="utf-8") - if mode is not None: - path.chmod(mode) + candidate.write_bytes(encoded) + _copy_inode_metadata(candidate, metadata) + # Both candidate publication and original restoration require links. + # Probe the same files/directory before withdrawing the live name. + for source in (candidate, path): + probe = recovery / "link-probe" + os.link(source, probe, follow_symlinks=False) + probe.unlink() + if path.is_symlink() or path.read_bytes() != raw or _inode_metadata(path) != metadata: + raise SourceChangedError("source changed before metadata publication") + os.rename(path, original) + moved = True + if original.is_symlink() or original.read_bytes() != raw or _inode_metadata(original) != metadata: + raise SourceChangedError(f"source changed during publication; preserved at {original}") + try: + os.link(candidate, path) # Atomic create-if-absent; never replace a competing edit. + except FileExistsError as error: + raise SourceChangedError(f"competing source preserved; prior inode at {original}") from error + if original.read_bytes() != raw: + raise SourceChangedError(f"open writer changed original inode; inspect preserved source at {original}") + return original + except BaseException as failure: + if moved: + try: + os.link(original, path, follow_symlinks=False) + except FileExistsError: + pass # Preserve the live name and the recovery inode independently. + except OSError as error: + raise SourceChangedError(f"source retained at {original}; restore failed: {error}") from error + if isinstance(failure, OSError): + raise OSError(f"publication failed; original retained at {original}: {failure}") from failure + raise + finally: + candidate.unlink(missing_ok=True) + (recovery / "link-probe").unlink(missing_ok=True) + if not moved: + recovery.rmdir() diff --git a/src/pubskill_lib/narrative.py b/src/pubskill_lib/narrative.py index 243c406..a266194 100644 --- a/src/pubskill_lib/narrative.py +++ b/src/pubskill_lib/narrative.py @@ -1,8 +1,7 @@ """Model reasoning layer: evidence-bound narrative generation. -A narrative is descriptive evidence about one file, bound to the exact -content hash that produced it. If the file changes and the narrative does -not, the narrative is stale and the assembler flags it. +A narrative is descriptive evidence about one file, bound to the exact content +hash that produced it. Successful generations also record provider and model. """ from __future__ import annotations @@ -46,7 +45,6 @@ def narrate_file( provider_list: list[providers.Provider], now: str, ) -> Narrative: - """Return an evidence-bound narrative entry for one file.""" entry = { "id": narrative_id(ev.sha256), "summary": "", @@ -68,12 +66,13 @@ def narrate_file( entry["provider"] = "none" else: try: - summary, name = providers.chat_with_fallback( + summary, name, model = providers.chat_with_fallback( provider_list, SYSTEM_PROMPT, _user_prompt(ev, text) ) entry["summary"] = " ".join(summary.split()) entry["provider"] = name - except Exception as exc: # noqa: BLE001 - provider failure is hmmm + entry["model"] = model or "hmmm" + except Exception as exc: # provider failure remains visible as hmmm hmmm.append(f"model reasoning failed: {type(exc).__name__}") if hmmm: @@ -83,5 +82,4 @@ def narrate_file( def is_stale(entry: dict[str, str], current_sha256: str) -> bool: - """A narrative is stale when its evidence hash no longer matches.""" return entry.get("evidence_sha256") != current_sha256 diff --git a/src/pubskill_lib/providers.py b/src/pubskill_lib/providers.py index 13eab5e..59ae102 100644 --- a/src/pubskill_lib/providers.py +++ b/src/pubskill_lib/providers.py @@ -1,13 +1,8 @@ -"""BYOK provider access. Credentials come from environment/.env and are -never printed, logged, or returned by this layer. +"""BYOK provider access. -Supported providers (configurable through environment): - - OPENAI_API_KEY, OPENAI_BASE_URL (default api.openai.com/v1), OPENAI_MODEL - ANTHROPIC_API_KEY, ANTHROPIC_BASE_URL (default api.anthropic.com), ANTHROPIC_MODEL - -Multiple providers are attempted sequentially by default; callers may also -drive them concurrently across files. +Credentials may come from the process environment or a local .env file. Base +URL overrides are process-environment-only so a target repository cannot pair +an operator's ambient API key with a repository-controlled endpoint. """ from __future__ import annotations @@ -19,6 +14,7 @@ DEFAULT_OPENAI_BASE = "https://api.openai.com/v1" DEFAULT_ANTHROPIC_BASE = "https://api.anthropic.com" +_BASE_URL_KEYS = {"OPENAI_BASE_URL", "ANTHROPIC_BASE_URL"} def load_dotenv(path: str | Path = ".env") -> dict[str, str]: @@ -41,8 +37,10 @@ def load_dotenv(path: str | Path = ".env") -> dict[str, str]: def env_with_dotenv(path: str | Path = ".env") -> dict[str, str]: - """Merged os.environ plus .env values (os.environ wins).""" + """Merge .env with os.environ; base URL overrides come only from os.environ.""" merged = dict(load_dotenv(path)) + for key in _BASE_URL_KEYS: + merged.pop(key, None) merged.update(os.environ) return merged @@ -63,15 +61,15 @@ def __init__(self, name: str, env: dict[str, str]): self.name = name self.env = env self.key = env.get(self.key_env(), "") - self.model = env.get(self.model_env(), self.default_model()) + self.model = env.get(self.model_env()) or self.default_model() - def key_env(self) -> str: # pragma: no cover - overridden + def key_env(self) -> str: raise NotImplementedError - def model_env(self) -> str: # pragma: no cover - overridden + def model_env(self) -> str: raise NotImplementedError - def default_model(self) -> str: # pragma: no cover - overridden + def default_model(self) -> str: raise NotImplementedError def configured(self) -> bool: @@ -80,7 +78,7 @@ def configured(self) -> bool: def describe(self) -> str: return f"{self.name} model={self.model or 'hmmm'} key={_mask(self.key) if self.key else 'absent'}" - def chat(self, system: str, user: str) -> str: # pragma: no cover - overridden + def chat(self, system: str, user: str) -> str: raise NotImplementedError @@ -100,10 +98,7 @@ def default_model(self) -> str: return "gpt-4o-mini" def chat(self, system: str, user: str) -> str: - headers = { - "Content-Type": "application/json", - "Authorization": f"Bearer {self.key}", - } + headers = {"Content-Type": "application/json", "Authorization": f"Bearer {self.key}"} payload = { "model": self.model, "messages": [ @@ -148,17 +143,19 @@ def chat(self, system: str, user: str) -> str: def configured_providers(env: dict[str, str]) -> list[Provider]: - """Return providers with credentials, in stable order.""" + """Return configured providers in stable fallback order.""" providers = [OpenAIProvider(env), AnthropicProvider(env)] - return [p for p in providers if p.configured()] + return [provider for provider in providers if provider.configured()] -def chat_with_fallback(providers: list[Provider], system: str, user: str) -> tuple[str, str]: - """Try providers sequentially. Returns (text, provider_name) or raises.""" +def chat_with_fallback( + providers: list[Provider], system: str, user: str +) -> tuple[str, str, str]: + """Try providers sequentially. Return (text, provider_name, model).""" errors: list[str] = [] for provider in providers: try: - return provider.chat(system, user), provider.name - except Exception as exc: # noqa: BLE001 - boundary to hmmm, never to crash + return provider.chat(system, user), provider.name, provider.model + except Exception as exc: # boundary failure remains visible without leaking secrets errors.append(f"{provider.name}: {type(exc).__name__}") raise RuntimeError("; ".join(errors) or "no providers configured") diff --git a/src/pubskill_lib/ratios.py b/src/pubskill_lib/ratios.py index 7df538a..1e08484 100644 --- a/src/pubskill_lib/ratios.py +++ b/src/pubskill_lib/ratios.py @@ -27,6 +27,7 @@ import re from pathlib import Path +from . import source_boundaries RATIO_IDS = ("loc_comments", "imports_exports", "calls_definitions") SHEBANG_RE = re.compile(r"^#!.*$") @@ -135,10 +136,11 @@ def render_ratios_line(marker: str, values: dict[str, str]) -> str: return f"{marker} ratios: {body}" -def strip_ratios_lines(text: str, marker: str) -> list[str]: - """Return the file's lines with every ratios line removed.""" - line_re = _ratios_line_re(marker) - return [line for line in text.splitlines() if not line_re.match(line.rstrip())] +def strip_ratios_lines(text: str, marker: str, adapter=None) -> list[str]: + """Remove only the reserved bookends, preserving source-literal contents.""" + lines = text.splitlines() + indices, _ = source_boundaries.metadata_indices(lines, marker, adapter) + return [line for index, line in enumerate(lines) if index not in indices] def opening_index(lines: list[str], adapter: LanguageRatioAdapter | None) -> int: @@ -148,11 +150,11 @@ def opening_index(lines: list[str], adapter: LanguageRatioAdapter | None) -> int adapter, the default rule applies: a shebang stays first and the seal follows it. """ - if adapter is not None: - protected = adapter.opening_boundary(lines) - else: - protected = [0] if lines and SHEBANG_RE.match(lines[0].rstrip()) else [] - return max(protected, default=-1) + 1 + return source_boundaries.opening_index(lines, adapter) + + +class UnsupportedPlacementError(ValueError): + """Protected source lines conflict with the pinned canonical seal boundary.""" def place_ratios( @@ -166,8 +168,7 @@ def place_ratios( Returns ``(new_text, changed)``. Existing ratios lines are removed and re-placed. The closing line is the last non-blank line. """ - line_re = _ratios_line_re(marker) - lines = [raw for raw in text.splitlines() if not line_re.match(raw.rstrip())] + lines = strip_ratios_lines(text, marker, adapter) opening = render_ratios_line(marker, values) lines.insert(opening_index(lines, adapter), opening) @@ -179,6 +180,9 @@ def place_ratios( new_text = "\n".join(lines) if lines: new_text += "\n" + from . import _msdmd_universal + if _msdmd_universal.ratios_placement(new_text, marker) != (True, True): + raise UnsupportedPlacementError("protected source prologue conflicts with pinned canonical RATIOS placement; mutation skipped") return new_text, new_text != text diff --git a/src/pubskill_lib/source_boundaries.py b/src/pubskill_lib/source_boundaries.py new file mode 100644 index 0000000..be76b25 --- /dev/null +++ b/src/pubskill_lib/source_boundaries.py @@ -0,0 +1,40 @@ +"""Identify examiner metadata only at its reserved source placement boundaries. + +Fence-shaped text elsewhere remains source data, including inside multiline +strings. Entry parsing remains owned by the packaged canonical msdmd parser. +""" +from __future__ import annotations + +import re + + +def opening_index(lines, adapter=None): + if adapter is not None: + protected = adapter.opening_boundary(lines) + else: + protected = [0] if lines and lines[0].startswith("#!") else [] + return max(protected, default=-1) + 1 + + +def metadata_indices(lines, marker, adapter=None): + """Return reserved RATIOS indices and one complete opening NARRATIVE span.""" + ratio_line = re.compile(rf"^{re.escape(marker)}\s*ratios:\s*.+?\s*$") + ratios = set() + opening = opening_index(lines, adapter) + if opening < len(lines) and ratio_line.fullmatch(lines[opening]): + ratios.add(opening) + opening += 1 + closing = len(lines) - 1 + while closing >= 0 and not lines[closing].strip(): + closing -= 1 + if closing >= 0 and ratio_line.fullmatch(lines[closing]): + ratios.add(closing) + narrative = set() + if opening < len(lines) and lines[opening].rstrip() == f"{marker} === NARRATIVE ===": + for index in range(opening + 1, len(lines)): + if not lines[index].startswith(marker): + break + if lines[index].rstrip() == f"{marker} === END NARRATIVE ===": + narrative.update(range(opening, index + 1)) + break + return ratios, narrative diff --git a/tests/test_examine.py b/tests/test_examine.py index e3f3d43..669b8b1 100644 --- a/tests/test_examine.py +++ b/tests/test_examine.py @@ -1,3 +1,4 @@ +import json import os import shutil import subprocess @@ -87,16 +88,11 @@ def test_python_opening_boundary_respects_encoding_header(self): self.assertEqual([0, 1], adapter.opening_boundary(lines)) self.assertEqual(2, ratios.opening_index(lines, adapter)) - new, _ = ratios.place_ratios( - "\n".join(lines) + "\n", - "#", - {"loc_comments": "1:0", "imports_exports": "1:0", "calls_definitions": "0:0"}, - adapter, - ) - out = new.splitlines() - self.assertTrue(out[0].startswith("#!")) - self.assertIn("coding", out[1]) - self.assertTrue(out[2].startswith("# ratios:")) + with self.assertRaises(ratios.UnsupportedPlacementError): + ratios.place_ratios( + "\n".join(lines) + "\n", "#", + {"loc_comments": "1:0", "imports_exports": "1:0", "calls_definitions": "0:0"}, adapter, + ) def test_find_internal_dependencies_python(self): from pubskill_lib import ratios_adapters @@ -169,9 +165,22 @@ def test_dry_run_reports_without_writing(self): self.assertEqual(0, result.returncode, result.stderr) self.assertEqual(before, (self.root / "tool.py").read_text()) + def test_json_apply_reports_recovery_paths(self): + (self.root / "recovery_probe.py").write_text("print('fresh source')\n") + result = self._run("--apply", "--json") + self.assertEqual(0, result.returncode, result.stderr) + report = json.loads(result.stdout) + self.assertTrue(report["preserved_sources"]) + for relative in report["preserved_sources"].values(): + self.assertTrue((self.root / relative).is_file()) + def test_apply_writes_ratios_and_assembles_docs(self): + (self.root / "recovery_probe.py").write_text("print('fresh source')\n") + shell_before = (self.root / "run.sh").read_bytes() result = self._run("--apply", "--out", "docs/examiner") self.assertEqual(0, result.returncode, result.stderr) + self.assertIn("preserved source:", result.stdout) + self.assertIn(".examiner-originals-", result.stdout) tool = (self.root / "tool.py").read_text().splitlines() self.assertTrue(tool[0].startswith("#!")) @@ -180,7 +189,8 @@ def test_apply_writes_ratios_and_assembles_docs(self): shell = (self.root / "run.sh").read_text().splitlines() self.assertTrue(shell[0].startswith("#!")) - self.assertTrue(shell[1].startswith("# ratios: loc_comments=hmmm")) + self.assertEqual(shell_before, (self.root / "run.sh").read_bytes()) + self.assertFalse(any("ratios:" in line for line in shell)) ts = (self.root / "lib" / "util.ts").read_text().splitlines() self.assertTrue(ts[0].startswith("// ratios: loc_comments=")) @@ -201,4 +211,4 @@ def test_apply_writes_ratios_and_assembles_docs(self): if __name__ == "__main__": - unittest.main() + unittest.main() \ No newline at end of file diff --git a/tests/test_idempotence.py b/tests/test_idempotence.py new file mode 100644 index 0000000..1aaed4d --- /dev/null +++ b/tests/test_idempotence.py @@ -0,0 +1,61 @@ +import tempfile +import unittest +from pathlib import Path + +from pubskill_lib import evidence, examine, narrative + + +class ExaminerIdempotenceTests(unittest.TestCase): + def test_generated_metadata_does_not_stale_its_own_narrative(self): + class FakeProvider: + name = "fake" + model = "model-1" + + def chat(self, system, user): + return "Prints a greeting." + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source = root / "tool.py" + source.write_text( + "#!/usr/bin/env python3\n" + "print('hi')\n", + encoding="utf-8", + ) + + first_evidence = evidence.inventory(root) + _, first_report = examine._apply(root, first_evidence, [FakeProvider()], True) + first_output = source.read_text(encoding="utf-8") + self.assertTrue(first_report["changed"]) + self.assertEqual((True, True), evidence._canonical_msdmd.ratios_placement(first_output)) + + second_evidence = evidence.inventory(root) + self.assertEqual(1, len(second_evidence)) + entry = second_evidence[0].narrative_entries[0] + self.assertFalse(narrative.is_stale(entry, second_evidence[0].sha256)) + + _, second_report = examine._apply(root, second_evidence, [], False) + second_output = source.read_text(encoding="utf-8") + self.assertEqual(first_output, second_output) + self.assertEqual([], second_report["changed"]) + + def test_source_hash_excludes_generated_narrative_and_ratios(self): + plain = "print('hi')\n" + decorated = ( + "# ratios: loc_comments=1:0 imports_exports=0:0 calls_definitions=1:0\n" + "# === NARRATIVE ===\n" + "# id: examiner_x\n" + "# summary: Prints hi.\n" + "# evidence_sha256: x\n" + "# === END NARRATIVE ===\n" + "print('hi')\n" + "# ratios: loc_comments=1:0 imports_exports=0:0 calls_definitions=1:0\n" + ) + self.assertEqual( + evidence.source_text(plain, "#"), + evidence.source_text(decorated, "#"), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_provenance.py b/tests/test_provenance.py new file mode 100644 index 0000000..79202c7 --- /dev/null +++ b/tests/test_provenance.py @@ -0,0 +1,52 @@ +import json +import re +import unittest +from pathlib import Path + + +REPO = Path(__file__).resolve().parents[1] +PIN_RE = re.compile(r"`([0-9a-f]{40})`") + + +def _source_pin() -> str: + source = (REPO / "SOURCE.md").read_text(encoding="utf-8") + match = PIN_RE.search(source) + if match is None: + raise AssertionError("SOURCE.md has no pinned SHA") + return match.group(1) + + +class PublicationProvenanceTests(unittest.TestCase): + def test_source_pin_matches_vendored_skill_manifest(self): + vendored = (REPO / ".agents" / "skills" / "README.md").read_text(encoding="utf-8") + vendored_pin = PIN_RE.search(vendored) + self.assertIsNotNone(vendored_pin) + self.assertEqual(_source_pin(), vendored_pin.group(1)) + + def test_fixture_source_pin_matches_publication_pin(self): + expected = json.loads( + (REPO / "examples" / "neglected-repo" / "expected-findings.json").read_text( + encoding="utf-8" + ) + ) + self.assertEqual(_source_pin(), expected["source_pin"]) + + def test_packaged_parser_matches_vendored_canonical_bytes(self): + vendored = REPO / ".agents" / "skills" / "msdmd" / "parsers" / "universal.py" + packaged = REPO / "src" / "pubskill_lib" / "_msdmd_universal.py" + self.assertEqual(vendored.read_bytes(), packaged.read_bytes()) + + def test_local_secret_files_are_ignored(self): + ignore = (REPO / ".gitignore").read_text(encoding="utf-8").splitlines() + self.assertIn(".env", ignore) + self.assertIn(".env.*", ignore) + self.assertIn("*.egg-info/", ignore) + + def test_readme_does_not_claim_unpublished_v020_tag(self): + readme = (REPO / "README.md").read_text(encoding="utf-8") + self.assertIn("release pending", readme) + self.assertNotIn("**shipped** — `v0.2`", readme) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_repairs.py b/tests/test_repairs.py new file mode 100644 index 0000000..c03c999 --- /dev/null +++ b/tests/test_repairs.py @@ -0,0 +1,648 @@ +"""Regression tests for audit findings repaired on 2026-09-10.""" + +import hashlib +import json +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from pubskill_lib import audit, evidence, examine, msdmd_writer, narrative, providers, ratios + + +class CredentialBoundaryTests(unittest.TestCase): + def test_dotenv_cannot_override_provider_base_url(self): + with tempfile.TemporaryDirectory() as tmp: + env_file = Path(tmp) / ".env" + env_file.write_text( + "OPENAI_API_KEY=repo-key\nOPENAI_BASE_URL=https://attacker.invalid/v1\n", + encoding="utf-8", + ) + with patch.dict(os.environ, {"OPENAI_API_KEY": "operator-key"}, clear=True): + merged = providers.env_with_dotenv(env_file) + self.assertEqual("operator-key", merged["OPENAI_API_KEY"]) + self.assertNotIn("OPENAI_BASE_URL", merged) + + def test_process_environment_may_set_base_url(self): + with tempfile.TemporaryDirectory() as tmp: + env_file = Path(tmp) / ".env" + env_file.write_text("OPENAI_BASE_URL=https://attacker.invalid/v1\n", encoding="utf-8") + with patch.dict( + os.environ, + { + "OPENAI_API_KEY": "operator-key", + "OPENAI_BASE_URL": "https://operator.example/v1", + }, + clear=True, + ): + merged = providers.env_with_dotenv(env_file) + self.assertEqual("https://operator.example/v1", merged["OPENAI_BASE_URL"]) + + def test_blank_model_override_uses_provider_default(self): + openai = providers.OpenAIProvider({"OPENAI_API_KEY": "key", "OPENAI_MODEL": ""}) + anthropic = providers.AnthropicProvider({"ANTHROPIC_API_KEY": "key", "ANTHROPIC_MODEL": ""}) + self.assertEqual(openai.default_model(), openai.model) + self.assertEqual(anthropic.default_model(), anthropic.model) + + +class NarrativeBoundaryTests(unittest.TestCase): + def test_fence_shaped_literal_data_remains_source(self): + class FakeProvider: + name, model = "fake", "model-1" + def chat(self, system, user): + return "Stores a literal string." + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + path = root / "literal.py" + original = 'payload = """\n# === NARRATIVE ===\n# id: literal_data\n# summary: alpha\n# === END NARRATIVE ===\n# ratios: loc_comments=1:2 imports_exports=3:4 calls_definitions=5:6\n"""\n' + path.write_text(original) + before = evidence.read_evidence(root, path) + self.assertEqual([], before.narrative_entries) + path.write_text(original.replace("alpha", "beta")) + self.assertNotEqual(before.sha256, evidence.read_evidence(root, path).sha256) + path.write_text(original) + examine._apply(root, [before], [FakeProvider()], True) + namespace = {} + exec(compile(path.read_bytes(), str(path), "exec"), namespace) + expected = {} + exec(compile(original, str(path), "exec"), expected) + self.assertEqual(expected["payload"], namespace["payload"]) + after = evidence.read_evidence(root, path) + self.assertEqual(before.sha256, after.sha256) + self.assertEqual(1, len(after.narrative_entries)) + first = path.read_bytes() + examine._apply(root, [after], [], False) + self.assertEqual(first, path.read_bytes()) + + def test_provider_cannot_overwrite_a_concurrent_source_edit(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + path = root / "concurrent.py" + path.write_text("print('old')\n") + concurrent = b"print('concurrent edit')\n" + class EditingProvider: + name, model = "fake", "model-1" + def chat(self, system, user): + path.write_bytes(concurrent) + return "Prints old." + _, report = examine._apply(root, [evidence.read_evidence(root, path)], [EditingProvider()], True) + self.assertEqual(concurrent, path.read_bytes()) + self.assertEqual([], report["changed"]) + self.assertIn("concurrent.py", report["hmmm"]) + + def test_conditional_publication_preserves_competing_writes(self): + import os + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "source.py" + original = b"original\n" + concurrent = b"concurrent\n" + path.write_bytes(original) + link = os.link + def competing_write(source, target, **kwargs): + if Path(source).name == "candidate" and Path(target) == path: + path.write_bytes(concurrent) + return link(source, target, **kwargs) + with patch("pubskill_lib.msdmd_writer.os.link", side_effect=competing_write): + with self.assertRaises(msdmd_writer.SourceChangedError): + msdmd_writer.write_text_safely(path, "new\n", expected_raw=original) + self.assertEqual(concurrent, path.read_bytes()) + self.assertEqual([original], [p.read_bytes() for p in Path(tmp).glob(".examiner-originals-*/original")]) + + def test_failed_publication_preserves_source_and_external_hardlinks(self): + import os + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "source.py" + original = b"original\n" + path.write_bytes(original) + alias = Path(tmp) / "external.py" + alias.hardlink_to(path) + link = os.link + def fail_candidate(source, target, **kwargs): + if Path(source).name == "candidate" and Path(target) == path: + raise OSError("publication failed") + return link(source, target, **kwargs) + with patch("pubskill_lib.msdmd_writer.os.link", side_effect=fail_candidate): + with self.assertRaises(OSError): + msdmd_writer.write_text_safely(path, "new\n", expected_raw=original) + self.assertEqual(original, path.read_bytes()) + with path.open("r+b") as writer: + recovery = msdmd_writer.write_text_safely(path, "new\n", expected_raw=original) + writer.write(b"late edit") + writer.truncate() + self.assertEqual(b"late edit", recovery.read_bytes()) + self.assertEqual(b"late edit", alias.read_bytes()) + self.assertEqual(b"new\n", path.read_bytes()) + + def test_unrelated_similarly_named_source_is_not_canonical_authority(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + for name in ("other/pubskill_lib/_msdmd_universal.py", "src/pubskill_lib/_msdmd_universal.py"): + path = root / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("print('ordinary user code')\n") + item = evidence.read_evidence(root, path) + self.assertEqual(1, examine._plan(root, [item])["supported_files"]) + _, report = examine._apply(root, [item], [], False) + self.assertTrue(report["changed"]) + self.assertEqual([], report["preserved_authority"]) + + def test_candidate_setup_failure_cleans_recovery_storage(self): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "source.py" + path.write_text("original\n") + for operation in ("write_bytes", "chmod"): + with patch.object(Path, operation, side_effect=OSError("quota")): + with self.assertRaises(OSError): + msdmd_writer.write_text_safely(path, "new\n") + self.assertEqual("original\n", path.read_text()) + self.assertEqual([], list(Path(tmp).glob(".examiner-originals-*"))) + + def test_publication_preserves_inode_metadata_or_refuses_to_move(self): + import os + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "source.py" + path.write_text("original\n") + path.chmod(0o751) + os.setxattr(path, "user.pubskill_test", b"retained") + original_metadata = msdmd_writer._inode_metadata(path) + msdmd_writer.write_text_safely(path, "new\n") + self.assertEqual(original_metadata, msdmd_writer._inode_metadata(path)) + with patch("pubskill_lib.msdmd_writer.os.setxattr", side_effect=OSError("metadata denied")): + with self.assertRaises(OSError): + msdmd_writer.write_text_safely(path, "third\n") + self.assertEqual("new\n", path.read_text()) + self.assertEqual(original_metadata, msdmd_writer._inode_metadata(path)) + + def test_assembled_narrative_is_stale_after_preserving_concurrent_edit(self): + import contextlib, io + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + path = root / "source.py" + original = "print('old')\n" + path.write_text(original) + ev = evidence.read_evidence(root, path) + decorated, _ = msdmd_writer.upsert_narrative(original, "#", {"id": "old_narrative", "summary": "Old summary", "evidence_sha256": ev.sha256}, path) + path.write_text(decorated) + class EditingProvider: + name, model = "fake", "model" + def chat(self, system, user): + path.write_text(path.read_text().replace("print('old')", "print('edited')")) + return "Generated stale summary" + with patch("pubskill_lib.providers.configured_providers", return_value=[EditingProvider()]), contextlib.redirect_stdout(io.StringIO()): + self.assertEqual(0, examine.main(["--repo", str(root), "--apply", "--narrate"])) + output = (root / "docs/examiner/EXAMINER.md").read_text() + self.assertIn("Old summary", output) + self.assertIn("> stale:", output) + self.assertIn("print('edited')", path.read_text()) + + def test_apply_preserves_packaged_canonical_parser(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + parser = root / "src/pubskill_lib/_msdmd_universal.py" + parser.parent.mkdir(parents=True) + original = Path(evidence._canonical_msdmd.__file__).read_bytes() + parser.write_bytes(original) + ev = evidence.read_evidence(root, parser) + _, report = examine._apply(root, [ev], [], True) + self.assertEqual(original, parser.read_bytes()) + self.assertEqual([], report["changed"]) + self.assertEqual([ev.path], report["preserved_authority"]) + self.assertEqual(0, examine._plan(root, [ev])["supported_files"]) + + def test_narrative_preserves_python_shebang_and_encoding_header(self): + text = ( + "#!/usr/bin/env python3\n" + "# -*- coding: latin-1 -*-\n" + "# ratios: loc_comments=1:1 imports_exports=0:0 calls_definitions=1:0\n" + "print('hi')\n" + "# ratios: loc_comments=1:1 imports_exports=0:0 calls_definitions=1:0\n" + ) + entry = { + "id": "examiner_abc", + "summary": "Prints hi.", + "evidence_sha256": "abc", + "model": "none", + "provider": "none", + "generated_at": "now", + "stale": "false", + } + new, changed = msdmd_writer.upsert_narrative(text, "#", entry, Path("tool.py")) + lines = new.splitlines() + self.assertTrue(changed) + self.assertTrue(lines[0].startswith("#!")) + self.assertIn("coding:", lines[1]) + self.assertTrue(lines[2].startswith("# ratios:")) + self.assertEqual("# === NARRATIVE ===", lines[3]) + + def test_successful_narrative_records_model(self): + class FakeProvider: + name = "fake" + model = "model-1" + + def chat(self, system, user): + return "Does one thing." + + ev = evidence.FileEvidence(path="x.py", language="py", marker="#", sha256="abc") + result = narrative.narrate_file(ev, "print('x')\n", [FakeProvider()], "now") + self.assertEqual("fake", result.entry["provider"]) + self.assertEqual("model-1", result.entry["model"]) + + def test_generated_narrative_does_not_change_ratios(self): + class FakeProvider: + name = "fake" + model = "model-1" + + def chat(self, system, user): + return "Generated prose mentions fake_call() and comments." + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + path = root / "x.py" + original = "print('x')\n" + path.write_text(original, encoding="utf-8") + ev = evidence.read_evidence(root, path) + expected = ratios.RatiosEngine().compute(path, original) + examine._apply(root, [ev], [FakeProvider()], True) + written = evidence.read_evidence(root, path) + self.assertTrue(written.ratios_lines) + for key, value in expected.items(): + self.assertIn(f"{key}={value}", written.ratios_lines[0]) + + def test_apply_skips_parser_supported_language_without_safe_ratio_adapter(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + path = root / "index.php" + original = "