diff --git a/.github/workflows/gh494-publish-regression.yml b/.github/workflows/gh494-publish-regression.yml index 13383543..040bf04e 100644 --- a/.github/workflows/gh494-publish-regression.yml +++ b/.github/workflows/gh494-publish-regression.yml @@ -1,33 +1,43 @@ -name: Prepare reviewed atomic publication fix +name: Lexical publication regression on: push: branches: [main] - paths: ['.github/workflows/gh494-publish-regression.yml'] + paths: + - 'src/indexer/**' + - 'src/search/tantivy.rs' + - 'src/search/quill_bridge.rs' + - 'tests/gh494_publish_resume.rs' + - 'tests/util/**' + - 'Cargo.toml' + - 'Cargo.lock' + - 'rust-toolchain.toml' + - '.github/workflows/gh494-publish-regression.yml' + workflow_dispatch: -# Temporary preparation of one reviewed patch. No branch/ref writes occur here. -# The explicit source commit restores the read-only regression workflow. +# The reviewed edit is now on main. This workflow only reads and tests code; +# it cannot commit, push, or replay the one-time source transformation. permissions: contents: read concurrency: - group: gh494-atomic-certification-preparation - cancel-in-progress: false + group: gh494-regressions-${{ github.ref }} + cancel-in-progress: true jobs: - prepare: - if: github.repository == 'Dicklesworthstone/coding_agent_session_search' && github.ref == 'refs/heads/main' + publication-regressions: runs-on: ubuntu-24.04 - timeout-minutes: 15 - permissions: - contents: write + timeout-minutes: 60 env: + CARGO_BUILD_JOBS: '2' + CARGO_PROFILE_TEST_DEBUG: '0' + CARGO_PROFILE_DEV_DEBUG: '0' RUST_MIN_STACK: '134217728' steps: - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 with: persist-credentials: false - - name: Read pinned toolchain + - name: Read the repository's dated toolchain id: toolchain shell: bash run: | @@ -40,33 +50,21 @@ jobs: - uses: dtolnay/rust-toolchain@881ba7bf39a41cda34ac9e123fb41b44ed08232f with: toolchain: ${{ steps.toolchain.outputs.name }} - components: rustfmt - - name: Apply only the reviewed patch to the reviewed blobs + components: rustfmt, clippy + - uses: Swatinem/rust-cache@ad397744b0d591a723ab90405b7247fac0e6b8db + - name: Record exact source and compiler shell: bash - env: - GH_TOKEN: ${{ github.token }} run: | - set -euo pipefail - mkdir -p "$RUNNER_TEMP/gh494-prepared" - test "$(git hash-object src/indexer/mod.rs)" = 512485aa2cad27f886326ae486bde1c16649ad24 - test "$(git hash-object src/indexer/lexical_publish.rs)" = 55e3f8bcba0dbf52b2b62c7aa7feafcd5935e0b2 - test "$(git hash-object src/indexer/gh494_tests.rs)" = 00fa2e10355bbfc572b6ecba071e6254d530b0cc - test "$(git hash-object tests/gh494_publish_resume.rs)" = 97360316a5f0b300a8ee9ac0e777dfc27a3cdcb4 - gh api repos/${{ github.repository }}/git/blobs/d88d7587ebbcad8f7c7adc86efc7ce2429a087f9 --jq .content | base64 --decode > "$RUNNER_TEMP/reviewed.patch.gz" - test "$(git hash-object "$RUNNER_TEMP/reviewed.patch.gz")" = d88d7587ebbcad8f7c7adc86efc7ce2429a087f9 - gzip --decompress --stdout "$RUNNER_TEMP/reviewed.patch.gz" > "$RUNNER_TEMP/reviewed.patch" - test "$(git hash-object "$RUNNER_TEMP/reviewed.patch")" = 028f6f48a3e64795982ac1f6d44f102a5ee103e2 - git apply --check --whitespace=error "$RUNNER_TEMP/reviewed.patch" - git apply --whitespace=error "$RUNNER_TEMP/reviewed.patch" - rustfmt --edition 2024 --config skip_children=true src/indexer/lexical_publish.rs src/indexer/gh494_tests.rs tests/gh494_publish_resume.rs - rustfmt --edition 2024 --config skip_children=true --check src/indexer/lexical_publish.rs src/indexer/gh494_tests.rs tests/gh494_publish_resume.rs - rustfmt --edition 2024 --config skip_children=true --emit stdout src/indexer/mod.rs > "$RUNNER_TEMP/mod.parsed.rs" - git diff --check - git diff --binary > "$RUNNER_TEMP/gh494-prepared/reviewed.patch" - printf '%s\n' src/indexer/gh494_tests.rs src/indexer/lexical_publish.rs src/indexer/mod.rs tests/gh494_publish_resume.rs > "$RUNNER_TEMP/expected-paths" - git diff --name-only | sort | diff - "$RUNNER_TEMP/expected-paths" - { git rev-parse HEAD; rustc --version --verbose; cargo --version; } > "$RUNNER_TEMP/gh494-prepared/source-and-compiler.txt" - - name: Run the exact diagnostic helper tests + mkdir -p "$RUNNER_TEMP/gh494-results" + { + git rev-parse HEAD + git hash-object src/indexer/mod.rs src/indexer/lexical_publish.rs src/indexer/gh494_tests.rs tests/gh494_publish_resume.rs + rustc --version --verbose + cargo --version + } | tee "$RUNNER_TEMP/gh494-results/source-and-compiler.txt" + rustfmt --edition 2024 --check src/indexer/lexical_publish.rs src/indexer/gh494_tests.rs + rustfmt --edition 2024 --config skip_children=true --check tests/gh494_publish_resume.rs + - name: Execute the exact diagnostic helper independently of the monolithic test build shell: bash run: | python3 - <<'PY' @@ -80,36 +78,38 @@ jobs: (root / 'Cargo.toml').write_text(manifest) (root / 'src/lib.rs').write_text('mod lexical_publish;\n') shutil.copyfile('src/indexer/lexical_publish.rs', root / 'src/lexical_publish.rs') + print('Exact diagnostic-helper dependency versions:', deps) PY - cargo test --manifest-path "$RUNNER_TEMP/gh494-phase-harness/Cargo.toml" gh494_ -- --nocapture 2>&1 | tee "$RUNNER_TEMP/gh494-prepared/diagnostics.log" - - name: Store immutable candidate blobs without updating any ref + cargo test --manifest-path "$RUNNER_TEMP/gh494-phase-harness/Cargo.toml" gh494_ -- --nocapture 2>&1 | tee "$RUNNER_TEMP/gh494-results/diagnostics.log" + - name: Require all three real rebuild regressions to be present shell: bash - env: - GH_TOKEN: ${{ github.token }} run: | - python3 - <<'PY' - import base64, hashlib, json, os, pathlib, subprocess - paths = ['src/indexer/mod.rs', 'src/indexer/lexical_publish.rs', 'src/indexer/gh494_tests.rs', 'tests/gh494_publish_resume.rs'] - report = {'base_commit': os.environ['GITHUB_SHA'], 'patch_blob': 'd88d7587ebbcad8f7c7adc86efc7ce2429a087f9', 'blobs': {}} - for path in paths: - data = pathlib.Path(path).read_bytes() - expected = hashlib.sha1(b'blob ' + str(len(data)).encode() + b'\0' + data).hexdigest() - payload = json.dumps({'content': base64.b64encode(data).decode(), 'encoding': 'base64'}) - response = subprocess.run(['gh', 'api', '--method', 'POST', f"repos/{os.environ['GITHUB_REPOSITORY']}/git/blobs", '--input', '-'], input=payload, text=True, capture_output=True, check=True) - sha = json.loads(response.stdout)['sha'] - assert sha == expected, (path, sha, expected) - before = subprocess.check_output(['git', 'rev-parse', f'HEAD:{path}'], text=True).strip() - report['blobs'][path] = {'before_sha': before, 'sha': sha} - print(path, sha) - output = pathlib.Path(os.environ['RUNNER_TEMP']) / 'gh494-prepared/blobs.json' - output.write_text(json.dumps(report, indent=2) + '\n') - PY - tar -czf "$RUNNER_TEMP/gh494-prepared/candidate-source.tar.gz" src/indexer/mod.rs src/indexer/lexical_publish.rs src/indexer/gh494_tests.rs tests/gh494_publish_resume.rs - - name: Preserve source, diff, blob identities, and validation receipts + cargo test --locked --lib gh494_ -- --list 2>&1 | tee "$RUNNER_TEMP/gh494-results/test-list.log" + for name in \ + gh494_publish_failure_at_eof_is_finalized_on_retry_instead_of_returning_success \ + gh494_zero_conversations_still_publish_a_readable_completed_generation \ + gh494_completed_marker_cannot_hide_a_live_document_count_mismatch + do + grep -F "$name: test" "$RUNNER_TEMP/gh494-results/test-list.log" + done + - name: Execute publication refusal, EOF retry, empty archive, and completion validation + shell: bash + run: | + cargo test --locked --lib gh494_ -- --test-threads=1 --nocapture 2>&1 | tee "$RUNNER_TEMP/gh494-results/rebuild-regressions.log" + - name: Execute CLI EOF resume and repeated read-only searches + shell: bash + run: | + cargo test --locked --test gh494_publish_resume -- --test-threads=1 --nocapture 2>&1 | tee "$RUNNER_TEMP/gh494-results/cli-resume-regressions.log" + - name: Retain hollow-generation and publication rollback coverage + shell: bash + run: | + cargo test --locked --lib hollow -- --test-threads=1 2>&1 | tee "$RUNNER_TEMP/gh494-results/hollow.log" + cargo test --locked --lib publish_staged_lexical_index -- --test-threads=1 2>&1 | tee "$RUNNER_TEMP/gh494-results/publication-rollback.log" + - name: Preserve validation receipts, including failures if: always() uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 with: - name: gh494-atomic-publication-candidate-${{ github.sha }} - path: ${{ runner.temp }}/gh494-prepared/ + name: gh494-publication-results-${{ github.sha }} + path: ${{ runner.temp }}/gh494-results/ if-no-files-found: warn retention-days: 14 diff --git a/.github/workflows/logical-archive.yml b/.github/workflows/logical-archive.yml index ad08a5ac..8c1165db 100644 --- a/.github/workflows/logical-archive.yml +++ b/.github/workflows/logical-archive.yml @@ -4,8 +4,6 @@ on: branches: [main] paths: - 'src/logical_archive/**' - - 'src/bookmarks.rs' - - 'tests/bookmarks_cli.rs' - 'src/main.rs' - 'src/franken_sync.rs' - 'src/storage/**' @@ -21,33 +19,6 @@ concurrency: group: logical-archive-${{ github.ref }} cancel-in-progress: true jobs: - format-reviewed-files: - runs-on: ubuntu-24.04 - timeout-minutes: 10 - steps: - - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 - with: - ref: ${{ github.sha }} - persist-credentials: false - - name: Format copies of the reviewed sources without changing the checkout - shell: bash - run: | - set -euo pipefail - evidence="$RUNNER_TEMP/reviewed-format" - mkdir -p "$evidence" - git rev-parse HEAD > "$evidence/source.txt" - rustfmt --version > "$evidence/toolchain.txt" - for file in src/bookmarks.rs src/logical_archive/export.rs src/logical_archive/export_tests.rs tests/cli_logical_archive.rs; do - mkdir -p "$evidence/$(dirname "$file")" - sha256sum "$file" >> "$evidence/input-sha256.txt" - rustfmt --edition 2024 --config skip_children=true --emit stdout "$file" > "$evidence/$file" - done - - uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 - with: - name: reviewed-format-${{ github.sha }} - path: ${{ runner.temp }}/reviewed-format/ - if-no-files-found: error - retention-days: 7 locked-cass: runs-on: ubuntu-24.04 timeout-minutes: 45 @@ -72,7 +43,7 @@ jobs: rustc --version | tee "$RUNNER_TEMP/archive-evidence/toolchain.txt" sha256sum Cargo.lock | tee "$RUNNER_TEMP/archive-evidence/lockfile.txt" # Parse without changing source or presenting formatting as runtime proof. - rustfmt --edition 2024 --emit stdout src/logical_archive/mod.rs src/bookmarks.rs tests/cli_logical_archive.rs > "$RUNNER_TEMP/archive-evidence/formatted-source.txt" + rustfmt --edition 2024 --emit stdout src/logical_archive/mod.rs tests/cli_logical_archive.rs > "$RUNNER_TEMP/archive-evidence/formatted-source.txt" - name: Run locked production archive tests shell: bash run: | @@ -96,34 +67,10 @@ jobs: from pathlib import Path text = (Path(os.environ['RUNNER_TEMP']) / 'archive-evidence/cli.log').read_text() counts = re.findall(r'test result: ok\. (\d+) passed; 0 failed;', text) - if not counts or max(map(int, counts)) < 5: - raise SystemExit('Expected all five real-binary archive regressions, including FIFO refusal') + if not counts or max(map(int, counts)) < 3: + raise SystemExit('Expected all three real-binary archive regressions') PY sha256sum target/debug/cass | tee "$RUNNER_TEMP/archive-evidence/binary.txt" - - name: Run bookmark storage and no-clobber export regressions - if: ${{ always() && !cancelled() }} - shell: bash - run: | - set -euo pipefail - mkdir -p "$RUNNER_TEMP/archive-evidence" - cargo test --locked --lib bookmarks::tests:: -- --nocapture --test-threads=2 2>&1 | tee "$RUNNER_TEMP/archive-evidence/bookmarks.log" - python3 - <<'PY' - import os, re - from pathlib import Path - text = (Path(os.environ['RUNNER_TEMP']) / 'archive-evidence/bookmarks.log').read_text() - counts = re.findall(r'test result: ok\. (\d+) passed; 0 failed;', text) - if not counts or max(map(int, counts)) < 20: - raise SystemExit('Expected all 20 bookmark regressions') - PY - cargo test --locked --test bookmarks_cli -- --nocapture --test-threads=2 2>&1 | tee "$RUNNER_TEMP/archive-evidence/bookmarks-cli.log" - python3 - <<'PY' - import os, re - from pathlib import Path - text = (Path(os.environ['RUNNER_TEMP']) / 'archive-evidence/bookmarks-cli.log').read_text() - counts = re.findall(r'test result: ok\. (\d+) passed; 0 failed;', text) - if not counts or max(map(int, counts)) == 0: - raise SystemExit('Expected nonempty bookmark command regression results') - PY - uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 if: always() with: diff --git a/.github/workflows/message-coordinate-regression.yml b/.github/workflows/message-coordinate-regression.yml index e186fd27..ae5b6585 100644 --- a/.github/workflows/message-coordinate-regression.yml +++ b/.github/workflows/message-coordinate-regression.yml @@ -17,8 +17,6 @@ on: - docs/reference/QUICK_REFERENCE.md - tests/cli_robot.rs - tests/cli_view_budget.rs - - tests/util/timeout.rs - - tests/spec_field_mask_contract.rs - tests/cli_dispatch_coverage.rs - tests/spec_view_expand_error_envelopes.rs - tests/golden/robot/capabilities.json.golden @@ -36,54 +34,6 @@ concurrency: cancel-in-progress: true jobs: - subprocess-contract: - if: github.repository == 'Dicklesworthstone/coding_agent_session_search' - strategy: - fail-fast: false - matrix: - runner: [ubuntu-24.04-arm, ubuntu-latest, macos-14] - runs-on: ${{ matrix.runner }} - timeout-minutes: 10 - steps: - - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 - with: - ref: ${{ github.sha }} - persist-credentials: false - - uses: dtolnay/rust-toolchain@881ba7bf39a41cda34ac9e123fb41b44ed08232f - with: - components: rustfmt, clippy - # Compile the exact shared helper, not a copy or simulation. This small - # consumer can diagnose cleanup failures independently of a full CASS build. - - name: Prepare the exact-source subprocess contract consumer - shell: bash - run: | - set -euo pipefail - git rev-parse HEAD - git hash-object tests/util/timeout.rs - rustc --version - python3 - <<'PY' - import json, os, pathlib - root = pathlib.Path(os.environ['RUNNER_TEMP']) / 'cass-subprocess-contract' - root.mkdir() - source = pathlib.Path(os.environ['GITHUB_WORKSPACE']) / 'tests/util/timeout.rs' - manifest = '\n'.join([ - '[package]', 'name = "cass-subprocess-contract"', 'version = "0.0.0"', - 'edition = "2024"', '[lib]', 'path = ' + json.dumps(str(source)), - '[dev-dependencies]', 'tempfile = "=3.27.0"', '', - ]) - (root / 'Cargo.toml').write_text(manifest) - with open(os.environ['GITHUB_ENV'], 'a') as output: - output.write('CONTRACT_MANIFEST=' + str(root / 'Cargo.toml') + '\n') - PY - cargo generate-lockfile --manifest-path "$RUNNER_TEMP/cass-subprocess-contract/Cargo.toml" - - name: Verify timeout, orphan-pipe and harness-death cleanup - shell: bash - run: | - set -euo pipefail - cargo test --locked --manifest-path "$CONTRACT_MANIFEST" -- --nocapture - cargo clippy --locked --manifest-path "$CONTRACT_MANIFEST" --all-targets -- -D warnings - rustfmt --edition 2024 --check tests/util/timeout.rs - verify: if: github.repository == 'Dicklesworthstone/coding_agent_session_search' runs-on: ubuntu-24.04-arm @@ -105,26 +55,19 @@ jobs: - name: Record exact source and compiler identity run: git rev-parse HEAD && rustc --version && cargo --version - uses: Swatinem/rust-cache@ad397744b0d591a723ab90405b7247fac0e6b8db - with: - cache-on-failure: true - - name: Run reconciliation, coordinate, field-mask and deadline regressions - run: cargo test --locked --no-fail-fast --test lexical_reconcile_canary --test gh493_message_coordinates --test gh493_reported_coordinates --test gh493_windowed_followup --test cli_view_budget --test spec_field_mask_contract -- --nocapture - # Keep every gate failing on errors, but do not let one failure hide the - # remaining independent contracts or the production compiler diagnostics. + - name: Run reconciliation, reported divergence and canonical window regressions + run: cargo test --locked --no-fail-fast --test lexical_reconcile_canary --test gh493_message_coordinates --test gh493_reported_coordinates --test gh493_windowed_followup --test cli_view_budget -- --nocapture - name: Check existing dispatch and error envelopes - if: ${{ !cancelled() }} run: | cargo test --locked --test cli_dispatch_coverage parse_expand_with_context cargo test --locked --test spec_view_expand_error_envelopes - name: Check raw-line compatibility and robot discovery - if: ${{ !cancelled() }} run: | cargo test --locked --test cli_robot view_ - cargo test --locked --test cli_robot fields_ cargo test --locked --test cli_robot capabilities_matches_golden_contract - name: Lint the regression targets and production dependency - if: ${{ !cancelled() }} run: cargo clippy --locked --test gh493_message_coordinates --test gh493_reported_coordinates --test gh493_windowed_followup --test lexical_reconcile_canary -- -D warnings + # Formatting remains blocking, but cannot hide runtime regression results. - name: Check coordinate implementation formatting if: ${{ !cancelled() }} - run: rustfmt --edition 2024 --config skip_children=true --check src/lib.rs src/search/query.rs src/followup_coordinates.rs src/followup_coordinates/window.rs tests/gh493_message_coordinates.rs tests/gh493_reported_coordinates.rs tests/gh493_windowed_followup.rs tests/cli_robot.rs tests/cli_dispatch_coverage.rs tests/util/timeout.rs tests/spec_field_mask_contract.rs + run: rustfmt --edition 2024 --config skip_children=true --check src/lib.rs src/search/query.rs src/followup_coordinates.rs src/followup_coordinates/window.rs tests/gh493_message_coordinates.rs tests/gh493_reported_coordinates.rs tests/gh493_windowed_followup.rs tests/cli_robot.rs tests/cli_dispatch_coverage.rs diff --git a/scripts/verify_ann_shards.py b/scripts/verify_ann_shards.py index 90cb0f32..20d54c60 100644 --- a/scripts/verify_ann_shards.py +++ b/scripts/verify_ann_shards.py @@ -21,42 +21,15 @@ def once(text: str, old: str, new: str) -> str: return text.replace(old, new, 1) -def mirror_module_tree(module: Path, destination: Path) -> list[Path]: - """Preserve normal Rust child-module lookup without changing source bytes. - - An absolute #[path] include changed the lookup base of the split ANN - module, so rustc searched query/execution.rs instead of - query/ann_shards/execution.rs. Mirror the actual tree under the harness's - inline `extracted` module and use a normal `mod ann_shards` declaration. - """ - paths = [module, *sorted(module.with_suffix("").rglob("*.rs"))] - for path in paths: - target = destination / path.relative_to(module.parent) - target.parent.mkdir(parents=True, exist_ok=True) - content = path.read_bytes() - target.write_bytes(content) - if target.read_bytes() != content: - raise RuntimeError(f"Source mirror differs from {path}") - return paths - - def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("cass_root", type=Path) parser.add_argument("frankensearch_root", type=Path) parser.add_argument("destination", type=Path) args = parser.parse_args() - root, upstream, dest = ( - path.resolve() - for path in (args.cass_root, args.frankensearch_root, args.destination) - ) - if any(dest == checkout or checkout in dest.parents for checkout in (root, upstream)): - parser.error("destination must be outside both source checkouts") + root, upstream, dest = (path.resolve() for path in (args.cass_root, args.frankensearch_root, args.destination)) generator = root / "scripts/verify_message_topk_fsvi.py" - subprocess.run( - [sys.executable, str(generator), str(root), str(upstream), str(dest)], - check=True, - ) + subprocess.run([sys.executable, str(generator), str(root), str(upstream), str(dest)], check=True) spec = importlib.util.spec_from_file_location("exact_fsvi", generator) if spec is None or spec.loader is None: raise RuntimeError("Cannot load existing source extractor") @@ -67,74 +40,31 @@ def main() -> None: source_path = dest / "src/lib.rs" source = source_path.read_text() vector = (root / "src/search/vector_index.rs").read_text() - directory = next( - line for line in vector.splitlines() - if line.startswith("pub const VECTOR_INDEX_DIR:") - ) + directory = next(line for line in vector.splitlines() if line.startswith("pub const VECTOR_INDEX_DIR:")) # Include the complete production reporting module, including Serialize and # its exact fallback receipt. Do not substitute harness-only reporting types. - source = once( - source, - "mod search { pub mod vector_index {", - "mod search { #[path = " + json.dumps(str(reporting)) - + "] pub mod ann_index;\npub mod vector_index {\n" + directory + "\n", - ) - source = once( - source, - "mod frankensearch { pub mod core {", - "mod frankensearch { pub mod index { pub use frankensearch_index::*; } pub mod core {", - ) - source = once( - source, - "struct SemanticCandidateRetryState {", - "#[derive(Debug, Default)]\nstruct SemanticCandidateRetryState {", - ) - source = once( - source, - "mod extracted {\nuse super::*;", - "mod extracted {\nuse super::*;\nuse frankensearch_index::{HnswIndex as FsHnswIndex, HNSW_DEFAULT_EF_SEARCH as FS_HNSW_DEFAULT_EF_SEARCH};", - ) + source = once(source, "mod search { pub mod vector_index {", "mod search { #[path = " + json.dumps(str(reporting)) + "] pub mod ann_index;\npub mod vector_index {\n" + directory + "\n") + source = once(source, "mod frankensearch { pub mod core {", "mod frankensearch { pub mod index { pub use frankensearch_index::*; } pub mod core {") + source = once(source, "struct SemanticCandidateRetryState {", "#[derive(Debug, Default)]\nstruct SemanticCandidateRetryState {") + source = once(source, "mod extracted {\nuse super::*;", "mod extracted {\nuse super::*;\nuse frankensearch_index::{HnswIndex as FsHnswIndex, HNSW_DEFAULT_EF_SEARCH as FS_HNSW_DEFAULT_EF_SEARCH};") failure = "#[derive(Debug)]\n" + helper.block(query, r"struct SemanticAnnOpenFailure\b") loader = helper.function(query, "open_fs_semantic_ann_index") - multiplier = next( - line for line in query.splitlines() - if line.startswith("const ANN_CANDIDATE_MULTIPLIER:") - ) + multiplier = next(line for line in query.splitlines() if line.startswith("const ANN_CANDIDATE_MULTIPLIER:")) module = root / "src/search/query/ann_shards.rs" - module_paths = mirror_module_tree(module, dest / "src/extracted") - extra = failure + loader + multiplier + "\nmod ann_shards;\n" - if not source.endswith("}\n"): - raise ValueError("Expected the existing extracted module's closing brace") + extra = failure + loader + multiplier + "\n#[path = " + json.dumps(str(module)) + "]\nmod ann_shards;\n" + assert source.endswith("}\n") source = source[:-2] + extra + "}\n" source_path.write_text(source) manifest_path = dest / "Cargo.toml" manifest = manifest_path.read_text() - old = next( - line for line in manifest.splitlines() - if line.startswith("frankensearch-index =") - ) - manifest = once( - manifest, old, - old.replace("default-features = false", 'default-features = false, features = ["ann"]'), - ) - manifest_path.write_text( - manifest + '\n[dependencies.serde]\nversion = "1"\nfeatures = ["derive"]\n' - ) - recorded_paths = [ - *module_paths, - root / "src/search/query.rs", - root / "src/search/vector_index.rs", - root / "src/search/ann_index.rs", - ] - receipt = { - "scope": "actual CASS all-shard admission and merge with native HNSW; not full CASS or model/CLI qualification", - "source_sha256": { - str(path.relative_to(root)): hashlib.sha256(path.read_bytes()).hexdigest() - for path in recorded_paths - }, - "loader_sha256": hashlib.sha256(loader.encode()).hexdigest(), - "harness_sha256": hashlib.sha256(source.encode()).hexdigest(), - } + old = next(line for line in manifest.splitlines() if line.startswith("frankensearch-index =")) + manifest = once(manifest, old, old.replace("default-features = false", 'default-features = false, features = ["ann"]')) + manifest_path.write_text(manifest + '\n[dependencies.serde]\nversion = "1"\nfeatures = ["derive"]\n') + receipt = {"scope": "actual CASS all-shard admission and merge with native HNSW; not full CASS or model/CLI qualification", + "source_sha256": {str(path.relative_to(root)): hashlib.sha256(path.read_bytes()).hexdigest() + for path in [module, root / "src/search/query.rs", root / "src/search/vector_index.rs", root / "src/search/ann_index.rs"]}, + "loader_sha256": hashlib.sha256(loader.encode()).hexdigest(), + "harness_sha256": hashlib.sha256(source.encode()).hexdigest()} (dest / "ann-source-receipt.json").write_text(json.dumps(receipt, indent=2) + "\n") print(json.dumps(receipt, indent=2)) diff --git a/src/bookmarks.rs b/src/bookmarks.rs index 3fd94a6d..84fb6ce7 100644 --- a/src/bookmarks.rs +++ b/src/bookmarks.rs @@ -488,8 +488,7 @@ pub enum BookmarksCommand { }, /// Export all bookmarks as JSON (to a file, or stdout when no `--output`). Export { - /// New destination file; existing files and links are never replaced. - /// Parent directories are created as needed. + /// Destination file; parent directories are created as needed. #[arg(long, short = 'o')] output: Option, /// Emit a single JSON document instead of human-readable lines. @@ -792,11 +791,16 @@ fn execute(command: &BookmarksCommand, out: &mut dyn Write) -> BookmarkCliResult let payload = serde_json::to_string_pretty(&bookmarks).map_err(|e| { BookmarkCliError::io("serializing bookmarks", e.to_string()) })?; - write_export_file(path, payload.as_bytes()).map_err(|e| { - BookmarkCliError::io( - &format!("exporting bookmarks to {}", path.display()), - format!("{e:#}"), - ) + if let Some(parent) = path.parent().filter(|p| !p.as_os_str().is_empty()) { + std::fs::create_dir_all(parent).map_err(|e| { + BookmarkCliError::io( + &format!("creating export directory {}", parent.display()), + e, + ) + })?; + } + std::fs::write(path, payload).map_err(|e| { + BookmarkCliError::io(&format!("writing {}", path.display()), e) })?; let shown = path.display().to_string(); if json { @@ -873,37 +877,6 @@ fn execute(command: &BookmarksCommand, out: &mut dyn Write) -> BookmarkCliResult } } -/// Publish complete private bytes without ever opening the destination for -/// writing. A preflight exists() check would race other exporters and miss -/// dangling symlinks; no-clobber publication is the authority for both cases. -fn write_export_file(path: &Path, payload: &[u8]) -> Result<()> { - let parent = path - .parent() - .filter(|parent| !parent.as_os_str().is_empty()) - .unwrap_or_else(|| Path::new(".")); - std::fs::create_dir_all(parent).context("creating bookmark export directory")?; - // Stage beside the destination so publication does not cross filesystems. - // NamedTempFile creates owner-only files on Unix, including while writing. - let mut staged = - tempfile::NamedTempFile::new_in(parent).context("creating private bookmark export")?; - staged - .write_all(payload) - .context("writing private bookmark export")?; - staged - .as_file() - .sync_all() - .context("syncing private bookmark export")?; - staged - .persist_noclobber(path) - .map_err(|error| error.error) - .context("publishing bookmark export without overwriting; choose a new --output path")?; - #[cfg(unix)] - std::fs::File::open(parent) - .and_then(|directory| directory.sync_all()) - .context("bookmark export was published, but directory durability could not be confirmed")?; - Ok(()) -} - fn open_store(db_path: &Path) -> BookmarkCliResult { BookmarkStore::open(db_path) .map_err(|e| BookmarkCliError::io("opening bookmarks database", format!("{e:#}"))) @@ -1088,180 +1061,6 @@ mod tests { ); } - #[test] - fn file_export_preserves_existing_bytes_and_error_cause() -> Result<()> { - let dir = tempdir()?; - let output = dir.path().join("existing.json"); - std::fs::write(&output, b"previous private export")?; - let error = write_export_file(&output, b"replacement").expect_err("must not overwrite"); - assert_eq!( - error - .downcast_ref::() - .expect("I/O cause") - .kind(), - std::io::ErrorKind::AlreadyExists - ); - assert!(error.to_string().contains("choose a new --output path")); - assert_eq!(std::fs::read(&output)?, b"previous private export"); - Ok(()) - } - - #[test] - fn file_export_publishes_complete_private_bytes_in_new_directory() -> Result<()> { - let dir = tempdir()?; - let output = dir.path().join("nested/δ-export.json"); - let payload = serde_json::to_vec_pretty(&[ - Bookmark::new("Unicode δ", "/session.jsonl", "codex", "/workspace") - .with_note("private note\nsecond line"), - ])?; - write_export_file(&output, &payload)?; - assert_eq!(std::fs::read(&output)?, payload); - let restored: Vec = serde_json::from_slice(&std::fs::read(&output)?)?; - assert_eq!(restored[0].note, "private note\nsecond line"); - #[cfg(unix)] - { - use std::os::unix::fs::PermissionsExt; - assert_eq!(std::fs::metadata(&output)?.permissions().mode() & 0o077, 0); - } - Ok(()) - } - - #[cfg(unix)] - #[test] - fn file_export_refuses_live_and_dangling_symlinks() -> Result<()> { - use std::os::unix::fs::symlink; - let dir = tempdir()?; - let target = dir.path().join("private.db"); - std::fs::write(&target, b"database bytes")?; - let missing = dir.path().join("missing.db"); - for (name, target) in [("live.json", &target), ("dangling.json", &missing)] { - let output = dir.path().join(name); - symlink(target, &output)?; - assert!(write_export_file(&output, b"not a database").is_err()); - assert_eq!(std::fs::read_link(&output)?, *target); - assert!(std::fs::symlink_metadata(&output)?.file_type().is_symlink()); - } - assert_eq!(std::fs::read(&target)?, b"database bytes"); - assert!(!missing.exists()); - Ok(()) - } - - #[test] - fn file_export_refuses_hardlinks_and_directories() -> Result<()> { - let dir = tempdir()?; - let target = dir.path().join("private.db"); - std::fs::write(&target, b"database bytes")?; - let linked = dir.path().join("alias.json"); - std::fs::hard_link(&target, &linked)?; - assert!(write_export_file(&linked, b"replacement").is_err()); - assert_eq!(std::fs::read(&target)?, b"database bytes"); - assert_eq!(std::fs::read(&linked)?, b"database bytes"); - assert!(write_export_file(dir.path(), b"replacement").is_err()); - assert!(dir.path().is_dir()); - Ok(()) - } - - #[test] - fn file_export_competing_writers_publish_exactly_one_complete_payload() -> Result<()> { - use std::sync::{Arc, Barrier}; - let dir = tempdir()?; - let output = dir.path().join("race.json"); - let barrier = Arc::new(Barrier::new(2)); - let writers = [b'a', b'b'].map(|marker| { - let output = output.clone(); - let barrier = Arc::clone(&barrier); - std::thread::spawn(move || { - let payload = vec![marker; 65_536]; - barrier.wait(); - (marker, write_export_file(&output, &payload)) - }) - }); - let mut winner = None; - for writer in writers { - let (marker, result) = writer.join().expect("export thread"); - match result { - Ok(()) => assert!(winner.replace(marker).is_none(), "only one winner"), - Err(error) => assert_eq!( - error - .downcast_ref::() - .expect("I/O cause") - .kind(), - std::io::ErrorKind::AlreadyExists - ), - } - } - assert_eq!( - std::fs::read(output)?, - vec![winner.expect("one winner"); 65_536] - ); - Ok(()) - } - - #[test] - fn command_file_export_cannot_overwrite_its_bookmark_database() -> Result<()> { - let dir = tempdir()?; - let database = bookmarks_path_in(dir.path()); - let store = BookmarkStore::open(&database)?; - let id = store.add(&Bookmark::new("Keep me", "/saved.jsonl", "codex", "/ws"))?; - drop(store); - let args = BookmarksArgs { - command: BookmarksCommand::Export { - output: Some(database.clone()), - json: true, - data_dir: Some(dir.path().to_path_buf()), - }, - }; - let mut stdout = Vec::new(); - assert_eq!( - run_bookmarks_command_to(args, &mut stdout)?, - BOOKMARKS_EXIT_IO - ); - assert!(stdout.is_empty(), "a refused export has no success receipt"); - let store = BookmarkStore::open(&database)?; - assert_eq!(store.count()?, 1); - assert_eq!(store.get(id)?.expect("retained bookmark").title, "Keep me"); - Ok(()) - } - - #[test] - fn command_file_export_round_trips_and_refuses_repeated_destination() -> Result<()> { - let dir = tempdir()?; - let store = BookmarkStore::open(&bookmarks_path_in(dir.path()))?; - store.add(&Bookmark::new("Keep me", "/saved.jsonl", "codex", "/ws"))?; - drop(store); - for json in [false, true] { - let output = dir.path().join(format!("export-{json}.json")); - let args = BookmarksArgs { - command: BookmarksCommand::Export { - output: Some(output.clone()), - json, - data_dir: Some(dir.path().to_path_buf()), - }, - }; - let mut stdout = Vec::new(); - assert_eq!( - run_bookmarks_command_to(args.clone(), &mut stdout)?, - BOOKMARKS_EXIT_OK - ); - if json { - let receipt: serde_json::Value = serde_json::from_slice(&stdout)?; - assert_eq!(receipt["count"], 1); - assert_eq!(receipt["success"], true); - } - let bytes = std::fs::read(&output)?; - let restored: Vec = serde_json::from_slice(&bytes)?; - assert_eq!(restored[0].title, "Keep me"); - stdout.clear(); - assert_eq!( - run_bookmarks_command_to(args, &mut stdout)?, - BOOKMARKS_EXIT_IO - ); - assert!(stdout.is_empty()); - assert_eq!(std::fs::read(&output)?, bytes); - } - Ok(()) - } - #[test] fn test_create_bookmark() { let bookmark = Bookmark::new("Test", "/path/file.rs", "claude_code", "/workspace") diff --git a/src/lib.rs b/src/lib.rs index 287d0d8f..a2948887 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7000,12 +7000,13 @@ pub async fn run_with_parsed(parsed: ParsedCli) -> CliResult<()> { .flatten() .collect(); - // Human teaching notes can precede execution. For robots, defer notes - // until success: on failure stderr must remain one structured error - // envelope, not plaintext followed by JSON. Successful recovery retains - // its documented teaching note without contaminating stdout. - if !all_notes.is_empty() && !is_doc_mode && !is_robot_mode { - emit_correction_notes(&all_notes, false); + // Teaching notes go to stderr for humans AND robots: README promises that + // agents learn the canonical syntax from a stderr note, and stdout stays + // data-only in robot mode so the note can never corrupt a JSON payload. + // Only the robot-docs/--robot-help surfaces stay quiet, because their + // stderr is part of the documented docs stream. + if !all_notes.is_empty() && !is_doc_mode { + emit_correction_notes(&all_notes, is_robot_mode); } let result = execute_cli( @@ -7017,10 +7018,6 @@ pub async fn run_with_parsed(parsed: ParsedCli) -> CliResult<()> { ) .await; - if result.is_ok() && !all_notes.is_empty() && !is_doc_mode && is_robot_mode { - emit_correction_notes(&all_notes, true); - } - if let Some(path) = &cli.trace_file { let duration_ms = start_instant.elapsed().as_millis(); let exit_code = result.as_ref().map_or_else(|e| e.code, |()| 0); @@ -7104,8 +7101,8 @@ pub fn try_run_with_parsed_fast(parsed: ParsedCli) -> Result, Box< .flatten() .collect(); - if !all_notes.is_empty() && !is_robot_mode { - emit_correction_notes(&all_notes, false); + if !all_notes.is_empty() { + emit_correction_notes(&all_notes, is_robot_mode); } let result = match command.expect("fast command was matched above") { @@ -7133,10 +7130,6 @@ pub fn try_run_with_parsed_fast(parsed: ParsedCli) -> Result, Box< _ => unreachable!("non-fast command passed the fast-command guard"), }; - if result.is_ok() && !all_notes.is_empty() && is_robot_mode { - emit_correction_notes(&all_notes, true); - } - if let Some(path) = &cli.trace_file { let duration_ms = start_instant.elapsed().as_millis(); let exit_code = result.as_ref().map_or_else(|e| e.code, |()| 0); diff --git a/src/logical_archive/export.rs b/src/logical_archive/export.rs index 6a695cfb..279ad84f 100644 --- a/src/logical_archive/export.rs +++ b/src/logical_archive/export.rs @@ -97,15 +97,6 @@ pub fn schema_version(connection: &Connection) -> Result { .map_err(|_| anyhow!("source does not contain a supported canonical schema version")) } -/// FTS5 owns a closed set of shadow names, not the entire `_` namespace. -/// Call only for a discovered virtual root: without that owner even an exact -/// shadow-like name is ordinary data and must be exported or explicitly refused. -fn is_fts5_shadow_table(name: &str, root: &str) -> bool { - name.strip_prefix(root).is_some_and(|suffix| { - matches!(suffix, "_config" | "_content" | "_data" | "_docsize" | "_idx") - }) -} - /// Metadata is bounded independently of the number of canonical rows. Never /// execute stored CREATE statements: only inspect a short virtual-table prefix. pub fn tables(connection: &Connection) -> Result> { @@ -131,9 +122,7 @@ pub fn tables(connection: &Connection) -> Result> { let mut output = Vec::new(); for name in physical { if name.starts_with("sqlite_") - || virtual_roots - .iter() - .any(|root| is_fts5_shadow_table(&name, root)) + || virtual_roots.iter().any(|root| name == *root || name.starts_with(&format!("{root}_"))) { continue; } @@ -250,10 +239,7 @@ pub fn sync_parent(destination: &Path) -> Result<()> { } pub fn verify_file(path: &Path) -> Result<(Header, Completion)> { - // Apply the same regular-file admission as import before any blocking read. - // File::open alone can wait forever on a FIFO before framing limits apply. - let input = super::import::open_input(path)?; - codec::verify(&mut BufReader::new(input)) + codec::verify(&mut BufReader::new(File::open(path).context("cannot open logical archive")?)) } #[cfg(test)] diff --git a/src/logical_archive/export_tests.rs b/src/logical_archive/export_tests.rs index 9f7e0346..340f1bfa 100644 --- a/src/logical_archive/export_tests.rs +++ b/src/logical_archive/export_tests.rs @@ -73,152 +73,6 @@ fn unknown_unkeyed_tables_are_not_silently_omitted() { assert!(!output.exists()); } -#[test] -fn fts5_shadow_names_are_exact_not_prefixes() { - for suffix in ["config", "content", "data", "docsize", "idx"] { - assert!(is_fts5_shadow_table( - &format!("fts_messages_{suffix}"), - "fts_messages" - )); - } - for name in [ - "fts_messages", - "fts_messages_", - "fts_messages_notes", - "fts_messages_data_backup", - "fts_messages_datax", - "other_fts_messages_data", - "fts_messages2_data", - ] { - assert!(!is_fts5_shadow_table(name, "fts_messages"), "omitted {name}"); - } -} - -#[test] -fn similarly_named_logical_tables_survive_export_and_affect_the_digest() -> Result<()> { - let source = tempfile::tempdir()?; - let destination = tempfile::tempdir()?; - let database = source.path().join("agent_search.db"); - fixture(&database); - let connection = Connection::open(path_text(&database)?)?; - connection.execute_batch( - "CREATE VIRTUAL TABLE fts_messages USING fts5(body); - INSERT INTO fts_messages(rowid, body) VALUES (7, 'derived transcript'); - CREATE TABLE fts_messages_notes (id INTEGER PRIMARY KEY, note TEXT NOT NULL); - INSERT INTO fts_messages_notes VALUES (1, 'authoritative annotation'); - CREATE TABLE fts_messages_data_backup (id INTEGER PRIMARY KEY, raw BLOB); - INSERT INTO fts_messages_data_backup VALUES (2, X'0001FF');", - )?; - connection.close()?; - let before = contents(source.path()); - let output = destination.path().join("archive.jsonl"); - let first = export_file(&database, &output, "shadow-scope".to_owned())?; - assert_eq!(first, verify_file(&output)?); - assert_eq!(first.1.records, 5); - assert_eq!(first.1.tables["fts_messages_notes"], 1); - assert_eq!(first.1.tables["fts_messages_data_backup"], 1); - for name in [ - "fts_messages", - "fts_messages_config", - "fts_messages_content", - "fts_messages_data", - "fts_messages_docsize", - "fts_messages_idx", - ] { - assert!( - !first.1.tables.contains_key(name), - "exported derived table {name}" - ); - } - let mut reader = BufReader::new(File::open(&output)?); - let mut saw_note = false; - let mut saw_blob = false; - let mut table_name = String::new(); - let mut line = 1; - while let Some(record) = codec::read_record(&mut reader, line)? { - match record { - Record::Table { table } => table_name = table.name, - Record::Row { values } if table_name == "fts_messages_notes" => { - assert_eq!( - values, - vec![ - Cell::Integer(1), - Cell::Text("authoritative annotation".to_owned()) - ] - ); - saw_note = true; - } - Record::Row { values } if table_name == "fts_messages_data_backup" => { - assert_eq!( - values, - vec![Cell::Integer(2), Cell::Blob("AAH/".to_owned())] - ); - saw_blob = true; - } - _ => {} - } - line += 1; - } - assert!( - saw_note && saw_blob, - "real typed rows, not just descriptors, must survive" - ); - assert_eq!(before, contents(source.path())); - - let connection = Connection::open(path_text(&database)?)?; - connection.execute("UPDATE fts_messages_notes SET note = 'changed annotation' WHERE id = 1")?; - connection.close()?; - let changed = destination.path().join("changed.jsonl"); - let second = export_file(&database, &changed, "shadow-scope".to_owned())?; - assert_eq!(second, verify_file(&changed)?); - assert_eq!(first.1.tables, second.1.tables); - assert_ne!(first.1.content_sha256, second.1.content_sha256); - Ok(()) -} - -#[test] -fn shadow_like_names_without_a_virtual_owner_are_exported() -> Result<()> { - let source = tempfile::tempdir()?; - let destination = tempfile::tempdir()?; - let database = source.path().join("agent_search.db"); - fixture(&database); - let connection = Connection::open(path_text(&database)?)?; - connection.execute_batch( - "CREATE TABLE fts_messages_data (id INTEGER PRIMARY KEY, body TEXT); - INSERT INTO fts_messages_data VALUES (1, 'not owned by FTS5');", - )?; - connection.close()?; - let output = destination.path().join("archive.jsonl"); - let receipt = export_file(&database, &output, "no-virtual-owner".to_owned())?; - assert_eq!(receipt, verify_file(&output)?); - assert_eq!(receipt.1.tables["fts_messages_data"], 1); - assert_eq!(receipt.1.records, 4); - Ok(()) -} - -#[test] -fn unkeyed_fts_prefix_table_is_refused_not_silently_omitted() -> Result<()> { - let source = tempfile::tempdir()?; - let destination = tempfile::tempdir()?; - let database = source.path().join("agent_search.db"); - fixture(&database); - let connection = Connection::open(path_text(&database)?)?; - connection.execute_batch( - "CREATE VIRTUAL TABLE fts_messages USING fts5(body); - CREATE TABLE fts_messages_notes (body TEXT); - INSERT INTO fts_messages_notes VALUES ('do not silently lose me');", - )?; - connection.close()?; - let before = contents(source.path()); - let output = destination.path().join("archive.jsonl"); - let error = export_file(&database, &output, "unkeyed-prefix".to_owned()) - .expect_err("unsupported authoritative data must refuse the whole export"); - assert!(error.to_string().contains("fts_messages_notes"), "{error:#}"); - assert!(!output.exists()); - assert_eq!(before, contents(source.path())); - Ok(()) -} - #[test] fn canonical_empty_archive_schema_has_an_exportable_snapshot() { let source = tempfile::tempdir().unwrap(); @@ -250,23 +104,3 @@ fn symlink_sources_and_destination_locks_are_refused() { symlink(&database, root.path().join(".archive.jsonl.logical-archive.lock")).unwrap(); assert!(DestinationLock::acquire(&output).is_err()); } - -#[cfg(unix)] -#[test] -fn verification_refuses_nonregular_inputs_before_decoding() -> Result<()> { - use std::os::unix::fs::symlink; - let root = tempfile::tempdir()?; - let target = root.path().join("private.jsonl"); - fs::write(&target, "PRIVATE-NOT-JSON\n")?; - let linked = root.path().join("linked.jsonl"); - symlink(&target, &linked)?; - for input in [root.path(), linked.as_path(), Path::new("/dev/null")] { - let error = verify_file(input).expect_err("nonregular input must be refused"); - let message = format!("{error:#}"); - assert!(message.contains("regular, non-symlink"), "{message}"); - assert!(!message.contains("PRIVATE-NOT-JSON")); - } - assert_eq!(fs::read(&target)?, b"PRIVATE-NOT-JSON\n"); - assert_eq!(fs::read_link(&linked)?, target); - Ok(()) -} diff --git a/src/main.rs b/src/main.rs index 3e449c7e..40044a27 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,7 +3,6 @@ #![cfg_attr(not(test), deny(unsafe_code))] mod logical_archive; -mod search_service; fn env_requests_robot_output() -> bool { let cass_output_format = dotenvy::var("CASS_OUTPUT_FORMAT") @@ -281,16 +280,6 @@ fn main() -> anyhow::Result<()> { } }; } - if raw_command_name(&raw_args) == Some("serve") { - let result = search_service::run(raw_args); - if !coding_agent_search::shutdown_thread_local_bridge_runtimes() { - tracing::warn!("search service bridge teardown exceeded its deadline"); - } - return match result { - Ok(()) => Ok(()), - Err(err) => handle_fatal_error(err), - }; - } let parsed = match coding_agent_search::parse_cli(raw_args) { Ok(parsed) => parsed, Err(err) => handle_fatal_error(err), diff --git a/src/search/query/ann_shards/execution.rs b/src/search/query/ann_shards/execution.rs index 010ca2b7..d61e4d1c 100644 --- a/src/search/query/ann_shards/execution.rs +++ b/src/search/query/ann_shards/execution.rs @@ -1,49 +1,11 @@ //! Execution over a retained native cohort and its durable updates. //! -//! Native HNSW covers only the persisted main slab. Small retained WAL deltas -//! are scored once per request and merged with that graph's current winners. -//! Oversized deltas retain complete-cohort exact recovery; no serving path -//! reopens, compacts, or publishes an artifact. +//! Native HNSW covers only the persisted main slab. A readable WAL does not +//! invalidate that graph's main-slab identity, but it adds or supersedes live +//! documents. Never report the main-only page as complete in that state. use super::*; -/// Shared across the cohort, not multiplied by shard count or refill passes. -/// These bound additional delta work/state, not graph ownership or total RSS. -const MAX_NATIVE_WAL_RECORDS: usize = 4_096; -const MAX_NATIVE_WAL_COMPONENTS: usize = 4_194_304; - -struct WalShardOverlay<'a> { - /// Include every updated document, even when its replacement is filtered - /// out. An obsolete main-slab score must never resurrect through the graph. - shadowed: HashSet<&'a str>, - hits: Vec, - has_more_candidates: bool, -} - -type WalCohortOverlay<'a> = HashMap>; - -fn wal_delta_fits_budget(shards: impl IntoIterator) -> bool { - let mut records = 0usize; - let mut components = 0usize; - for (count, dimension) in shards { - let Some(next_records) = records.checked_add(count) else { - return false; - }; - let Some(next_components) = count - .checked_mul(dimension) - .and_then(|count| components.checked_add(count)) - else { - return false; - }; - if next_records > MAX_NATIVE_WAL_RECORDS || next_components > MAX_NATIVE_WAL_COMPONENTS { - return false; - } - records = next_records; - components = next_components; - } - true -} - #[derive(Debug)] struct PendingWalDelta { dimension: usize, @@ -51,7 +13,7 @@ struct PendingWalDelta { impl std::fmt::Display for PendingWalDelta { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - formatter.write_str("retained WAL updates exceed the native delta work budget") + formatter.write_str("retained WAL updates require delta-aware semantic retrieval") } } @@ -59,10 +21,10 @@ impl std::error::Error for PendingWalDelta {} impl SemanticAnnShardSet { /// Refill underfilled native windows before paying for a complete exact scan. - /// Every pass reuses one scored WAL overlay and the same retained cohort, - /// query and filter. The graph-call/window budget is shared across shards. - /// Native failure, oversized WAL deltas or unresolved underfill still recover - /// the entire exact cohort; incomplete native results never escape recovery. + /// Extra native passes keep the same retained cohort, query and filter, and + /// have a shared graph-call/window budget. A complete initial page performs + /// no extra work. Native failure or unresolved underfill still recovers the + /// entire exact cohort; no partial native/exact ranking is returned. /// Invalid queries and mismatched context owners remain errors, not fallbacks. pub(in super::super) fn search_with_exact_fallback( &self, @@ -75,75 +37,51 @@ impl SemanticAnnShardSet { SemanticCandidateRetryState, Option, )> { - if !self.validate_native_query(&context.artifacts, embedding, fetch_limit)? { - return Ok((Vec::new(), SemanticCandidateRetryState::default(), None)); - } let mut budget = NativeRefillBudget::new(fetch_limit, context.artifacts.len()); let mut native_window = fetch_limit; let mut stats: Option = None; let mut best_by_message: HashMap = HashMap::new(); - let (native_messages, reason) = - match self.prepare_wal_overlay(embedding, fetch_limit, filter) { - Ok(overlay) => loop { - match self.search_prepared(embedding, native_window, filter, &overlay) { - Ok((hits, mut retry, measured)) => { - if let Some(measured) = measured { - match stats.as_mut() { - Some(total) => accumulate_native_stats(total, &measured), - None => stats = Some(measured), - } - } - // A wider approximate window need not contain all - // earlier winners. All passes apply the same WAL - // shadow set, so retaining these scores is safe. - for hit in hits { - best_by_message - .entry(hit.message_id) - .and_modify(|best| { - if hit.score.total_cmp(&best.score).is_gt() { - best.score = hit.score; - best.chunk_idx = hit.chunk_idx; - } - }) - .or_insert(hit); - } - let omitted_retained = best_by_message.len() > fetch_limit; - let hits = SearchClient::collapse_semantic_results( - best_by_message, - fetch_limit, - ); - if hits.len() >= fetch_limit || !retry.has_more_candidates { - retry.has_more_candidates |= omitted_retained; - return Ok((hits, retry, stats)); - } - let Some(next_window) = budget.next_window() else { - let reason = if filter.is_some() { - AnnExactFallbackReason::FilteredCandidateUnderfill - } else { - AnnExactFallbackReason::MessageCandidateUnderfill - }; - break (hits.len(), reason); - }; - best_by_message = - hits.into_iter().map(|hit| (hit.message_id, hit)).collect(); - native_window = next_window; + let (native_messages, reason) = loop { + match self.search(&context.artifacts, embedding, native_window, filter) { + Ok((hits, mut retry, measured)) => { + if let Some(measured) = measured { + match stats.as_mut() { + Some(total) => accumulate_native_stats(total, &measured), + None => stats = Some(measured), } - Err(error) => match error.downcast::() { - Ok(failure) => { - // Keep completed work, but discard ALL native - // and delta hits before exact cohort recovery. - match stats.as_mut() { - Some(total) => { - accumulate_native_stats(total, &failure.completed_stats); - } - None => stats = Some(failure.completed_stats), - } - break (0, AnnExactFallbackReason::NativeSearchFailed); - } - Err(error) => return Err(error), - }, } - }, + // A larger approximate window need not be a strict superset + // of the previous one. Keep its earlier valid winners too; + // between passes, retain only the requested message page. + for hit in hits { + best_by_message + .entry(hit.message_id) + .and_modify(|best| { + if hit.score.total_cmp(&best.score).is_gt() { + best.score = hit.score; + best.chunk_idx = hit.chunk_idx; + } + }) + .or_insert(hit); + } + let omitted_retained = best_by_message.len() > fetch_limit; + let hits = + SearchClient::collapse_semantic_results(best_by_message, fetch_limit); + if hits.len() >= fetch_limit || !retry.has_more_candidates { + retry.has_more_candidates |= omitted_retained; + return Ok((hits, retry, stats)); + } + let Some(next_window) = budget.next_window() else { + let reason = if filter.is_some() { + AnnExactFallbackReason::FilteredCandidateUnderfill + } else { + AnnExactFallbackReason::MessageCandidateUnderfill + }; + break (hits.len(), reason); + }; + best_by_message = hits.into_iter().map(|hit| (hit.message_id, hit)).collect(); + native_window = next_window; + } Err(error) if error.is::() => { let pending = error .downcast_ref::() @@ -152,10 +90,27 @@ impl SemanticAnnShardSet { dimension: pending.dimension, ..Default::default() }); - (0, AnnExactFallbackReason::WalDeltaRequiresExact) + break (0, AnnExactFallbackReason::WalDeltaRequiresExact); } - Err(error) => return Err(error), - }; + Err(error) => match error.downcast::() { + Ok(failure) => { + // Include completed calls from earlier refill passes, + // not just the prefix of the pass that failed. Discard + // ALL native hits before complete-cohort exact recovery. + match stats.as_mut() { + Some(total) => { + accumulate_native_stats(total, &failure.completed_stats); + } + None => stats = Some(failure.completed_stats), + } + break (0, AnnExactFallbackReason::NativeSearchFailed); + } + // Do not turn a caller/provenance error into an expensive + // exact scan, or mistake it for a failed derived graph. + Err(error) => return Err(error), + }, + } + }; let started = std::time::Instant::now(); let (exact, exact_retry) = SearchClient::search_exact_semantic_indexes(context, embedding, fetch_limit, filter)?; @@ -179,12 +134,13 @@ impl SemanticAnnShardSet { Ok((exact, exact_retry, stats)) } - /// Search the admitted main graphs plus their bounded, current WAL overlays. + /// Search each admitted graph, then merge by best chunk per message. /// - /// Statistics continue to describe native calls only. Their recall estimate - /// is a backend heuristic, not a measured guarantee for the fused page. - /// Delta work is separately traced; it never makes approximate main-slab - /// retrieval exact. Additional state is O(k + bounded delta), not O(archive). + /// Statistics describe actual native calls: sizes, requested/returned raw + /// candidates, and native-search times are summed; ef is the maximum and + /// estimated recall the minimum of the backend heuristics, NOT a measured + /// recall guarantee for the fused message page. Candidate state is O(k), + /// not O(k * shards); the retained graphs themselves are not memory-capped. pub(in super::super) fn search( &self, artifacts: &Arc>, @@ -196,134 +152,23 @@ impl SemanticAnnShardSet { SemanticCandidateRetryState, Option, )> { - if !self.validate_native_query(artifacts, embedding, fetch_limit)? { - return Ok((Vec::new(), SemanticCandidateRetryState::default(), None)); - } - let overlay = self.prepare_wal_overlay(embedding, fetch_limit, filter)?; - self.search_prepared(embedding, fetch_limit, filter, &overlay) - } - - /// Preserve owner validation even for zero-limit requests. As before, a - /// zero-limit request does not validate/scour vectors or prepare any delta. - fn validate_native_query( - &self, - artifacts: &Arc>, - embedding: &[f32], - fetch_limit: usize, - ) -> Result { if !Arc::ptr_eq(&self.artifacts, artifacts) { bail!("native ANN cohort does not match the admitted semantic context"); } if fetch_limit == 0 { - return Ok(false); + return Ok((Vec::new(), SemanticCandidateRetryState::default(), None)); } let dimension = self.artifacts[0].index().dimension(); if embedding.len() != dimension || embedding.iter().any(|value| !value.is_finite()) { bail!("native ANN query must have the admitted dimension and finite values"); } - Ok(true) - } - - /// All metadata admission precedes allocation/scoring and graph execution. - /// Borrow the reader's already deduplicated last-write-wins rows, never the - /// WAL pathname. Cache only per-shard top-k messages and document shadows. - fn prepare_wal_overlay<'a>( - &'a self, - embedding: &[f32], - fetch_limit: usize, - filter: Option<&dyn FsSearchFilter>, - ) -> Result> { - if !wal_delta_fits_budget(self.artifacts.iter().map(|artifact| { - let index = artifact.index(); - (index.wal_record_count(), index.dimension()) - })) { - return Err(PendingWalDelta { - dimension: embedding.len(), - } - .into()); + if self + .artifacts + .iter() + .any(|artifact| artifact.index().wal_record_count() > 0) + { + return Err(PendingWalDelta { dimension }.into()); } - let started = std::time::Instant::now(); - let mut overlay = HashMap::new(); - let mut scored = 0usize; - let mut records = 0usize; - for (ordinal, artifact) in self.artifacts.iter().enumerate() { - let index = artifact.index(); - let count = index.wal_record_count(); - if count == 0 { - continue; - } - let mut shadowed = HashSet::with_capacity(count); - let mut best_by_message: HashMap = HashMap::new(); - for (doc_id, vector) in index.wal_records() { - shadowed.insert(doc_id); - let parsed = parse_semantic_doc_id(doc_id).ok_or_else(|| { - anyhow!("retained WAL has an invalid semantic document identity") - })?; - if filter.is_some_and(|filter| !filter.matches(doc_id, None)) { - continue; - } - // Use the same f32 kernel as upstream exact WAL retrieval, - // including the retained reader's F16 replay quantization. - let score = frankensearch::index::dot_product_f32_f32(vector, embedding) - .map_err(|error| anyhow!("retained WAL scoring failed: {error}"))?; - if !score.is_finite() { - bail!("retained WAL scoring produced a non-finite result"); - } - scored += 1; - best_by_message - .entry(parsed.message_id) - .and_modify(|best| { - if score.total_cmp(&best.score).is_gt() { - best.score = score; - best.chunk_idx = parsed.chunk_idx; - } - }) - .or_insert(VectorSearchResult { - message_id: parsed.message_id, - chunk_idx: parsed.chunk_idx, - score, - }); - } - records += count; - let has_more_candidates = best_by_message.len() > fetch_limit; - let hits = SearchClient::collapse_semantic_results(best_by_message, fetch_limit); - overlay.insert( - ordinal, - WalShardOverlay { - shadowed, - hits, - has_more_candidates, - }, - ); - } - if !overlay.is_empty() { - tracing::debug!( - delta_shards = overlay.len(), - delta_records = records, - delta_records_scored = scored, - delta_search_time_us = - u64::try_from(started.elapsed().as_micros()).unwrap_or(u64::MAX), - "prepared retained WAL overlay once for native retrieval" - ); - } - Ok(overlay) - } - - /// Called only after request validation and complete delta preparation. - /// Each shadow set is shard-local: a replacement in one artifact cannot - /// erase a distinct current contribution from another retained artifact. - fn search_prepared( - &self, - embedding: &[f32], - fetch_limit: usize, - filter: Option<&dyn FsSearchFilter>, - overlay: &WalCohortOverlay<'_>, - ) -> Result<( - Vec, - SemanticCandidateRetryState, - Option, - )> { - let dimension = embedding.len(); let candidate_limit = fetch_limit .saturating_mul(ANN_CANDIDATE_MULTIPLIER) .max(fetch_limit); @@ -335,19 +180,23 @@ impl SemanticAnnShardSet { let mut best_by_message = HashMap::new(); let mut has_more_candidates = false; for (ordinal, graph) in self.graphs.iter().enumerate() { + // A large global page must not request a huge beam from a small + // shard. The native graph count bounds its candidate request. let candidate = candidate_limit.min(graph.len()); let ef = FS_HNSW_DEFAULT_EF_SEARCH.max(candidate); - // Borrow the exact source selected at admission, never its path. - // The native call also checks source extent and returned row IDs. + // Native underfill repair needs the exact source selected during + // admission. Borrow its retained reader, never reopen its path. + // The source-backed call also checks extent and returned row IDs. let source = self.artifacts[ordinal].index(); - let delta = overlay.get(&ordinal); let (hits, measured) = graph .knn_search_with_stats_against(source, embedding, candidate, ef) .map_err(|_| NativeAnnSearchFailure::new(ordinal, stats.clone()))?; if measured.dimension != dimension || hits.len() > candidate { return Err(NativeAnnSearchFailure::new(ordinal, stats).into()); } - // A failed later shard cannot return an earlier partial page. + // A failure above returns no partial success assembled from earlier + // shards. No graph is skipped because it produced no filtered hits. + // Post-top-k document deduplication can shorten a complete window. has_more_candidates |= candidate < measured.index_size; stats.index_size = stats.index_size.saturating_add(measured.index_size); stats.ef_search = stats.ef_search.max(measured.ef_search); @@ -357,28 +206,12 @@ impl SemanticAnnShardSet { stats.estimated_recall = stats.estimated_recall.min(measured.estimated_recall as f32); stats.is_approximate |= measured.is_approximate; for hit in &hits { - if delta.is_some_and(|delta| delta.shadowed.contains(hit.doc_id.as_str())) { - continue; - } if filter.is_none_or(|filter| filter.matches(&hit.doc_id, None)) { SearchClient::record_fs_semantic_hit(&mut best_by_message, hit); } } - if let Some(delta) = delta { - has_more_candidates |= delta.has_more_candidates; - for hit in &delta.hits { - best_by_message - .entry(hit.message_id) - .and_modify(|best| { - if hit.score.total_cmp(&best.score).is_gt() { - best.score = hit.score; - best.chunk_idx = hit.chunk_idx; - } - }) - .or_insert_with(|| hit.clone()); - } - } - // Exhausting the graph does not exhaust a larger WAL message page. + // Raw graph exhaustion is not message-page exhaustion: collapsing + // an exhaustive native window can still discard eligible messages. has_more_candidates |= best_by_message.len() > fetch_limit; best_by_message = SearchClient::collapse_semantic_results(best_by_message, fetch_limit) .into_iter() @@ -387,10 +220,9 @@ impl SemanticAnnShardSet { } tracing::debug!( shards = self.graphs.len(), - delta_shards = overlay.len(), native_candidates = stats.k_returned, returned_messages = best_by_message.len(), - "native ANN and retained WAL shard merge complete" + "native ANN shard merge complete" ); Ok(( SearchClient::collapse_semantic_results(best_by_message, fetch_limit), @@ -402,26 +234,3 @@ impl SemanticAnnShardSet { )) } } - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn wal_budget_is_shared_across_shards() { - assert!(wal_delta_fits_budget([(2_048, 384), (2_048, 384)])); - assert!(!wal_delta_fits_budget([(2_048, 384), (2_049, 384)])); - assert!(!wal_delta_fits_budget(std::iter::repeat_n((1, 1), 4_097))); - } - - #[test] - fn wal_budget_bounds_scalar_work_and_handles_overflow() { - assert!(wal_delta_fits_budget([(4_096, 1_024)])); - assert!(!wal_delta_fits_budget([(4_096, 1_025)])); - assert!(!wal_delta_fits_budget([(1, usize::MAX)])); - assert!(!wal_delta_fits_budget([(2, usize::MAX)])); - assert!(!wal_delta_fits_budget([(usize::MAX, 2)])); - assert!(wal_delta_fits_budget([(0, usize::MAX)])); - assert!(wal_delta_fits_budget([])); - } -} diff --git a/src/search/query/ann_shards/wal_tests.rs b/src/search/query/ann_shards/wal_tests.rs index 8f5bf941..33c53f34 100644 --- a/src/search/query/ann_shards/wal_tests.rs +++ b/src/search/query/ann_shards/wal_tests.rs @@ -56,9 +56,7 @@ fn shard( if !updates.is_empty() { let mut source = FsVectorIndex::open_writer(&path).unwrap(); for (id, vector) in updates { - source - .append_batch(&[(id.clone(), vector.to_vec())]) - .unwrap(); + source.append_batch(&[(id.clone(), vector.to_vec())]).unwrap(); } assert!(source.wal_record_count() > 0); } @@ -66,10 +64,7 @@ fn shard( let artifact = SemanticIndexArtifact::open(path, Some(ann)).unwrap(); assert_eq!(artifact.index().record_count(), base.len()); if !updates.is_empty() { - assert!( - artifact.index().wal_record_count() > 0, - "reopen must replay durable updates" - ); + assert!(artifact.index().wal_record_count() > 0, "reopen must replay durable updates"); } artifact } @@ -88,9 +83,7 @@ fn context(artifacts: Vec) -> SemanticCandidateContext { } fn signature(hits: &[VectorSearchResult]) -> Vec<(u64, u8, u32)> { - hits.iter() - .map(|hit| (hit.message_id, hit.chunk_idx, hit.score.to_bits())) - .collect() + hits.iter().map(|hit| (hit.message_id, hit.chunk_idx, hit.score.to_bits())).collect() } fn files(directory: &Path) -> Vec<(PathBuf, Vec)> { @@ -121,31 +114,17 @@ fn a_full_native_page_cannot_hide_a_new_durable_wal_winner() { .knn_search_with_stats_against(ctx.artifacts[0].index(), &[1.0, 0.0], 2, 100) .unwrap(); assert_eq!(main_only.len(), 2, "fixture must fill the old native page"); - assert!( - main_only - .iter() - .all(|hit| parse_semantic_doc_id(&hit.doc_id).unwrap().message_id != 99) - ); + assert!(main_only.iter().all(|hit| parse_semantic_doc_id(&hit.doc_id).unwrap().message_id != 99)); let before = files(temp.path()); - let (expected, _) = - SearchClient::search_exact_semantic_indexes(&ctx, &[1.0, 0.0], 1, None).unwrap(); - let (actual, retry, stats) = set - .search_with_exact_fallback(&ctx, &[1.0, 0.0], 1, None) - .unwrap(); - assert_eq!( - actual.len(), - 1, - "native serving preserves the requested page" - ); + let (expected, expected_retry) = SearchClient::search_exact_semantic_indexes(&ctx, &[1.0, 0.0], 1, None).unwrap(); + let (actual, retry, stats) = set.search_with_exact_fallback(&ctx, &[1.0, 0.0], 1, None).unwrap(); assert_eq!(actual[0].message_id, 99); - assert_eq!(signature(&actual), signature(&expected[..1])); - assert!(retry.has_more_candidates); + assert_eq!(signature(&actual), signature(&expected)); + assert_eq!(retry.has_more_candidates, expected_retry.has_more_candidates); let stats = stats.unwrap(); - assert_eq!(stats.k_requested, 2, "the retained graph remains in use"); - assert!( - stats.exact_fallback.is_none(), - "a small WAL is not a full scan" - ); + assert!(!stats.is_approximate); + assert_eq!(stats.k_requested, 0, "the main-only graph must not execute"); + assert_eq!(stats.exact_fallback.unwrap().reason, AnnExactFallbackReason::WalDeltaRequiresExact); assert_eq!(files(temp.path()), before); } @@ -158,26 +137,13 @@ fn latest_wal_replacement_supersedes_a_better_old_graph_vector() { &[(doc(1, 3), [1.0, 0.0]), (doc(2, 3), [0.6, 0.8])], &[(doc(1, 3), [0.8, 0.6]), (doc(1, 3), [0.0, 1.0])], )]); - assert_eq!( - ctx.artifacts[0].index().wal_record_count(), - 1, - "last durable write wins" - ); + assert_eq!(ctx.artifacts[0].index().wal_record_count(), 1, "last durable write wins"); let set = SemanticAnnShardSet::open(Arc::clone(&ctx.artifacts)).unwrap(); - let before = files(temp.path()); - let (expected, _) = - SearchClient::search_exact_semantic_indexes(&ctx, &[1.0, 0.0], 2, None).unwrap(); - let (actual, _, stats) = set - .search_with_exact_fallback(&ctx, &[1.0, 0.0], 2, None) - .unwrap(); + let (expected, _) = SearchClient::search_exact_semantic_indexes(&ctx, &[1.0, 0.0], 2, None).unwrap(); + let (actual, _, _) = set.search_with_exact_fallback(&ctx, &[1.0, 0.0], 2, None).unwrap(); assert_eq!(actual[0].message_id, 2); assert_eq!(signature(&actual), signature(&expected)); - assert_eq!( - actual.iter().find(|hit| hit.message_id == 1).unwrap().score, - 0.0 - ); - assert!(stats.unwrap().exact_fallback.is_none()); - assert_eq!(files(temp.path()), before); + assert_eq!(actual.iter().find(|hit| hit.message_id == 1).unwrap().score, 0.0); } #[test] @@ -185,21 +151,13 @@ fn a_later_shards_wal_keeps_the_complete_source_scoped_cohort() { let temp = tempfile::tempdir().unwrap(); let ctx = context(vec![ shard(temp.path(), "base-a", &[(doc(1, 3), [0.6, 0.8])], &[]), - shard( - temp.path(), - "delta-b", - &[(doc(2, 3), [0.8, 0.6])], - &[(doc(3, 4), [1.0, 0.0]), (doc(4, 3), [0.9, 0.4358899])], - ), + shard(temp.path(), "delta-b", &[(doc(2, 3), [0.8, 0.6])], &[ + (doc(3, 4), [1.0, 0.0]), + (doc(4, 3), [0.9, 0.4358899]), + ]), ]); let set = SemanticAnnShardSet::open(Arc::clone(&ctx.artifacts)).unwrap(); - let before = files(temp.path()); for sources in [HashSet::from([3]), HashSet::new()] { - let expected = if sources.is_empty() { - Vec::new() - } else { - vec![4, 2, 1] - }; let filter = SemanticFilter { agents: Some(HashSet::from([1])), workspaces: Some(HashSet::from([2])), @@ -208,338 +166,35 @@ fn a_later_shards_wal_keeps_the_complete_source_scoped_cohort() { created_from: Some(100), created_to: Some(100), }; - let (actual, _, stats) = set - .search_with_exact_fallback(&ctx, &[1.0, 0.0], 3, Some(&filter)) - .unwrap(); - assert_eq!( - actual.iter().map(|hit| hit.message_id).collect::>(), - expected - ); - assert!(stats.unwrap().exact_fallback.is_none()); + let (expected, _) = SearchClient::search_exact_semantic_indexes(&ctx, &[1.0, 0.0], 3, Some(&filter)).unwrap(); + let (actual, _, stats) = set.search_with_exact_fallback(&ctx, &[1.0, 0.0], 3, Some(&filter)).unwrap(); + assert_eq!(signature(&actual), signature(&expected)); + assert!(actual.iter().all(|hit| hit.message_id != 3)); + assert_eq!(stats.unwrap().exact_fallback.unwrap().shard_count, 2); } - assert_eq!(files(temp.path()), before); } #[test] fn wal_recovery_never_reopens_renamed_sources_or_relaxes_request_validation() { let temp = tempfile::tempdir().unwrap(); - let ctx = context(vec![shard( - temp.path(), - "retained", - &[(doc(1, 3), [0.6, 0.8])], - &[(doc(2, 3), [1.0, 0.0])], - )]); + let ctx = context(vec![shard(temp.path(), "retained", &[(doc(1, 3), [0.6, 0.8])], &[(doc(2, 3), [1.0, 0.0])])]); let set = SemanticAnnShardSet::open(Arc::clone(&ctx.artifacts)).unwrap(); let fsvi = ctx.artifacts[0].fsvi_path(); + // Derive the actual WAL name from upstream rather than assuming its suffix. let wal = frankensearch::index::wal_path_for(fsvi); for path in [fsvi, wal.as_path(), ctx.artifacts[0].ann_path().unwrap()] { let extension = path.extension().unwrap().to_string_lossy(); std::fs::rename(path, path.with_extension(format!("{extension}-retained"))).unwrap(); } let before = files(temp.path()); - let (actual, _, stats) = set - .search_with_exact_fallback(&ctx, &[1.0, 0.0], 2, None) - .unwrap(); - assert_eq!( - actual.iter().map(|hit| hit.message_id).collect::>(), - vec![2, 1] - ); - assert!(stats.unwrap().exact_fallback.is_none()); - assert!( - set.search_with_exact_fallback(&ctx, &[f32::NAN, 0.0], 2, None) - .is_err() - ); - assert!( - set.search_with_exact_fallback(&ctx, &[1.0], 2, None) - .is_err() - ); + let (actual, _, _) = set.search_with_exact_fallback(&ctx, &[1.0, 0.0], 2, None).unwrap(); + assert_eq!(actual[0].message_id, 2); + assert!(set.search_with_exact_fallback(&ctx, &[f32::NAN, 0.0], 2, None).is_err()); + assert!(set.search_with_exact_fallback(&ctx, &[1.0], 2, None).is_err()); let other = context(ctx.artifacts.as_ref().clone()); - assert!( - set.search_with_exact_fallback(&other, &[1.0, 0.0], 2, None) - .is_err() - ); + assert!(set.search_with_exact_fallback(&other, &[1.0, 0.0], 2, None).is_err()); let (empty, _, stats) = set.search_with_exact_fallback(&ctx, &[], 0, None).unwrap(); assert!(empty.is_empty()); assert!(stats.is_none()); assert_eq!(files(temp.path()), before); } - -#[test] -fn wal_winners_fill_pages_larger_than_the_persisted_main_slab() { - let temp = tempfile::tempdir().unwrap(); - let updates = (1..=64) - .map(|id| (doc(id, 3), [1.0, 0.0])) - .collect::>(); - let ctx = context(vec![shard( - temp.path(), - "delta-heavy", - &[(doc(1_000, 3), [-1.0, 0.0])], - &updates, - )]); - let set = SemanticAnnShardSet::open(Arc::clone(&ctx.artifacts)).unwrap(); - let before = files(temp.path()); - let (hits, retry, stats) = set - .search_with_exact_fallback(&ctx, &[1.0, 0.0], 32, None) - .unwrap(); - assert_eq!( - hits.iter().map(|hit| hit.message_id).collect::>(), - (1..=32).collect::>() - ); - assert!(retry.has_more_candidates); - let stats = stats.unwrap(); - assert_eq!(stats.k_requested, 1, "no corpus-wide vector scan"); - assert!(stats.exact_fallback.is_none()); - assert_eq!(files(temp.path()), before); -} - -#[test] -fn wal_shadowing_is_local_to_its_retained_source_shard() { - let temp = tempfile::tempdir().unwrap(); - let ctx = context(vec![ - shard( - temp.path(), - "shadow-a", - &[(doc(1, 3), [1.0, 0.0])], - &[(doc(1, 3), [0.0, 1.0])], - ), - shard( - temp.path(), - "shadow-b", - &[(doc(1, 3), [0.8, 0.6]), (doc(2, 3), [0.6, 0.8])], - &[], - ), - ]); - let set = SemanticAnnShardSet::open(Arc::clone(&ctx.artifacts)).unwrap(); - let before = files(temp.path()); - let (hits, _, stats) = set - .search_with_exact_fallback(&ctx, &[1.0, 0.0], 2, None) - .unwrap(); - assert_eq!( - hits.iter().map(|hit| hit.message_id).collect::>(), - vec![1, 2] - ); - assert_eq!( - hits[0].score.to_bits(), - 0.8_f32.to_bits(), - "a shard-local replacement must not erase another shard's current score" - ); - assert!(stats.unwrap().exact_fallback.is_none()); - assert_eq!(files(temp.path()), before); -} - -#[test] -fn wal_chunk_replacement_keeps_unmodified_chunks_of_the_same_message() { - let temp = tempfile::tempdir().unwrap(); - let other_chunk = SemanticDocId { - message_id: 1, - chunk_idx: 1, - agent_id: 1, - workspace_id: 2, - source_id: 3, - role: 1, - created_at_ms: 100, - content_hash: None, - } - .to_doc_id_string(); - let ctx = context(vec![shard( - temp.path(), - "chunk-replacement", - &[ - (doc(1, 3), [1.0, 0.0]), - (other_chunk, [0.8, 0.6]), - (doc(2, 3), [0.6, 0.8]), - ], - &[(doc(1, 3), [0.0, 1.0])], - )]); - let set = SemanticAnnShardSet::open(Arc::clone(&ctx.artifacts)).unwrap(); - let before = files(temp.path()); - let (hits, _, stats) = set - .search_with_exact_fallback(&ctx, &[1.0, 0.0], 2, None) - .unwrap(); - assert_eq!( - hits.iter().map(|hit| hit.message_id).collect::>(), - vec![1, 2] - ); - assert_eq!(hits[0].chunk_idx, 1); - assert_eq!(hits[0].score.to_bits(), 0.8_f32.to_bits()); - assert!(stats.unwrap().exact_fallback.is_none()); - assert_eq!(files(temp.path()), before); -} - -#[test] -fn native_refills_reuse_the_prepared_wal_instead_of_filtering_it_again() { - use std::sync::atomic::{AtomicUsize, Ordering}; - - struct CountingSourceFilter { - wal_checks: AtomicUsize, - } - - impl FsSearchFilter for CountingSourceFilter { - fn matches(&self, doc_id: &str, _: Option<&serde_json::Value>) -> bool { - let parsed = parse_semantic_doc_id(doc_id).unwrap(); - if parsed.message_id == 9_999 { - self.wal_checks.fetch_add(1, Ordering::Relaxed); - } - parsed.source_id == 3 - } - - fn name(&self) -> &str { - "counting-wal-source-filter" - } - } - - let temp = tempfile::tempdir().unwrap(); - let mut base = (100..140) - .map(|id| (doc(id, 4), [1.0, 0.0])) - .collect::>(); - base.extend([(doc(1, 3), [0.8, 0.6]), (doc(2, 3), [0.6, 0.8])]); - let ctx = context(vec![shard( - temp.path(), - "refill-cache", - &base, - &[(doc(9_999, 4), [1.0, 0.0])], - )]); - let set = SemanticAnnShardSet::open(Arc::clone(&ctx.artifacts)).unwrap(); - let before = files(temp.path()); - let filter = CountingSourceFilter { - wal_checks: AtomicUsize::new(0), - }; - let (hits, _, stats) = set - .search_with_exact_fallback(&ctx, &[1.0, 0.0], 2, Some(&filter)) - .unwrap(); - assert_eq!( - hits.iter().map(|hit| hit.message_id).collect::>(), - vec![1, 2] - ); - let stats = stats.unwrap(); - assert!( - stats.k_requested > base.len(), - "the fixture must execute multiple native candidate windows" - ); - assert!(stats.exact_fallback.is_none()); - assert_eq!( - filter.wal_checks.load(Ordering::Relaxed), - 1, - "refill passes must reuse one prepared delta" - ); - assert_eq!(files(temp.path()), before); -} - -#[test] -fn oversized_cohort_wal_recovers_exactly_before_any_native_graph_call() { - let temp = tempfile::tempdir().unwrap(); - let mut artifacts = Vec::new(); - for (ordinal, count) in [(0_u64, 2_048_usize), (1_u64, 2_049_usize)] { - let base = (0..4) - .map(|offset| (doc(50_000 + ordinal * 10 + offset, 3), [-1.0, 0.0])) - .collect::>(); - let original = shard(temp.path(), &format!("budget-{ordinal}"), &base, &[]); - let path = original.fsvi_path().to_path_buf(); - let ann = original.ann_path().unwrap().to_path_buf(); - drop(original); - // One durable batch, not thousands of fsyncs. Each shard individually - // fits the row limit, but their combined resident delta does not. - let updates = (0..count) - .map(|offset| { - let score = if offset < 4 { - 1.0 - offset as f32 / 8.0 - ordinal as f32 / 16.0 - } else { - 0.0 - }; - ( - doc(1 + ordinal * 10_000 + offset as u64, 3), - vec![score, (1.0 - score * score).sqrt()], - ) - }) - .collect::>(); - { - let mut writer = FsVectorIndex::open_writer(&path).unwrap(); - writer.append_batch(&updates).unwrap(); - } - artifacts.push(SemanticIndexArtifact::open(&path, Some(ann)).unwrap()); - } - let ctx = context(artifacts); - assert_eq!( - ctx.artifacts - .iter() - .map(|artifact| artifact.index().wal_record_count()) - .sum::(), - 4_097 - ); - let set = SemanticAnnShardSet::open(Arc::clone(&ctx.artifacts)).unwrap(); - let before = files(temp.path()); - let (expected, expected_retry) = - SearchClient::search_exact_semantic_indexes(&ctx, &[1.0, 0.0], 1, None).unwrap(); - let (hits, retry, stats) = set - .search_with_exact_fallback(&ctx, &[1.0, 0.0], 1, None) - .unwrap(); - assert_eq!(signature(&hits), signature(&expected)); - assert_eq!(hits[0].message_id, 1); - assert_eq!( - retry.has_more_candidates, - expected_retry.has_more_candidates - ); - let stats = stats.unwrap(); - assert_eq!(stats.k_requested, 0); - assert_eq!(stats.k_returned, 0); - assert!(!stats.is_approximate); - let receipt = stats.exact_fallback.unwrap(); - assert_eq!(receipt.reason, AnnExactFallbackReason::WalDeltaRequiresExact); - assert_eq!(receipt.shard_count, 2); - assert_eq!(receipt.returned_messages, hits.len()); - assert_eq!(files(temp.path()), before); -} - -#[test] -fn later_native_failure_discards_earlier_graph_and_delta_winners() { - let temp = tempfile::tempdir().unwrap(); - let first = shard( - temp.path(), - "failure-a", - &[(doc(1, 3), [0.6, 0.8])], - &[(doc(10, 3), [0.8, 0.6])], - ); - let second = shard( - temp.path(), - "failure-b", - &[(doc(2, 3), [0.9, 0.4358899])], - &[], - ); - let wrong = shard( - temp.path(), - "wrong-graph", - &[(doc(30, 3), [1.0, 0.0]), (doc(31, 3), [0.0, 1.0])], - &[], - ); - let first_graph = - open_fs_semantic_ann_index(first.index(), first.ann_path().unwrap()).unwrap(); - let wrong_graph = - open_fs_semantic_ann_index(wrong.index(), wrong.ann_path().unwrap()).unwrap(); - let ctx = context(vec![first, second]); - // Exercise a failed private pairing using a real incompatible graph. - // Production admission remains all-or-nothing; no backend is mocked. - let set = SemanticAnnShardSet { - artifacts: Arc::clone(&ctx.artifacts), - graphs: vec![first_graph, wrong_graph], - }; - let before = files(temp.path()); - let (expected, _) = - SearchClient::search_exact_semantic_indexes(&ctx, &[1.0, 0.0], 1, None).unwrap(); - let (hits, _, stats) = set - .search_with_exact_fallback(&ctx, &[1.0, 0.0], 1, None) - .unwrap(); - assert_eq!(signature(&hits), signature(&expected)); - assert_eq!( - hits[0].message_id, - 2, - "the failed later shard must survive recovery" - ); - assert!(hits.iter().all(|hit| ![30, 31].contains(&hit.message_id))); - let stats = stats.unwrap(); - assert_eq!(stats.k_requested, 1, "keep only completed native work"); - assert!(!stats.is_approximate); - let receipt = stats.exact_fallback.unwrap(); - assert_eq!(receipt.reason, AnnExactFallbackReason::NativeSearchFailed); - assert_eq!(receipt.shard_count, 2); - assert_eq!(files(temp.path()), before); -} diff --git a/tests/cli_logical_archive.rs b/tests/cli_logical_archive.rs index 1aaee4dd..2c389000 100644 --- a/tests/cli_logical_archive.rs +++ b/tests/cli_logical_archive.rs @@ -148,70 +148,3 @@ fn real_binary_never_publishes_a_valid_prefix_or_an_unknown_version() { assert!(!target.exists()); } } - -#[cfg(unix)] -#[test] -fn real_binary_verify_refuses_a_fifo_without_waiting_for_a_writer() { - use std::os::unix::fs::FileTypeExt; - let root = tempfile::tempdir().unwrap(); - let input = root.path().join("history.fifo"); - assert!( - std::process::Command::new("mkfifo") - .arg(&input) - .status() - .expect("create FIFO fixture") - .success() - ); - assert!(fs::symlink_metadata(&input).unwrap().file_type().is_fifo()); - // No process opens the write end. The old File::open blocks indefinitely; - // the external deadline prevents the regression itself hanging the suite. - let output = command(root.path()) - .timeout(Duration::from_secs(5)) - .args(["archive", "verify"]) - .arg(&input) - .output() - .expect("verification must return a diagnostic, not time out"); - assert!(output.status.code().is_some(), "verifier was killed instead of refusing input"); - assert!(!output.status.success()); - assert!(output.stdout.is_empty()); - let error: Value = serde_json::from_slice(&output.stderr).expect("one JSON error"); - assert!(error.to_string().contains("regular, non-symlink")); - assert!(fs::symlink_metadata(&input).unwrap().file_type().is_fifo()); - assert!(!root.path().join("unused-default").exists()); -} - -#[cfg(unix)] -#[test] -fn real_binary_verify_preserves_regular_input_and_refuses_a_link_to_it() { - use std::os::unix::fs::symlink; - let root = tempfile::tempdir().unwrap(); - let source = root.path().join("source.db"); - drop(SqliteStorage::open(&source).unwrap()); - let input = root.path().join("history.jsonl"); - let exported = export(root.path(), &source, &input); - let before = fs::read(&input).unwrap(); - let verified = receipt( - command(root.path()) - .args(["archive", "verify"]) - .arg(&input) - .output() - .unwrap(), - ); - assert_eq!(verified["operation"], "verify"); - assert_eq!(verified["content_sha256"], exported["content_sha256"]); - assert_eq!(verified["tables"], exported["tables"]); - let linked = root.path().join("linked.jsonl"); - symlink(&input, &linked).unwrap(); - let output = command(root.path()) - .args(["archive", "verify"]) - .arg(&linked) - .output() - .unwrap(); - assert!(!output.status.success()); - assert!(output.stdout.is_empty()); - let error: Value = serde_json::from_slice(&output.stderr).expect("one JSON error"); - assert!(error.to_string().contains("regular, non-symlink")); - assert_eq!(fs::read(&input).unwrap(), before); - assert_eq!(fs::read_link(&linked).unwrap(), input); - assert!(!root.path().join("unused-default").exists()); -} diff --git a/tests/gh493_message_coordinates.rs b/tests/gh493_message_coordinates.rs index c00b7b0d..fb0957a1 100644 --- a/tests/gh493_message_coordinates.rs +++ b/tests/gh493_message_coordinates.rs @@ -973,74 +973,3 @@ fn shared_path_search_hits_round_trip_in_every_robot_projection() { assert_eq!(std::fs::read(&fixture.db).unwrap(), db_before); assert_eq!(std::fs::read(&fixture.path).unwrap(), source_before); } - -#[test] -fn corrected_robot_failures_are_one_error_envelope_and_success_still_teaches() { - let fixture = Fixture::new(&[0, 7, 12]); - let formats: &[&[&str]] = &[ - &["--json"], - &["--robot"], - &["--robot-format", "json"], - &["--robot-format", "compact"], - &["--robot-format", "jsonl"], - &["--format=json"], - &[], // Environment-selected structured output follows the same contract. - ]; - for subcommand in ["view", "expand"] { - for flags in formats { - let mut command = fixture.command(subcommand); - command - .arg(format!("source_path={}", fixture.path.display())) - .args([ - "source_id=local", - "conversation_id=999999", - "line_number=8", - "context=0", - ]) - .args(*flags); - if flags.is_empty() { - command.env("CASS_OUTPUT_FORMAT", "json"); - } - let output = command.output().unwrap(); - assert!(!output.status.success()); - assert!(output.stdout.is_empty(), "failed recovery emitted a target"); - let error: Value = serde_json::from_slice(&output.stderr).unwrap_or_else(|err| { - panic!( - "{subcommand} {flags:?}: {err}; stderr: {}", - String::from_utf8_lossy(&output.stderr) - ) - }); - assert_eq!(error["error"]["kind"], "indexed-session-required"); - assert_eq!(error["error"]["retryable"], false); - assert!( - error["error"]["hint"] - .as_str() - .is_some_and(|hint| hint.contains("same search hit")) - ); - } - let output = fixture - .command(subcommand) - .arg(format!("source_path={}", fixture.path.display())) - .args(["source_id=local", "line_number=8", "context=0", "--json"]) - .output() - .unwrap(); - assert!( - String::from_utf8_lossy(&output.stderr).contains("note: auto-corrected:"), - "successful robot recovery still teaches the canonical syntax" - ); - assert_target(&decode(output), subcommand, 8, fixture.conversation_id); - - let human = fixture - .command(subcommand) - .arg(format!("source_path={}", fixture.path.display())) - .args(["line_number=8", "conversation_id=999999", "context=0"]) - .output() - .unwrap(); - assert!(!human.status.success()); - assert!( - String::from_utf8_lossy(&human.stderr) - .contains("Note: Your command was auto-corrected:"), - "human diagnostics must retain their teaching note" - ); - } -} diff --git a/tests/spec_field_mask_contract.rs b/tests/spec_field_mask_contract.rs index abc9c0c6..5c356a62 100644 --- a/tests/spec_field_mask_contract.rs +++ b/tests/spec_field_mask_contract.rs @@ -1,22 +1,32 @@ //! INV-cass-21 — `cass search --fields` mask discipline contract. //! -//! Named presets omit message bodies but retain a complete canonical -//! follow-up anchor. GH493 established that a path and message ordinal alone -//! cannot distinguish multiple sessions sharing a provider database. The -//! source and archive-local conversation identity must survive projection. +//! AGENTS.md "Key Flags" documents `--fields minimal` as "Reduce payload: +//! `source_path`, `line_number`, `agent` only" — the lean key set agents +//! pipe through `jq` when context budget matters. Existing tests in +//! `tests/cli_robot.rs::fields_minimal_preset_expands` check that some +//! expected keys are present and a couple of other keys are absent, but +//! they do not lock the **exact** key set, nor do they prove the +//! token-savings promise that justifies using the flag in the first +//! place. //! -//! Four invariants: +//! Three invariants: //! -//! 1. `--fields minimal` emits exactly `agent`, `line_number`, `source_path`, -//! `source_id`, and `conversation_id`. No body, score, or unrelated fields. -//! 2. `--fields summary` adds exactly `title` and `score` to that anchor. -//! 3. `--fields minimal` produces strictly fewer total response bytes than -//! the default. Keeping identity must not defeat the payload savings. -//! 4. Explicit masks still emit exactly the caller's requested fields, even -//! when the caller deliberately omits part of the follow-up anchor. +//! 1. `--fields minimal` emits hits whose key set is **exactly** +//! `{agent, line_number, source_path}` — no extra, no missing. +//! Set equality is the strongest property; a regression that +//! added `score` "for compatibility" would slip past the existing +//! "score is null" check but fail this one. +//! 2. `--fields minimal` produces strictly fewer total response +//! bytes than the default. The whole reason to type the flag. +//! Bytes are a robust proxy for LLM tokens. +//! 3. `--fields ` emits hits whose key set is exactly +//! the requested list. The most powerful form of the flag: an +//! agent that wants only `score` and `source_path` for ranking- +//! adjacent work must be able to ask for those two and only those +//! two. //! -//! Exact key-set checks complement the shared-path, sparse-index round trips -//! in `gh493_message_coordinates`, which verify the identity values themselves. +//! Verified against the checked-in `search_demo_data` fixture with +//! the query `"the"` (2 aider hits). use std::cmp::Ordering; use std::collections::BTreeSet; @@ -139,45 +149,19 @@ fn assert_key_set_equals( } #[test] -fn fields_minimal_preset_emits_exactly_the_canonical_anchor_keys() -> TestResult { +fn fields_minimal_preset_emits_exactly_the_documented_three_keys() -> TestResult { let tmp = TempDir::new()?; let data_dir = copy_search_demo_fixture(tmp.path())?; let (_stdout, parsed) = run_search(&data_dir, &["--fields", "minimal", "--limit", "1"])?; let keys = first_hit_keys(&parsed)?; - let documented: BTreeSet = [ - "agent", - "line_number", - "source_path", - "source_id", - "conversation_id", - ] - .into_iter() - .map(String::from) - .collect(); + let documented: BTreeSet = ["agent", "line_number", "source_path"] + .iter() + .copied() + .map(String::from) + .collect(); assert_key_set_equals("--fields minimal", &keys, &documented) } -#[test] -fn fields_summary_retains_the_exact_anchor_without_message_bodies() -> TestResult { - let tmp = TempDir::new()?; - let data_dir = copy_search_demo_fixture(tmp.path())?; - let (_stdout, parsed) = run_search(&data_dir, &["--fields", "summary", "--limit", "1"])?; - let keys = first_hit_keys(&parsed)?; - let documented: BTreeSet = [ - "agent", - "line_number", - "source_path", - "source_id", - "conversation_id", - "title", - "score", - ] - .into_iter() - .map(String::from) - .collect(); - assert_key_set_equals("--fields summary", &keys, &documented) -} - #[test] fn fields_minimal_strictly_reduces_response_bytes_vs_default() -> TestResult { let tmp = TempDir::new()?; @@ -186,7 +170,9 @@ fn fields_minimal_strictly_reduces_response_bytes_vs_default() -> TestResult { let (minimal_stdout, _) = run_search(&data_dir, &["--fields", "minimal"])?; let default_bytes = default_stdout.len(); let minimal_bytes = minimal_stdout.len(); - // Retaining the complete anchor must not defeat the reason to use minimal. + // The entire agent-facing promise of `--fields minimal` is "Reduce + // payload" (AGENTS.md). A regression where minimal emits at least as + // many bytes as the default defeats the flag's reason for existing. ensure( !matches!( minimal_bytes.cmp(&default_bytes), diff --git a/tests/util/timeout.rs b/tests/util/timeout.rs index bca7d590..5348a5c2 100644 --- a/tests/util/timeout.rs +++ b/tests/util/timeout.rs @@ -176,7 +176,6 @@ pub fn spawn_with_timeout_or_diag( kill_child_process_group(child.id()); let _ = child.kill(); let _ = child.wait(); - retire_harness_death_reaper(&mut reaper); let _ = join_pipe_reader(stdout_reader, label, "stdout"); let _ = join_pipe_reader(stderr_reader, label, "stderr"); panic!("spawn_with_timeout_or_diag({label}): try_wait errored: {err}"); @@ -198,10 +197,8 @@ fn configure_child_process_group(_cmd: &mut Command) {} #[cfg(unix)] fn kill_child_process_group(pid: u32) { let process_group = format!("-{pid}"); - // End option parsing explicitly: the negative operand is a process group, - // not another signal option. The same rule applies to the detached reaper. let _ = Command::new("/bin/kill") - .args(["-KILL", "--", &process_group]) + .args(["-KILL", &process_group]) .stdin(Stdio::null()) .stdout(Stdio::null()) .stderr(Stdio::null()) @@ -229,7 +226,7 @@ fn spawn_death_reaper_for(watched_pid: u32, child_pid: u32) -> Option/dev/null && kill -0 \"$2\" 2>/dev/null; do sleep 1; done; if kill -0 \"$2\" 2>/dev/null; then /bin/kill -KILL -- \"-$2\" 2>/dev/null; fi"; + const REAPER_SCRIPT: &str = "while kill -0 \"$1\" 2>/dev/null && kill -0 \"$2\" 2>/dev/null; do sleep 1; done; if kill -0 \"$2\" 2>/dev/null; then /bin/kill -KILL \"-$2\" 2>/dev/null; fi"; let mut cmd = Command::new("/bin/sh"); cmd.arg("-c") .arg(REAPER_SCRIPT) @@ -402,16 +399,8 @@ mod tests { .stderr(std::process::Stdio::null()); sleeper.process_group(0); let mut child = sleeper.spawn().map_err(|e| format!("spawn sleeper: {e}"))?; - // Observe a process we actually started and reaped. A guessed PID near - // a platform's maximum is not evidence that the process is absent. - let mut watched = std::process::Command::new("/bin/sh") - .args(["-c", "exit 0"]) - .spawn() - .map_err(|e| format!("spawn watched process: {e}"))?; - let dead_pid = watched.id(); - watched - .wait() - .map_err(|e| format!("wait watched process: {e}"))?; + // A pid that cannot be alive: pid_max is at most 2^22 on Linux. + let dead_pid: u32 = 4_194_303; let mut reaper = super::spawn_death_reaper_for(dead_pid, child.id()) .ok_or_else(|| "spawn reaper".to_string())?; let started = std::time::Instant::now(); @@ -472,55 +461,20 @@ mod tests { /// cleanup: killing only the shell would leave `sleep` holding the /// stdout/stderr pipe FDs open, deadlocking the diagnostic join. #[test] + #[should_panic(expected = "exceeded timeout")] fn hung_child_triggers_timeout_panic_with_diagnostic() { - let started = Instant::now(); - let result = std::panic::catch_unwind(|| { - #[cfg(unix)] - let cmd = { - let mut cmd = Command::new("/bin/sh"); - cmd.arg("-c").arg("sleep 30"); - cmd - }; - #[cfg(not(unix))] - let mut cmd = Command::new("/bin/sleep"); - #[cfg(not(unix))] - cmd.arg("30"); - spawn_with_timeout_or_diag(cmd, "intentional_hang", None, Duration::from_millis(300)) - }); - let panic = result.expect_err("a stalled child must time out"); - let message = panic - .downcast_ref::() - .map(String::as_str) - .or_else(|| panic.downcast_ref::<&str>().copied()) - .unwrap_or_default(); - assert!( - message.contains("exceeded timeout"), - "unexpected panic: {message}" - ); - // Previously should_panic accepted a 30-second pipe-drain stall after a - // 300ms timeout. Allow CI scheduling slack, but reject that false pass. - assert!( - started.elapsed() < Duration::from_secs(10), - "timeout cleanup outlived its deadline: {:?}", - started.elapsed() - ); - } - - #[cfg(unix)] - #[test] - fn exited_child_does_not_leave_descendant_output_pipes_open() { - let mut cmd = Command::new("/bin/sh"); - cmd.args(["-c", "sleep 30 & printf ready; exit 0"]); - let started = Instant::now(); - let output = - spawn_with_timeout_or_diag(cmd, "orphaned_pipes", None, Duration::from_secs(2)); - assert!(output.status.success()); - assert_eq!(output.stdout, b"ready"); - assert!( - started.elapsed() < Duration::from_secs(10), - "an exited leader left a descendant holding its pipes: {:?}", - started.elapsed() - ); + #[cfg(unix)] + let cmd = { + let mut cmd = Command::new("/bin/sh"); + cmd.arg("-c").arg("sleep 30"); + cmd + }; + #[cfg(not(unix))] + let mut cmd = Command::new("/bin/sleep"); + #[cfg(not(unix))] + cmd.arg("30"); + let _ = + spawn_with_timeout_or_diag(cmd, "intentional_hang", None, Duration::from_millis(300)); } /// Proves the diagnostic dump includes the data_dir listing when