feat: Pass 1 canary file-write test for fast permission failure detection (v9.3.85)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: jsbattig

CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## v9.3.85
+
+### Enhancements
+
+- feat: Add canary file-write test to Pass 1 dependency map synthesis prompt. Before Claude spends cycles analyzing 125+ repos, it now runs a quick write/delete test on the target directory as Step 0. If the canary fails (OS permission denied or Claude CLI permission restricted), the process bails immediately with a diagnostic RuntimeError instead of wasting 50+ turns attempting analysis and retrying different write methods. Saves significant time and API tokens when permission issues exist.
+
 ## v9.3.84
 
 ### Bug Fixes

README.md

@@ -2,7 +2,7 @@
 
 AI-powered semantic code search for your codebase. Find code by meaning, not just keywords.
 
-**Version 9.3.84** - [Changelog](CHANGELOG.md) | [Migration Guide](docs/migration-to-v8.md) | [Architecture](docs/architecture.md)
+**Version 9.3.85** - [Changelog](CHANGELOG.md) | [Migration Guide](docs/migration-to-v8.md) | [Architecture](docs/architecture.md)
 
 ## Quick Navigation
 

src/code_indexer/__init__.py

@@ -6,5 +6,5 @@
 HNSW graph indexing (O(log N) complexity).
 """
 
-__version__ = "9.3.84"
+__version__ = "9.3.85"
 __author__ = "Seba Battig"

src/code_indexer/global_repos/dependency_map_analyzer.py

@@ -183,6 +183,7 @@ def run_pass_1_synthesis(
             # If staging_dir is not under golden_repos_root (e.g. in tests), use absolute
             pass1_file_rel = pass1_file
         pass1_file_abs = str(pass1_file)
+        staging_dir_abs = str(staging_dir)
 
         # Build synthesis prompt — output format + file instructions FIRST (primacy/recency)
         prompt = "# Domain Synthesis Task\n\n"
@@ -200,15 +201,25 @@ def run_pass_1_synthesis(
         prompt += '"evidence": "Brief justification referencing actual files/patterns observed"}\n'
         prompt += "]\n\n"
         prompt += "### File Output Instructions\n\n"
-        prompt += "1. Write the JSON array to this file path:\n"
+        prompt += "**STEP 0 — MANDATORY CANARY TEST (do this FIRST, before any analysis):**\n"
+        prompt += "Before doing ANY analysis work, test that you can write files by running:\n"
+        prompt += "```\n"
+        prompt += f"echo 'canary' > {pass1_file_abs}.canary && rm {pass1_file_abs}.canary && echo 'CANARY_OK'\n"
+        prompt += "```\n"
+        prompt += "- If you see `CANARY_OK`: proceed with analysis and file writing normally.\n"
+        prompt += "- If the write FAILS for ANY reason: STOP IMMEDIATELY. Do NOT attempt analysis. "
+        prompt += "Output ONLY this exact line to stdout:\n"
+        prompt += f"  `CANARY_FAIL: Cannot write to {staging_dir_abs} — [reason: OS permission denied | Claude permission denied | other]`\n"
+        prompt += "  Then exit. Do NOT retry with other write methods. Do NOT proceed with analysis.\n\n"
+        prompt += "**STEP 1** — Write the JSON array to this file path:\n"
         prompt += f"   - Relative from your cwd: `./{pass1_file_rel}`\n"
         prompt += f"   - Absolute path: `{pass1_file_abs}`\n\n"
-        prompt += "2. Validate the file with:\n"
+        prompt += "**STEP 2** — Validate the file with:\n"
         prompt += "   ```\n"
         prompt += f"   python3 -m json.tool {pass1_file_abs}\n"
         prompt += "   ```\n\n"
-        prompt += "3. If validation fails, fix the JSON errors and re-validate until it passes.\n\n"
-        prompt += "4. PREFERRED: Write to the file above. FALLBACK: If file writing is blocked by permissions, output ONLY the raw JSON array to stdout (no explanation, no commentary).\n\n"
+        prompt += "**STEP 3** — If validation fails, fix the JSON errors and re-validate until it passes.\n\n"
+        prompt += "**FALLBACK**: If file writing worked in the canary test but fails later during the actual write, output ONLY the raw JSON array to stdout (no explanation, no commentary).\n\n"
 
         # ── REPOSITORY DESCRIPTIONS (after file instructions) ──
         prompt += "## Repository Descriptions\n\n"
@@ -307,6 +318,20 @@ def run_pass_1_synthesis(
         )  # Pass 1 uses full timeout (heaviest phase: explores all repos)
         result = self._invoke_claude_cli(prompt, timeout, max_turns, allowed_tools=None, dangerously_skip_permissions=True)
 
+        # ── Canary failure fast-path ──
+        if "CANARY_FAIL" in result:
+            canary_msg = result.strip()
+            # Extract just the CANARY_FAIL line
+            for line in canary_msg.split("\n"):
+                if "CANARY_FAIL" in line:
+                    canary_msg = line.strip()
+                    break
+            raise RuntimeError(
+                f"Pass 1 file-write canary test failed: {canary_msg}. "
+                f"Claude CLI cannot write to {staging_dir_abs}. "
+                f"Check --dangerously-skip-permissions flag and OS file permissions."
+            )
+
         # ── File-based output: primary path (Story #349) ──
         logger.debug(f"Pass 1 raw output length: {len(result)} chars")
         domain_list = None

tests/unit/global_repos/test_pass1_file_output.py

@@ -555,3 +555,151 @@ def fake_invoke(prompt, timeout, max_turns, allowed_tools=None, **kwargs):
 
         assert call_count[0] == 2
         assert len(result) >= 1
+
+
+# ─── Canary file write test ───────────────────────────────────────────────────
+
+
+class TestPass1CanaryFileWriteTest:
+    """Tests for the STEP 0 mandatory canary file write test added to the Pass 1 prompt."""
+
+    def _capture_prompt(self, analyzer, staging_dir, repo_descriptions, repo_list, valid_domain_json):
+        """Helper: run pass 1 and capture the prompt sent to Claude CLI."""
+        captured_prompt = {}
+
+        def fake_invoke(prompt, timeout, max_turns, allowed_tools=None, **kwargs):
+            captured_prompt["value"] = prompt
+            return valid_domain_json
+
+        analyzer._invoke_claude_cli = fake_invoke
+        analyzer.run_pass_1_synthesis(staging_dir, repo_descriptions, repo_list, max_turns=10)
+        return captured_prompt["value"]
+
+    def test_prompt_contains_canary_test_command(
+        self, analyzer, staging_dir, repo_descriptions, repo_list, valid_domain_json
+    ):
+        """Prompt must include the canary echo command to test file writing."""
+        prompt = self._capture_prompt(analyzer, staging_dir, repo_descriptions, repo_list, valid_domain_json)
+
+        assert "CANARY_OK" in prompt
+        assert ".canary" in prompt
+        assert "echo 'canary'" in prompt
+
+    def test_prompt_contains_canary_fail_instruction(
+        self, analyzer, staging_dir, repo_descriptions, repo_list, valid_domain_json
+    ):
+        """Prompt must include the CANARY_FAIL output instruction."""
+        prompt = self._capture_prompt(analyzer, staging_dir, repo_descriptions, repo_list, valid_domain_json)
+
+        assert "CANARY_FAIL" in prompt
+        assert "STOP IMMEDIATELY" in prompt
+
+    def test_prompt_canary_includes_staging_dir_path(
+        self, analyzer, staging_dir, repo_descriptions, repo_list, valid_domain_json
+    ):
+        """Prompt CANARY_FAIL line must include the staging directory path."""
+        prompt = self._capture_prompt(analyzer, staging_dir, repo_descriptions, repo_list, valid_domain_json)
+
+        staging_dir_abs = str(staging_dir)
+        assert staging_dir_abs in prompt
+
+    def test_canary_fail_in_output_raises_runtime_error(
+        self, analyzer, staging_dir, repo_descriptions, repo_list
+    ):
+        """When CANARY_FAIL appears in output, RuntimeError is raised immediately."""
+        canary_fail_output = (
+            "CANARY_FAIL: Cannot write to /some/dir — [reason: Claude permission denied]"
+        )
+
+        def fake_invoke(prompt, timeout, max_turns, allowed_tools=None, **kwargs):
+            return canary_fail_output
+
+        analyzer._invoke_claude_cli = fake_invoke
+
+        with pytest.raises(RuntimeError) as exc_info:
+            analyzer.run_pass_1_synthesis(staging_dir, repo_descriptions, repo_list, max_turns=10)
+
+        error_msg = str(exc_info.value)
+        assert "canary" in error_msg.lower()
+        assert "CANARY_FAIL" in error_msg
+
+    def test_canary_fail_error_message_includes_staging_dir(
+        self, analyzer, staging_dir, repo_descriptions, repo_list
+    ):
+        """RuntimeError from CANARY_FAIL must include the staging directory path."""
+        canary_fail_output = (
+            "CANARY_FAIL: Cannot write to /some/dir — [reason: OS permission denied]"
+        )
+
+        def fake_invoke(prompt, timeout, max_turns, allowed_tools=None, **kwargs):
+            return canary_fail_output
+
+        analyzer._invoke_claude_cli = fake_invoke
+
+        with pytest.raises(RuntimeError) as exc_info:
+            analyzer.run_pass_1_synthesis(staging_dir, repo_descriptions, repo_list, max_turns=10)
+
+        error_msg = str(exc_info.value)
+        staging_dir_abs = str(staging_dir)
+        assert staging_dir_abs in error_msg, (
+            f"RuntimeError must include staging dir path. Got: {error_msg}"
+        )
+
+    def test_canary_fail_does_not_retry(
+        self, analyzer, staging_dir, repo_descriptions, repo_list
+    ):
+        """CANARY_FAIL must raise immediately — no retry attempt."""
+        call_count = [0]
+
+        def fake_invoke(prompt, timeout, max_turns, allowed_tools=None, **kwargs):
+            call_count[0] += 1
+            return "CANARY_FAIL: Cannot write — permission denied"
+
+        analyzer._invoke_claude_cli = fake_invoke
+
+        with pytest.raises(RuntimeError):
+            analyzer.run_pass_1_synthesis(staging_dir, repo_descriptions, repo_list, max_turns=10)
+
+        assert call_count[0] == 1, (
+            f"CANARY_FAIL must not trigger retry. Expected 1 call, got {call_count[0]}"
+        )
+
+    def test_canary_fail_multiline_output_extracts_fail_line(
+        self, analyzer, staging_dir, repo_descriptions, repo_list
+    ):
+        """CANARY_FAIL detection works when the fail line is embedded in multiline output."""
+        multiline_output = (
+            "Running canary test...\n"
+            "echo 'canary' > /path/to/file.canary\n"
+            "CANARY_FAIL: Cannot write to /path — [reason: OS permission denied]\n"
+            "Exiting as instructed."
+        )
+
+        def fake_invoke(prompt, timeout, max_turns, allowed_tools=None, **kwargs):
+            return multiline_output
+
+        analyzer._invoke_claude_cli = fake_invoke
+
+        with pytest.raises(RuntimeError) as exc_info:
+            analyzer.run_pass_1_synthesis(staging_dir, repo_descriptions, repo_list, max_turns=10)
+
+        error_msg = str(exc_info.value)
+        # Should extract just the CANARY_FAIL line
+        assert "CANARY_FAIL" in error_msg
+
+    def test_normal_output_without_canary_fail_proceeds_normally(
+        self, analyzer, staging_dir, repo_descriptions, repo_list, valid_domain_json
+    ):
+        """Output without CANARY_FAIL proceeds through normal file/stdout parsing."""
+        pass1_file = staging_dir / "pass1_domains.json"
+
+        def fake_invoke(prompt, timeout, max_turns, allowed_tools=None, **kwargs):
+            # Normal output with CANARY_OK in it (but no CANARY_FAIL)
+            pass1_file.write_text(valid_domain_json)
+            return "CANARY_OK\nFile written successfully."
+
+        analyzer._invoke_claude_cli = fake_invoke
+        result = analyzer.run_pass_1_synthesis(staging_dir, repo_descriptions, repo_list, max_turns=10)
+
+        assert len(result) == 1
+        assert result[0]["name"] == "authentication"