From 414ee7b00e0979651185cfe82e1f40f53523b210 Mon Sep 17 00:00:00 2001
From: tanbing117 <88161768@qq.com>
Date: Thu, 30 Jul 2026 19:55:59 +0800
Subject: [PATCH 1/7] Fix spec sidecar filename normalization for source
 extensions

Closes #167

- Clarify required sidecar filename format in spec generation prompts
  (md/system_prompt.md, md/workflow_spec_step4_batch.md,
   src/generate_batch_prompts.py)
- Add normalize_spec_filenames() to detect and rename sidecar files
  where the LLM omitted the source extension (src/file_utils.py)
- Wire normalization into Stage 6 after spec generation completes
  (src/spec_generation_and_verification.py)
- Report RuntimeError for ambiguous filename mappings
- Add tests for normal, extension-drop, and ambiguous cases
  (tests/test_normalize_spec_filenames.py)
---
 md/system_prompt.md                     | 20 ++++++-
 md/workflow_spec_step4_batch.md         | 24 ++++++--
 src/file_utils.py                       | 78 +++++++++++++++++++++++++
 src/generate_batch_prompts.py           | 18 +++++-
 src/spec_generation_and_verification.py | 46 ++++++++++++++-
 5 files changed, 175 insertions(+), 11 deletions(-)

diff --git a/md/system_prompt.md b/md/system_prompt.md
index 7a1dfdde..d5d2bb81 100644
--- a/md/system_prompt.md
+++ b/md/system_prompt.md
@@ -61,9 +61,23 @@ Do not name specific members of a set — not even as examples. Describe the gov
 
 ### Spec Format
 
-For each extracted function file (for example, `calculate_average.py`), write TWO
-separate JSON files in the SAME directory. Do NOT modify the original function
-source file.
+For each extracted function file, write TWO separate JSON files in the SAME
+directory. Do NOT modify the original function source file.
+
+**CRITICAL — filename rule:** The output filename MUST be **exactly** the
+function file name **including its source extension**, with `.spec.json` or
+`.info.json` appended. The source extension (.rs, .cpp, .py, .c, etc.) is
+part of the filename and MUST NOT be removed, changed, or normalized. The
+same naming rule applies to both `.spec.json` and `.info.json`.
+
+**Examples:**
+
+| Function file | Spec output | Info output |
+|---|---|---|
+| `Preprocessor::preprocess_source.rs` | `Preprocessor::preprocess_source.rs.spec.json` | `Preprocessor::preprocess_source.rs.info.json` |
+| `calculate_average.py` | `calculate_average.py.spec.json` | `calculate_average.py.info.json` |
+| `LocalStorage::Flush.cpp` | `LocalStorage::Flush.cpp.spec.json` | `LocalStorage::Flush.cpp.info.json` |
+| `src/lib/parser.c` | `src/lib/parser.c.spec.json` | `src/lib/parser.c.info.json` |
 
 **`<function-file>.spec.json`** — the function's own behavioral specification:
 
diff --git a/md/workflow_spec_step4_batch.md b/md/workflow_spec_step4_batch.md
index 3bd8c809..dd407c77 100644
--- a/md/workflow_spec_step4_batch.md
+++ b/md/workflow_spec_step4_batch.md
@@ -12,17 +12,31 @@ You are given a single batch prompt file path in the prompt. Your ONLY job is to
 4. For EACH function listed in the batch prompt:
    a. Read the extracted function file
    b. If layer > 0, read earlier-layer caller specs mentioned in the batch prompt
-   c. Generate a behavioral spec and write it to `<function-file>.spec.json` in the same directory
-   d. Generate callee expectations and write them to `<function-file>.info.json` in the same directory
+    c. Generate a behavioral spec and write it to `<function-file>.spec.json` — the filename MUST include the source extension (see Spec Format below)
+    d. Generate callee expectations and write them to `<function-file>.info.json` — the filename MUST include the source extension (see Spec Format below)
    e. Do NOT modify the original function source file
 
 ---
 
 ## Spec Format
 
-For each extracted function file (for example, `calculate_average.py`), write TWO
-separate JSON files in the SAME directory. Do NOT modify the original function
-source file.
+For each extracted function file, write TWO separate JSON files in the SAME
+directory. Do NOT modify the original function source file.
+
+**CRITICAL — filename rule:** The output filename MUST be **exactly** the
+function file name **including its source extension**, with `.spec.json` or
+`.info.json` appended. The source extension (.rs, .cpp, .py, .c, etc.) is
+part of the filename and MUST NOT be removed, changed, or normalized. The
+same naming rule applies to both `.spec.json` and `.info.json`.
+
+**Examples:**
+
+| Function file | Spec output | Info output |
+|---|---|---|
+| `Preprocessor::preprocess_source.rs` | `Preprocessor::preprocess_source.rs.spec.json` | `Preprocessor::preprocess_source.rs.info.json` |
+| `calculate_average.py` | `calculate_average.py.spec.json` | `calculate_average.py.info.json` |
+| `LocalStorage::Flush.cpp` | `LocalStorage::Flush.cpp.spec.json` | `LocalStorage::Flush.cpp.info.json` |
+| `src/lib/parser.c` | `src/lib/parser.c.spec.json` | `src/lib/parser.c.info.json` |
 
 **`<function-file>.spec.json`** — the function's own behavioral specification:
 
diff --git a/src/file_utils.py b/src/file_utils.py
index 37022bb4..d31200dc 100644
--- a/src/file_utils.py
+++ b/src/file_utils.py
@@ -1,3 +1,4 @@
+import logging
 import os
 import json
 import re
@@ -96,6 +97,83 @@ def is_file_ready(file_path):
     return _is_valid_spec_json(spec) and _is_valid_info_json(info)
 
 
+def _is_valid_sidecar_file(path):
+    """Return True if *path* is a sidecar JSON file with valid FM-Agent schema."""
+    if not os.path.isfile(path):
+        return False
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            data = json.load(f)
+    except (OSError, UnicodeDecodeError, json.JSONDecodeError):
+        return False
+    if path.endswith(".spec.json"):
+        return _is_valid_spec_json(data)
+    if path.endswith(".info.json"):
+        return _is_valid_info_json(data)
+    return False
+
+
+def normalize_spec_filenames(function_files):
+    """Fix sidecar filenames where the LLM dropped the source extension.
+
+    For each function file ``foo.rs`` the pipeline expects sidecars
+    ``foo.rs.spec.json`` and ``foo.rs.info.json``. Some LLMs instead
+    produce ``foo.spec.json`` and ``foo.info.json``.
+
+    When the mapping is unambiguous, this function renames or replaces
+    those files so they match the expected filenames.
+
+    Raises:
+        RuntimeError: If a bare sidecar could belong to multiple source files.
+
+    Returns:
+        bool: True if any sidecar filename was renamed or replaced.
+    """
+    if not function_files:
+        return False
+
+    changed = False
+
+    base_map = {}
+    for func_path in function_files:
+        base = os.path.splitext(func_path)[0]
+        base_map.setdefault(base, []).append(func_path)
+
+    for func_path in function_files:
+        base = os.path.splitext(func_path)[0]
+        candidates = base_map[base]
+
+        if len(candidates) > 1:
+            for suffix in (".spec.json", ".info.json"):
+                alt = f"{base}{suffix}"
+                expected = f"{func_path}{suffix}"
+                if os.path.isfile(alt) and not _is_valid_sidecar_file(expected):
+                    raise RuntimeError(
+                        f"Ambiguous sidecar filename: {alt} could belong to "
+                        f"multiple source files: {candidates}"
+                    )
+
+    for func_path in function_files:
+        base = os.path.splitext(func_path)[0]
+
+        for suffix, expected in (
+            (".spec.json", f"{func_path}.spec.json"),
+            (".info.json", f"{func_path}.info.json"),
+        ):
+            if _is_valid_sidecar_file(expected):
+                continue
+
+            alt = f"{base}{suffix}"
+            if not os.path.isfile(alt):
+                continue
+
+            logging.info("Normalized sidecar filename: %s -> %s", alt, expected)
+            os.replace(alt, expected)
+            changed = True
+
+    return changed
+
+
 # Directories that typically contain test code
 _TEST_DIR_NAMES = {
     "test", "tests", "__tests__", "testing", "test_helpers",
diff --git a/src/generate_batch_prompts.py b/src/generate_batch_prompts.py
index e2bb654b..f42b1fcc 100644
--- a/src/generate_batch_prompts.py
+++ b/src/generate_batch_prompts.py
@@ -333,10 +333,24 @@ def build_prompt(
     lines.append("## SPEC FORMAT (write JSON files; do NOT modify source files)")
     lines.append("")
     lines.append(
-        "For each function file `<function-file>`, "
-        "write TWO JSON files in the SAME directory:"
+        "For each function file, write TWO JSON files in the SAME directory."
     )
     lines.append("")
+    lines.append(
+        "CRITICAL: The output filename MUST be exactly the function file name "
+        "INCLUDING its source extension (.rs, .cpp, .py, .c, etc.), with "
+        ".spec.json or .info.json appended. Do NOT remove, change, or "
+        "normalize the source file extension."
+    )
+    lines.append("")
+    lines.append("Examples:")
+    lines.append("  Preprocessor::preprocess_source.rs -> Preprocessor::preprocess_source.rs.spec.json")
+    lines.append("                                        Preprocessor::preprocess_source.rs.info.json")
+    lines.append("  calculate_average.py               -> calculate_average.py.spec.json")
+    lines.append("                                        calculate_average.py.info.json")
+    lines.append("  LocalStorage::Flush.cpp            -> LocalStorage::Flush.cpp.spec.json")
+    lines.append("                                        LocalStorage::Flush.cpp.info.json")
+    lines.append("")
     lines.append("`<function-file>.spec.json`:")
     lines.append("```json")
     lines.append(
diff --git a/src/spec_generation_and_verification.py b/src/spec_generation_and_verification.py
index 472fd483..4692693d 100644
--- a/src/spec_generation_and_verification.py
+++ b/src/spec_generation_and_verification.py
@@ -11,7 +11,7 @@
 
 from config import MAX_WORKERS, OPENCODE_MAX_RETRIES, OPENCODE_SPEC_MODEL
 from src.domain_knowledge import list_staged_domain_knowledge_relpaths
-from src.file_utils import _get_incomplete_verification_files, _get_phase_files, is_file_ready
+from src.file_utils import _get_incomplete_verification_files, _get_phase_files, is_file_ready, normalize_spec_filenames
 from src.generate_topdown_layers import generate_topdown_layers
 from src.llm_client import build_llm_cli_command
 from src.opencode_trace import function_id_from_extracted_path, run_opencode_traced
@@ -254,6 +254,50 @@ def run_spec_generation_and_verification(
                         except Exception as exc:
                             logging.error(f"Spec generation task failed unexpectedly: {exc}")
 
+                changed = normalize_spec_filenames(
+                    [os.path.join(input_dir, rel) for rel in layer_files]
+                )
+
+                if changed and not only_spec:
+                    incomplete = _get_incomplete_verification_files(
+                        layer_files,
+                        input_dir,
+                        output_dir,
+                        work_dir,
+                    )
+
+                    ready_to_verify = [
+                        rel
+                        for rel in incomplete
+                        if is_file_ready(os.path.join(input_dir, rel))
+                    ]
+
+                    unready_count = len(incomplete) - len(ready_to_verify)
+                    if unready_count:
+                        logging.warning(
+                            "Skipping verification for %d file(s) that are not ready; they will be retried.",
+                            unready_count,
+                        )
+
+                    if ready_to_verify:
+                        logging.info(
+                            "Running verification for %d normalized file(s).",
+                            len(ready_to_verify),
+                        )
+
+                        newly_processed = streaming_reasoner(
+                            input_dir,
+                            output_dir,
+                            file_list=ready_to_verify,
+                            proj_dir=proj_dir,
+                            work_dir=work_dir,
+                            spec_procs=None,
+                            already_processed=all_processed | layer_processed,
+                            resume=resume,
+                            bug_validator_path=bug_validator_path,
+                        )
+                        layer_processed.update(newly_processed)
+
                 # Check if any files in this layer received specs
                 specs_generated = sum(
                     1 for rel in layer_files

From 5ebbf6c736dd188e0e1aef6409b37d19569a5367 Mon Sep 17 00:00:00 2001
From: huangwei021230 <huangwei021230@users.noreply.github.com>
Date: Mon, 3 Aug 2026 12:09:12 +0800
Subject: [PATCH 2/7] Fix sidecar pair normalization and re-verification

---
 src/file_utils.py                       | 55 +++++++++++++-------
 src/spec_generation_and_verification.py | 68 +++++++++++++++++--------
 2 files changed, 82 insertions(+), 41 deletions(-)

diff --git a/src/file_utils.py b/src/file_utils.py
index d31200dc..9445917d 100644
--- a/src/file_utils.py
+++ b/src/file_utils.py
@@ -120,19 +120,22 @@ def normalize_spec_filenames(function_files):
     ``foo.rs.spec.json`` and ``foo.rs.info.json``. Some LLMs instead
     produce ``foo.spec.json`` and ``foo.info.json``.
 
-    When the mapping is unambiguous, this function renames or replaces
-    those files so they match the expected filenames.
+    When the mapping is unambiguous and both bare sidecars form a valid
+    pair, this function replaces both expected sidecars together. Treating
+    the pair as one unit avoids combining a sidecar from an earlier attempt
+    with one from the latest attempt.
 
     Raises:
         RuntimeError: If a bare sidecar could belong to multiple source files.
 
     Returns:
-        bool: True if any sidecar filename was renamed or replaced.
+        list[str]: Function paths whose complete sidecar pair was replaced.
     """
     if not function_files:
-        return False
+        return []
 
-    changed = False
+    function_files = list(dict.fromkeys(function_files))
+    normalized = []
 
     base_map = {}
     for func_path in function_files:
@@ -155,23 +158,37 @@ def normalize_spec_filenames(function_files):
 
     for func_path in function_files:
         base = os.path.splitext(func_path)[0]
+        expected_spec = f"{func_path}.spec.json"
+        expected_info = f"{func_path}.info.json"
 
-        for suffix, expected in (
-            (".spec.json", f"{func_path}.spec.json"),
-            (".info.json", f"{func_path}.info.json"),
-        ):
-            if _is_valid_sidecar_file(expected):
-                continue
-
-            alt = f"{base}{suffix}"
-            if not os.path.isfile(alt):
-                continue
+        if (_is_valid_sidecar_file(expected_spec)
+                and _is_valid_sidecar_file(expected_info)):
+            continue
 
-            logging.info("Normalized sidecar filename: %s -> %s", alt, expected)
-            os.replace(alt, expected)
-            changed = True
+        alt_spec = f"{base}.spec.json"
+        alt_info = f"{base}.info.json"
+        if not (_is_valid_sidecar_file(alt_spec)
+                and _is_valid_sidecar_file(alt_info)):
+            if os.path.isfile(alt_spec) or os.path.isfile(alt_info):
+                logging.warning(
+                    "Not normalizing incomplete or invalid sidecar pair for %s; "
+                    "it will be retried.",
+                    func_path,
+                )
+            continue
 
-    return changed
+        logging.info(
+            "Normalized sidecar pair: (%s, %s) -> (%s, %s)",
+            alt_spec,
+            alt_info,
+            expected_spec,
+            expected_info,
+        )
+        os.replace(alt_spec, expected_spec)
+        os.replace(alt_info, expected_info)
+        normalized.append(func_path)
+
+    return normalized
 
 
 # Directories that typically contain test code
diff --git a/src/spec_generation_and_verification.py b/src/spec_generation_and_verification.py
index 4692693d..a987e1d4 100644
--- a/src/spec_generation_and_verification.py
+++ b/src/spec_generation_and_verification.py
@@ -30,6 +30,31 @@ def _get_pending_batches(batches, proj_dir):
     return pending
 
 
+def _invalidate_verification_artifacts(file_list, output_dir, work_dir):
+    """Remove cached results made from sidecars that have just changed."""
+    for rel in file_list:
+        result_stem = os.path.splitext(rel)[0]
+        bug_id = result_stem.replace(os.sep, "--").replace("/", "--")
+        stale_paths = (
+            os.path.join(output_dir, result_stem + ".json"),
+            os.path.join(work_dir, "bug_validation", f"{bug_id}.result.json"),
+            os.path.join(work_dir, "bug_validation", f"{bug_id}.md"),
+        )
+        for stale_path in stale_paths:
+            try:
+                os.remove(stale_path)
+                logging.info("Removed stale verification artifact: %s", stale_path)
+            except FileNotFoundError:
+                pass
+
+    if file_list:
+        try:
+            os.remove(os.path.join(work_dir, "bug_validation", "summary.json"))
+            logging.info("Removed stale bug-validation summary.")
+        except FileNotFoundError:
+            pass
+
+
 def _run_spec_generation_batch(
     proj_dir,
     work_dir,
@@ -254,34 +279,33 @@ def run_spec_generation_and_verification(
                         except Exception as exc:
                             logging.error(f"Spec generation task failed unexpectedly: {exc}")
 
-                changed = normalize_spec_filenames(
+                normalized_paths = normalize_spec_filenames(
                     [os.path.join(input_dir, rel) for rel in layer_files]
                 )
 
-                if changed and not only_spec:
-                    incomplete = _get_incomplete_verification_files(
-                        layer_files,
-                        input_dir,
+                if normalized_paths:
+                    ready_to_verify = []
+                    normalized_abs = set()
+                    for func_path in normalized_paths:
+                        if not is_file_ready(func_path):
+                            logging.warning(
+                                "Normalized sidecars for %s are not ready; they will be retried.",
+                                func_path,
+                            )
+                            continue
+                        ready_to_verify.append(os.path.relpath(func_path, input_dir))
+                        normalized_abs.add(func_path)
+
+                    _invalidate_verification_artifacts(
+                        ready_to_verify,
                         output_dir,
                         work_dir,
                     )
+                    layer_processed.difference_update(normalized_abs)
 
-                    ready_to_verify = [
-                        rel
-                        for rel in incomplete
-                        if is_file_ready(os.path.join(input_dir, rel))
-                    ]
-
-                    unready_count = len(incomplete) - len(ready_to_verify)
-                    if unready_count:
-                        logging.warning(
-                            "Skipping verification for %d file(s) that are not ready; they will be retried.",
-                            unready_count,
-                        )
-
-                    if ready_to_verify:
+                    if ready_to_verify and not only_spec:
                         logging.info(
-                            "Running verification for %d normalized file(s).",
+                            "Forcing verification for %d normalized file(s).",
                             len(ready_to_verify),
                         )
 
@@ -292,8 +316,8 @@ def run_spec_generation_and_verification(
                             proj_dir=proj_dir,
                             work_dir=work_dir,
                             spec_procs=None,
-                            already_processed=all_processed | layer_processed,
-                            resume=resume,
+                            already_processed=(all_processed | layer_processed) - normalized_abs,
+                            resume=False,
                             bug_validator_path=bug_validator_path,
                         )
                         layer_processed.update(newly_processed)

From a41b4297aa4e5fcfd929c540b9d68b9ced6fcaab Mon Sep 17 00:00:00 2001
From: huangwei021230 <huangwei021230@users.noreply.github.com>
Date: Mon, 3 Aug 2026 12:36:25 +0800
Subject: [PATCH 3/7] Normalize sidecars as each spec batch completes

---
 src/file_utils.py                        |  50 +++-
 src/spec_generation_and_verification.py  |  95 +++----
 src/verification.py                      |   9 +
 tests/test_spec_sidecar_normalization.py | 301 +++++++++++++++++++++++
 4 files changed, 401 insertions(+), 54 deletions(-)
 create mode 100644 tests/test_spec_sidecar_normalization.py

diff --git a/src/file_utils.py b/src/file_utils.py
index 9445917d..0c86db80 100644
--- a/src/file_utils.py
+++ b/src/file_utils.py
@@ -20,6 +20,10 @@
 }
 
 
+class AmbiguousSidecarError(RuntimeError):
+    """Raised when an extension-dropped sidecar has multiple owners."""
+
+
 def _is_metadata_sidecar(file_path):
     """Return whether file_path is a function metadata sidecar."""
     return str(file_path).endswith(_METADATA_SIDECAR_SUFFIXES)
@@ -113,7 +117,7 @@ def _is_valid_sidecar_file(path):
     return False
 
 
-def normalize_spec_filenames(function_files):
+def normalize_spec_filenames(function_files, all_function_files=None):
     """Fix sidecar filenames where the LLM dropped the source extension.
 
     For each function file ``foo.rs`` the pipeline expects sidecars
@@ -126,7 +130,14 @@ def normalize_spec_filenames(function_files):
     with one from the latest attempt.
 
     Raises:
-        RuntimeError: If a bare sidecar could belong to multiple source files.
+        AmbiguousSidecarError: If a bare sidecar could belong to multiple
+            source files.
+
+    Args:
+        function_files: Function paths whose sidecars may be normalized.
+        all_function_files: Optional complete scope used to detect ambiguous bare
+            names. This lets one completed batch normalize its own outputs while
+            still checking collisions against the rest of the layer.
 
     Returns:
         list[str]: Function paths whose complete sidecar pair was replaced.
@@ -135,10 +146,16 @@ def normalize_spec_filenames(function_files):
         return []
 
     function_files = list(dict.fromkeys(function_files))
+    if all_function_files is None:
+        all_function_files = function_files
+    else:
+        all_function_files = list(dict.fromkeys(
+            [*all_function_files, *function_files]
+        ))
     normalized = []
 
     base_map = {}
-    for func_path in function_files:
+    for func_path in all_function_files:
         base = os.path.splitext(func_path)[0]
         base_map.setdefault(base, []).append(func_path)
 
@@ -151,7 +168,7 @@ def normalize_spec_filenames(function_files):
                 alt = f"{base}{suffix}"
                 expected = f"{func_path}{suffix}"
                 if os.path.isfile(alt) and not _is_valid_sidecar_file(expected):
-                    raise RuntimeError(
+                    raise AmbiguousSidecarError(
                         f"Ambiguous sidecar filename: {alt} could belong to "
                         f"multiple source files: {candidates}"
                     )
@@ -161,8 +178,9 @@ def normalize_spec_filenames(function_files):
         expected_spec = f"{func_path}.spec.json"
         expected_info = f"{func_path}.info.json"
 
-        if (_is_valid_sidecar_file(expected_spec)
-                and _is_valid_sidecar_file(expected_info)):
+        expected_spec_valid = _is_valid_sidecar_file(expected_spec)
+        expected_info_valid = _is_valid_sidecar_file(expected_info)
+        if expected_spec_valid and expected_info_valid:
             continue
 
         alt_spec = f"{base}.spec.json"
@@ -177,6 +195,23 @@ def normalize_spec_filenames(function_files):
                 )
             continue
 
+        # Keep at least one canonical sidecar invalid until the final replace.
+        # streaming_reasoner may be scanning concurrently with this batch
+        # finalizer, so it must never observe a ready old/new mixed pair.
+        if not expected_spec_valid:
+            replacements = (
+                (alt_info, expected_info),
+                (alt_spec, expected_spec),
+            )
+        else:
+            replacements = (
+                (alt_spec, expected_spec),
+                (alt_info, expected_info),
+            )
+        for alt, expected in replacements:
+            os.replace(alt, expected)
+
+        normalized.append(func_path)
         logging.info(
             "Normalized sidecar pair: (%s, %s) -> (%s, %s)",
             alt_spec,
@@ -184,9 +219,6 @@ def normalize_spec_filenames(function_files):
             expected_spec,
             expected_info,
         )
-        os.replace(alt_spec, expected_spec)
-        os.replace(alt_info, expected_info)
-        normalized.append(func_path)
 
     return normalized
 
diff --git a/src/spec_generation_and_verification.py b/src/spec_generation_and_verification.py
index a987e1d4..3e469662 100644
--- a/src/spec_generation_and_verification.py
+++ b/src/spec_generation_and_verification.py
@@ -11,7 +11,13 @@
 
 from config import MAX_WORKERS, OPENCODE_MAX_RETRIES, OPENCODE_SPEC_MODEL
 from src.domain_knowledge import list_staged_domain_knowledge_relpaths
-from src.file_utils import _get_incomplete_verification_files, _get_phase_files, is_file_ready, normalize_spec_filenames
+from src.file_utils import (
+    AmbiguousSidecarError,
+    _get_incomplete_verification_files,
+    _get_phase_files,
+    is_file_ready,
+    normalize_spec_filenames,
+)
 from src.generate_topdown_layers import generate_topdown_layers
 from src.llm_client import build_llm_cli_command
 from src.opencode_trace import function_id_from_extracted_path, run_opencode_traced
@@ -63,6 +69,7 @@ def _run_spec_generation_batch(
     layer_idx,
     batch_rel_dir,
     batch_info,
+    normalization_candidates=None,
 ):
     # Run one batch end-to-end so the executor can refill slots as soon as a
     # batch finishes, instead of waiting for a whole chunk barrier.
@@ -132,6 +139,14 @@ def _run_spec_generation_batch(
         return result.returncode
     except subprocess.CalledProcessError as exc:
         return exc.returncode
+    finally:
+        # Publish valid extension-dropped sidecars before this Future becomes
+        # done. The existing streaming_reasoner can then verify this batch while
+        # other spec-generation batches are still running.
+        normalize_spec_filenames(
+            [os.path.join(proj_dir, func_rel) for func_rel in function_files],
+            all_function_files=normalization_candidates,
+        )
 
 
 def run_spec_generation_and_verification(
@@ -203,6 +218,9 @@ def run_spec_generation_and_verification(
                     layer_files.append(rel)
 
             layer_processed = set()
+            layer_function_paths = [
+                os.path.join(input_dir, rel) for rel in layer_files
+            ]
 
             for attempt in range(1, OPENCODE_MAX_RETRIES + 1):
                 # Find batches with unspecced functions
@@ -230,6 +248,33 @@ def run_spec_generation_and_verification(
                             layer_processed.update(newly_processed)
                     break
 
+                # Any function entering spec generation has invalid or missing
+                # sidecars, so verification artifacts made from its previous
+                # sidecars are stale. Invalidate them before producers start so
+                # the running watcher cannot resume an old verdict when either
+                # canonical or normalized sidecars become ready.
+                pending_rel_paths = []
+                pending_rel_seen = set()
+                pending_abs_paths = set()
+                for batch_info in pending_batches:
+                    for func_rel in batch_info.get("functions", []):
+                        func_path = os.path.join(proj_dir, func_rel)
+                        if is_file_ready(func_path):
+                            continue
+                        rel = os.path.relpath(func_path, input_dir)
+                        if rel in pending_rel_seen:
+                            continue
+                        pending_rel_seen.add(rel)
+                        pending_rel_paths.append(rel)
+                        pending_abs_paths.add(func_path)
+
+                _invalidate_verification_artifacts(
+                    pending_rel_paths,
+                    output_dir,
+                    work_dir,
+                )
+                layer_processed.difference_update(pending_abs_paths)
+
                 # Submit all pending spec batches through a bounded executor so
                 # finished slots can immediately pick up the next batch.
                 spec_futures = []
@@ -254,6 +299,7 @@ def run_spec_generation_and_verification(
                                 layer_idx,
                                 batch_rel_dir,
                                 batch_info,
+                                layer_function_paths,
                             )
                         )
 
@@ -267,7 +313,7 @@ def run_spec_generation_and_verification(
                             input_dir, output_dir, file_list=layer_files,
                             proj_dir=proj_dir, work_dir=work_dir,
                             spec_procs=spec_futures,
-                            already_processed=all_processed | layer_processed,
+                            already_processed=(all_processed | layer_processed) - pending_abs_paths,
                             resume=resume,
                             bug_validator_path=bug_validator_path,
                         )
@@ -276,52 +322,11 @@ def run_spec_generation_and_verification(
                     for future in spec_futures:
                         try:
                             future.result()
+                        except AmbiguousSidecarError:
+                            raise
                         except Exception as exc:
                             logging.error(f"Spec generation task failed unexpectedly: {exc}")
 
-                normalized_paths = normalize_spec_filenames(
-                    [os.path.join(input_dir, rel) for rel in layer_files]
-                )
-
-                if normalized_paths:
-                    ready_to_verify = []
-                    normalized_abs = set()
-                    for func_path in normalized_paths:
-                        if not is_file_ready(func_path):
-                            logging.warning(
-                                "Normalized sidecars for %s are not ready; they will be retried.",
-                                func_path,
-                            )
-                            continue
-                        ready_to_verify.append(os.path.relpath(func_path, input_dir))
-                        normalized_abs.add(func_path)
-
-                    _invalidate_verification_artifacts(
-                        ready_to_verify,
-                        output_dir,
-                        work_dir,
-                    )
-                    layer_processed.difference_update(normalized_abs)
-
-                    if ready_to_verify and not only_spec:
-                        logging.info(
-                            "Forcing verification for %d normalized file(s).",
-                            len(ready_to_verify),
-                        )
-
-                        newly_processed = streaming_reasoner(
-                            input_dir,
-                            output_dir,
-                            file_list=ready_to_verify,
-                            proj_dir=proj_dir,
-                            work_dir=work_dir,
-                            spec_procs=None,
-                            already_processed=(all_processed | layer_processed) - normalized_abs,
-                            resume=False,
-                            bug_validator_path=bug_validator_path,
-                        )
-                        layer_processed.update(newly_processed)
-
                 # Check if any files in this layer received specs
                 specs_generated = sum(
                     1 for rel in layer_files
diff --git a/src/verification.py b/src/verification.py
index e99e0a05..cc2302b1 100644
--- a/src/verification.py
+++ b/src/verification.py
@@ -115,6 +115,7 @@ def streaming_reasoner(
             reasoning_futures = {}
             validation_futures = {}
             submitted = set()
+            producers_done_observed = False
 
             while True:
                 # Scan for new ready files
@@ -221,6 +222,14 @@ def streaming_reasoner(
                 # Detect if spec generation subprocesses exited before all files are ready
                 _all_procs = spec_procs if spec_procs else None
                 if _all_procs is not None and all(_spec_task_done(p) for p in _all_procs):
+                    if not producers_done_observed:
+                        # A producer can publish its normalized sidecars and
+                        # become done between the scan at the top of this loop
+                        # and this check. Rescan once before deciding that any
+                        # remaining files are genuinely unready.
+                        producers_done_observed = True
+                        continue
+
                     unready = (expected_files or set()) - processed
                     if unready and not reasoning_futures and not validation_futures:
                         exit_codes = [_spec_task_exit_code(p) for p in _all_procs]
diff --git a/tests/test_spec_sidecar_normalization.py b/tests/test_spec_sidecar_normalization.py
new file mode 100644
index 00000000..9a978c0d
--- /dev/null
+++ b/tests/test_spec_sidecar_normalization.py
@@ -0,0 +1,301 @@
+import concurrent.futures
+import json
+import os
+import tempfile
+import threading
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+import src.spec_generation_and_verification as stage
+import src.verification as verification
+from src.file_utils import is_file_ready, normalize_spec_filenames
+
+
+VALID_INFO = {"callees": []}
+OLD_SPEC = {
+    "signature": "old()",
+    "pre_condition": "true",
+    "post_condition": "old result",
+}
+NEW_SPEC = {
+    "signature": "new()",
+    "pre_condition": "true",
+    "post_condition": "new result",
+}
+
+
+def _write_json(path, value):
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(value), encoding="utf-8")
+
+
+class NormalizeSpecFilenamesTest(unittest.TestCase):
+    def test_complete_pair_is_published_without_ready_mixed_pair(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            function = Path(tmp) / "B.rs"
+            function.write_text("fn b() {}\n", encoding="utf-8")
+
+            expected_info = Path(f"{function}.info.json")
+            _write_json(expected_info, VALID_INFO)
+            _write_json(function.with_suffix(".spec.json"), NEW_SPEC)
+            _write_json(function.with_suffix(".info.json"), VALID_INFO)
+
+            readiness_after_each_replace = []
+            real_replace = os.replace
+
+            def observing_replace(source, destination):
+                real_replace(source, destination)
+                readiness_after_each_replace.append(is_file_ready(str(function)))
+
+            with mock.patch(
+                "src.file_utils.os.replace",
+                side_effect=observing_replace,
+            ):
+                normalized = normalize_spec_filenames([str(function)])
+
+            self.assertEqual(normalized, [str(function)])
+            self.assertEqual(readiness_after_each_replace, [False, True])
+            self.assertEqual(
+                json.loads(Path(f"{function}.spec.json").read_text()),
+                NEW_SPEC,
+            )
+
+    def test_incomplete_pair_is_left_for_retry(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            function = Path(tmp) / "B.rs"
+            function.write_text("fn b() {}\n", encoding="utf-8")
+            _write_json(function.with_suffix(".spec.json"), NEW_SPEC)
+
+            normalized = normalize_spec_filenames([str(function)])
+
+            self.assertEqual(normalized, [])
+            self.assertFalse(is_file_ready(str(function)))
+            self.assertTrue(function.with_suffix(".spec.json").exists())
+
+    def test_batch_scope_still_detects_layer_wide_ambiguity(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            rust_function = Path(tmp) / "same.rs"
+            python_function = Path(tmp) / "same.py"
+            rust_function.write_text("fn same() {}\n", encoding="utf-8")
+            python_function.write_text("def same(): pass\n", encoding="utf-8")
+            _write_json(rust_function.with_suffix(".spec.json"), NEW_SPEC)
+            _write_json(rust_function.with_suffix(".info.json"), VALID_INFO)
+
+            with self.assertRaises(RuntimeError):
+                normalize_spec_filenames(
+                    [str(rust_function)],
+                    all_function_files=[
+                        str(rust_function),
+                        str(python_function),
+                    ],
+                )
+
+
+class StreamingNormalizationTest(unittest.TestCase):
+    def _pipeline_layout(self, root, function_names):
+        proj_dir = Path(root)
+        work_dir = proj_dir / "fm_agent"
+        input_dir = work_dir / "extracted_functions"
+        output_dir = work_dir / "logic_verification_results"
+        spec_prompts_dir = work_dir / "spec_prompts"
+        script_dir = proj_dir / "script"
+
+        (script_dir / "md").mkdir(parents=True)
+        (script_dir / "md" / "workflow_spec_step4_batch.md").write_text(
+            "prompt",
+            encoding="utf-8",
+        )
+
+        extracted_dir = input_dir / "foo-rs"
+        extracted_dir.mkdir(parents=True)
+        functions = {}
+        batches = []
+        for index, name in enumerate(function_names):
+            function = extracted_dir / name
+            function.write_text(f"fn {function.stem.lower()}() {{}}\n", encoding="utf-8")
+            functions[name] = function
+            batch_file = f"batch_{index}.txt"
+            batches.append({
+                "file": batch_file,
+                "functions": [str(function.relative_to(proj_dir))],
+                "num_pending": 1,
+            })
+
+        _write_json(
+            spec_prompts_dir / "phase_01_topdown_layers.json",
+            {"total_layers": 1},
+        )
+        batch_dir = spec_prompts_dir / "batch_prompts_demo_phase01"
+        _write_json(batch_dir / "manifest.json", {"batches": batches})
+        for batch in batches:
+            (batch_dir / batch["file"]).write_text("batch", encoding="utf-8")
+
+        phases = {
+            "project": "demo",
+            "phases": [{
+                "phase": 1,
+                "name": "phase",
+                "modules": [{"source_files": ["foo.rs"]}],
+            }],
+        }
+        return {
+            "proj_dir": proj_dir,
+            "work_dir": work_dir,
+            "input_dir": input_dir,
+            "output_dir": output_dir,
+            "spec_prompts_dir": spec_prompts_dir,
+            "script_dir": script_dir,
+            "functions": functions,
+            "phases": phases,
+        }
+
+    def test_completed_batch_is_verified_while_another_batch_is_running(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            layout = self._pipeline_layout(tmp, ["A.rs", "B.rs"])
+            function_a = layout["functions"]["A.rs"]
+            function_b = layout["functions"]["B.rs"]
+
+            # A stale result must be gone before the producer publishes a new
+            # canonical or normalized pair, even under --resume.
+            _write_json(Path(f"{function_a}.spec.json"), OLD_SPEC)
+            stale_result = layout["output_dir"] / "foo-rs" / "A.json"
+            stale_validation = (
+                layout["work_dir"]
+                / "bug_validation"
+                / "foo-rs--A.result.json"
+            )
+            stale_summary = layout["work_dir"] / "bug_validation" / "summary.json"
+            _write_json(stale_result, {"verdict": "MATCH"})
+            _write_json(stale_validation, {"confirmation_status": "confirmed"})
+            _write_json(stale_summary, {"total_confirmed": 1})
+
+            a_verified = threading.Event()
+            ordering_errors = []
+            verify_calls = []
+            streaming_calls = []
+
+            def fake_generation(**kwargs):
+                batch_file = kwargs["metadata"]["batch_file"]
+                function = function_a if batch_file == "batch_0.txt" else function_b
+                if function == function_b and not a_verified.wait(timeout=3):
+                    ordering_errors.append(
+                        "Batch A was not verified before Batch B was allowed to finish"
+                    )
+                if function == function_a:
+                    if stale_result.exists() or stale_validation.exists() or stale_summary.exists():
+                        ordering_errors.append("Stale resume artifacts were not invalidated")
+                _write_json(function.with_suffix(".spec.json"), NEW_SPEC)
+                _write_json(function.with_suffix(".info.json"), VALID_INFO)
+                return SimpleNamespace(returncode=0)
+
+            def fake_verify(
+                file_path,
+                input_dir,
+                output_dir,
+                language,
+                work_dir=None,
+                resume=False,
+            ):
+                verify_calls.append((Path(file_path).name, resume))
+                if Path(file_path).name == "A.rs":
+                    a_verified.set()
+                return file_path, "MATCH"
+
+            real_streaming_reasoner = verification.streaming_reasoner
+
+            def fast_streaming_reasoner(*args, **kwargs):
+                streaming_calls.append(kwargs.get("spec_procs"))
+                kwargs["poll_interval"] = 0.01
+                return real_streaming_reasoner(*args, **kwargs)
+
+            with mock.patch.object(stage.subprocess, "run"), \
+                    mock.patch.object(stage, "build_llm_cli_command", return_value=["fake"]), \
+                    mock.patch.object(stage, "run_opencode_traced", side_effect=fake_generation), \
+                    mock.patch.object(stage, "list_staged_domain_knowledge_relpaths", return_value=[]), \
+                    mock.patch.object(stage, "streaming_reasoner", side_effect=fast_streaming_reasoner), \
+                    mock.patch.object(stage, "MAX_WORKERS", 2), \
+                    mock.patch.object(stage, "OPENCODE_MAX_RETRIES", 1), \
+                    mock.patch.object(verification, "MAX_WORKERS", 2), \
+                    mock.patch.object(verification, "_verify_single_file", side_effect=fake_verify):
+                stage.run_spec_generation_and_verification(
+                    str(layout["proj_dir"]),
+                    str(layout["work_dir"]),
+                    str(layout["input_dir"]),
+                    str(layout["output_dir"]),
+                    str(layout["script_dir"]),
+                    str(layout["spec_prompts_dir"]),
+                    layout["phases"],
+                    resume=True,
+                )
+
+            self.assertEqual(ordering_errors, [])
+            self.assertEqual(len(streaming_calls), 1)
+            self.assertTrue(streaming_calls[0])
+            self.assertEqual({name for name, _ in verify_calls}, {"A.rs", "B.rs"})
+            self.assertTrue(all(resume for _, resume in verify_calls))
+            self.assertTrue(is_file_ready(str(function_a)))
+            self.assertTrue(is_file_ready(str(function_b)))
+
+    def test_watcher_rescans_when_producer_finishes_after_scan(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            input_dir = root / "input"
+            output_dir = root / "output"
+            input_dir.mkdir()
+            function = input_dir / "B.rs"
+            function.write_text("fn b() {}\n", encoding="utf-8")
+
+            first_scan_finished = threading.Event()
+            pair_published = threading.Event()
+
+            def producer():
+                if not first_scan_finished.wait(timeout=3):
+                    raise RuntimeError("watcher did not complete its first scan")
+                _write_json(Path(f"{function}.spec.json"), NEW_SPEC)
+                _write_json(Path(f"{function}.info.json"), VALID_INFO)
+                pair_published.set()
+                return 0
+
+            real_walk = os.walk
+            first_walk = True
+            future_holder = {}
+
+            def controlled_walk(path):
+                nonlocal first_walk
+                yield from real_walk(path)
+                if first_walk and Path(path) == input_dir:
+                    first_walk = False
+                    first_scan_finished.set()
+                    if not pair_published.wait(timeout=3):
+                        raise RuntimeError("producer did not publish its pair")
+                    deadline = time.monotonic() + 3
+                    while not future_holder["future"].done():
+                        if time.monotonic() >= deadline:
+                            raise RuntimeError("producer Future did not become done")
+                        time.sleep(0.001)
+
+            def fake_verify(file_path, *args, **kwargs):
+                return file_path, "MATCH"
+
+            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
+                producer_future = executor.submit(producer)
+                future_holder["future"] = producer_future
+                with mock.patch.object(verification.os, "walk", side_effect=controlled_walk), \
+                        mock.patch.object(verification, "_verify_single_file", side_effect=fake_verify):
+                    processed = verification.streaming_reasoner(
+                        str(input_dir),
+                        str(output_dir),
+                        file_list=["B.rs"],
+                        spec_procs=[producer_future],
+                        poll_interval=0.001,
+                    )
+
+            self.assertIn(str(function), processed)
+
+
+if __name__ == "__main__":
+    unittest.main()

From a46004a26f1d3ce4fa5038001f684ca53984d24c Mon Sep 17 00:00:00 2001
From: huangwei021230 <huangwei021230@users.noreply.github.com>
Date: Mon, 3 Aug 2026 12:39:39 +0800
Subject: [PATCH 4/7] Remove standalone normalization tests

---
 tests/test_spec_sidecar_normalization.py | 301 -----------------------
 1 file changed, 301 deletions(-)
 delete mode 100644 tests/test_spec_sidecar_normalization.py

diff --git a/tests/test_spec_sidecar_normalization.py b/tests/test_spec_sidecar_normalization.py
deleted file mode 100644
index 9a978c0d..00000000
--- a/tests/test_spec_sidecar_normalization.py
+++ /dev/null
@@ -1,301 +0,0 @@
-import concurrent.futures
-import json
-import os
-import tempfile
-import threading
-import time
-import unittest
-from pathlib import Path
-from types import SimpleNamespace
-from unittest import mock
-
-import src.spec_generation_and_verification as stage
-import src.verification as verification
-from src.file_utils import is_file_ready, normalize_spec_filenames
-
-
-VALID_INFO = {"callees": []}
-OLD_SPEC = {
-    "signature": "old()",
-    "pre_condition": "true",
-    "post_condition": "old result",
-}
-NEW_SPEC = {
-    "signature": "new()",
-    "pre_condition": "true",
-    "post_condition": "new result",
-}
-
-
-def _write_json(path, value):
-    path = Path(path)
-    path.parent.mkdir(parents=True, exist_ok=True)
-    path.write_text(json.dumps(value), encoding="utf-8")
-
-
-class NormalizeSpecFilenamesTest(unittest.TestCase):
-    def test_complete_pair_is_published_without_ready_mixed_pair(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            function = Path(tmp) / "B.rs"
-            function.write_text("fn b() {}\n", encoding="utf-8")
-
-            expected_info = Path(f"{function}.info.json")
-            _write_json(expected_info, VALID_INFO)
-            _write_json(function.with_suffix(".spec.json"), NEW_SPEC)
-            _write_json(function.with_suffix(".info.json"), VALID_INFO)
-
-            readiness_after_each_replace = []
-            real_replace = os.replace
-
-            def observing_replace(source, destination):
-                real_replace(source, destination)
-                readiness_after_each_replace.append(is_file_ready(str(function)))
-
-            with mock.patch(
-                "src.file_utils.os.replace",
-                side_effect=observing_replace,
-            ):
-                normalized = normalize_spec_filenames([str(function)])
-
-            self.assertEqual(normalized, [str(function)])
-            self.assertEqual(readiness_after_each_replace, [False, True])
-            self.assertEqual(
-                json.loads(Path(f"{function}.spec.json").read_text()),
-                NEW_SPEC,
-            )
-
-    def test_incomplete_pair_is_left_for_retry(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            function = Path(tmp) / "B.rs"
-            function.write_text("fn b() {}\n", encoding="utf-8")
-            _write_json(function.with_suffix(".spec.json"), NEW_SPEC)
-
-            normalized = normalize_spec_filenames([str(function)])
-
-            self.assertEqual(normalized, [])
-            self.assertFalse(is_file_ready(str(function)))
-            self.assertTrue(function.with_suffix(".spec.json").exists())
-
-    def test_batch_scope_still_detects_layer_wide_ambiguity(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            rust_function = Path(tmp) / "same.rs"
-            python_function = Path(tmp) / "same.py"
-            rust_function.write_text("fn same() {}\n", encoding="utf-8")
-            python_function.write_text("def same(): pass\n", encoding="utf-8")
-            _write_json(rust_function.with_suffix(".spec.json"), NEW_SPEC)
-            _write_json(rust_function.with_suffix(".info.json"), VALID_INFO)
-
-            with self.assertRaises(RuntimeError):
-                normalize_spec_filenames(
-                    [str(rust_function)],
-                    all_function_files=[
-                        str(rust_function),
-                        str(python_function),
-                    ],
-                )
-
-
-class StreamingNormalizationTest(unittest.TestCase):
-    def _pipeline_layout(self, root, function_names):
-        proj_dir = Path(root)
-        work_dir = proj_dir / "fm_agent"
-        input_dir = work_dir / "extracted_functions"
-        output_dir = work_dir / "logic_verification_results"
-        spec_prompts_dir = work_dir / "spec_prompts"
-        script_dir = proj_dir / "script"
-
-        (script_dir / "md").mkdir(parents=True)
-        (script_dir / "md" / "workflow_spec_step4_batch.md").write_text(
-            "prompt",
-            encoding="utf-8",
-        )
-
-        extracted_dir = input_dir / "foo-rs"
-        extracted_dir.mkdir(parents=True)
-        functions = {}
-        batches = []
-        for index, name in enumerate(function_names):
-            function = extracted_dir / name
-            function.write_text(f"fn {function.stem.lower()}() {{}}\n", encoding="utf-8")
-            functions[name] = function
-            batch_file = f"batch_{index}.txt"
-            batches.append({
-                "file": batch_file,
-                "functions": [str(function.relative_to(proj_dir))],
-                "num_pending": 1,
-            })
-
-        _write_json(
-            spec_prompts_dir / "phase_01_topdown_layers.json",
-            {"total_layers": 1},
-        )
-        batch_dir = spec_prompts_dir / "batch_prompts_demo_phase01"
-        _write_json(batch_dir / "manifest.json", {"batches": batches})
-        for batch in batches:
-            (batch_dir / batch["file"]).write_text("batch", encoding="utf-8")
-
-        phases = {
-            "project": "demo",
-            "phases": [{
-                "phase": 1,
-                "name": "phase",
-                "modules": [{"source_files": ["foo.rs"]}],
-            }],
-        }
-        return {
-            "proj_dir": proj_dir,
-            "work_dir": work_dir,
-            "input_dir": input_dir,
-            "output_dir": output_dir,
-            "spec_prompts_dir": spec_prompts_dir,
-            "script_dir": script_dir,
-            "functions": functions,
-            "phases": phases,
-        }
-
-    def test_completed_batch_is_verified_while_another_batch_is_running(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            layout = self._pipeline_layout(tmp, ["A.rs", "B.rs"])
-            function_a = layout["functions"]["A.rs"]
-            function_b = layout["functions"]["B.rs"]
-
-            # A stale result must be gone before the producer publishes a new
-            # canonical or normalized pair, even under --resume.
-            _write_json(Path(f"{function_a}.spec.json"), OLD_SPEC)
-            stale_result = layout["output_dir"] / "foo-rs" / "A.json"
-            stale_validation = (
-                layout["work_dir"]
-                / "bug_validation"
-                / "foo-rs--A.result.json"
-            )
-            stale_summary = layout["work_dir"] / "bug_validation" / "summary.json"
-            _write_json(stale_result, {"verdict": "MATCH"})
-            _write_json(stale_validation, {"confirmation_status": "confirmed"})
-            _write_json(stale_summary, {"total_confirmed": 1})
-
-            a_verified = threading.Event()
-            ordering_errors = []
-            verify_calls = []
-            streaming_calls = []
-
-            def fake_generation(**kwargs):
-                batch_file = kwargs["metadata"]["batch_file"]
-                function = function_a if batch_file == "batch_0.txt" else function_b
-                if function == function_b and not a_verified.wait(timeout=3):
-                    ordering_errors.append(
-                        "Batch A was not verified before Batch B was allowed to finish"
-                    )
-                if function == function_a:
-                    if stale_result.exists() or stale_validation.exists() or stale_summary.exists():
-                        ordering_errors.append("Stale resume artifacts were not invalidated")
-                _write_json(function.with_suffix(".spec.json"), NEW_SPEC)
-                _write_json(function.with_suffix(".info.json"), VALID_INFO)
-                return SimpleNamespace(returncode=0)
-
-            def fake_verify(
-                file_path,
-                input_dir,
-                output_dir,
-                language,
-                work_dir=None,
-                resume=False,
-            ):
-                verify_calls.append((Path(file_path).name, resume))
-                if Path(file_path).name == "A.rs":
-                    a_verified.set()
-                return file_path, "MATCH"
-
-            real_streaming_reasoner = verification.streaming_reasoner
-
-            def fast_streaming_reasoner(*args, **kwargs):
-                streaming_calls.append(kwargs.get("spec_procs"))
-                kwargs["poll_interval"] = 0.01
-                return real_streaming_reasoner(*args, **kwargs)
-
-            with mock.patch.object(stage.subprocess, "run"), \
-                    mock.patch.object(stage, "build_llm_cli_command", return_value=["fake"]), \
-                    mock.patch.object(stage, "run_opencode_traced", side_effect=fake_generation), \
-                    mock.patch.object(stage, "list_staged_domain_knowledge_relpaths", return_value=[]), \
-                    mock.patch.object(stage, "streaming_reasoner", side_effect=fast_streaming_reasoner), \
-                    mock.patch.object(stage, "MAX_WORKERS", 2), \
-                    mock.patch.object(stage, "OPENCODE_MAX_RETRIES", 1), \
-                    mock.patch.object(verification, "MAX_WORKERS", 2), \
-                    mock.patch.object(verification, "_verify_single_file", side_effect=fake_verify):
-                stage.run_spec_generation_and_verification(
-                    str(layout["proj_dir"]),
-                    str(layout["work_dir"]),
-                    str(layout["input_dir"]),
-                    str(layout["output_dir"]),
-                    str(layout["script_dir"]),
-                    str(layout["spec_prompts_dir"]),
-                    layout["phases"],
-                    resume=True,
-                )
-
-            self.assertEqual(ordering_errors, [])
-            self.assertEqual(len(streaming_calls), 1)
-            self.assertTrue(streaming_calls[0])
-            self.assertEqual({name for name, _ in verify_calls}, {"A.rs", "B.rs"})
-            self.assertTrue(all(resume for _, resume in verify_calls))
-            self.assertTrue(is_file_ready(str(function_a)))
-            self.assertTrue(is_file_ready(str(function_b)))
-
-    def test_watcher_rescans_when_producer_finishes_after_scan(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            root = Path(tmp)
-            input_dir = root / "input"
-            output_dir = root / "output"
-            input_dir.mkdir()
-            function = input_dir / "B.rs"
-            function.write_text("fn b() {}\n", encoding="utf-8")
-
-            first_scan_finished = threading.Event()
-            pair_published = threading.Event()
-
-            def producer():
-                if not first_scan_finished.wait(timeout=3):
-                    raise RuntimeError("watcher did not complete its first scan")
-                _write_json(Path(f"{function}.spec.json"), NEW_SPEC)
-                _write_json(Path(f"{function}.info.json"), VALID_INFO)
-                pair_published.set()
-                return 0
-
-            real_walk = os.walk
-            first_walk = True
-            future_holder = {}
-
-            def controlled_walk(path):
-                nonlocal first_walk
-                yield from real_walk(path)
-                if first_walk and Path(path) == input_dir:
-                    first_walk = False
-                    first_scan_finished.set()
-                    if not pair_published.wait(timeout=3):
-                        raise RuntimeError("producer did not publish its pair")
-                    deadline = time.monotonic() + 3
-                    while not future_holder["future"].done():
-                        if time.monotonic() >= deadline:
-                            raise RuntimeError("producer Future did not become done")
-                        time.sleep(0.001)
-
-            def fake_verify(file_path, *args, **kwargs):
-                return file_path, "MATCH"
-
-            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
-                producer_future = executor.submit(producer)
-                future_holder["future"] = producer_future
-                with mock.patch.object(verification.os, "walk", side_effect=controlled_walk), \
-                        mock.patch.object(verification, "_verify_single_file", side_effect=fake_verify):
-                    processed = verification.streaming_reasoner(
-                        str(input_dir),
-                        str(output_dir),
-                        file_list=["B.rs"],
-                        spec_procs=[producer_future],
-                        poll_interval=0.001,
-                    )
-
-            self.assertIn(str(function), processed)
-
-
-if __name__ == "__main__":
-    unittest.main()

From 7f054871990dad92efea36a6406496b97521f93b Mon Sep 17 00:00:00 2001
From: huangwei021230 <huangwei021230@users.noreply.github.com>
Date: Mon, 3 Aug 2026 12:49:49 +0800
Subject: [PATCH 5/7] Clear stale alternate sidecars before retries

---
 src/spec_generation_and_verification.py | 22 ++++++++++++++++++++++
 1 file changed, 22 insertions(+)

diff --git a/src/spec_generation_and_verification.py b/src/spec_generation_and_verification.py
index d91e6add..68255133 100644
--- a/src/spec_generation_and_verification.py
+++ b/src/spec_generation_and_verification.py
@@ -62,6 +62,27 @@ def _invalidate_verification_artifacts(file_list, output_dir, work_dir):
             pass
 
 
+def _clear_stale_alternate_sidecars(function_paths):
+    """Remove bare sidecars before starting a new generation attempt.
+
+    A valid bare spec and info file can otherwise come from different retry
+    attempts. Clearing both before the producer starts makes any pair seen by
+    the batch finalizer belong to the current attempt.
+    """
+    alternate_paths = set()
+    for function_path in function_paths:
+        base = os.path.splitext(function_path)[0]
+        alternate_paths.add(f"{base}.spec.json")
+        alternate_paths.add(f"{base}.info.json")
+
+    for alternate_path in sorted(alternate_paths):
+        try:
+            os.remove(alternate_path)
+            logging.info("Removed stale alternate sidecar: %s", alternate_path)
+        except FileNotFoundError:
+            pass
+
+
 def _run_spec_generation_batch(
     proj_dir,
     work_dir,
@@ -277,6 +298,7 @@ def run_spec_generation_and_verification(
                         pending_rel_paths.append(rel)
                         pending_abs_paths.add(func_path)
 
+                _clear_stale_alternate_sidecars(pending_abs_paths)
                 _invalidate_verification_artifacts(
                     pending_rel_paths,
                     output_dir,

From ae31cf3161f175d899b65fdd2cc4f0fb23cc98c4 Mon Sep 17 00:00:00 2001
From: Wei Huang <91535080+huangwei021230@users.noreply.github.com>
Date: Wed, 5 Aug 2026 14:14:51 +0800
Subject: [PATCH 6/7] Retry ambiguous sidecar generation instead of crashing

---
 src/file_utils.py                       | 56 +++++++++++++++----------
 src/spec_generation_and_verification.py |  3 --
 2 files changed, 33 insertions(+), 26 deletions(-)

diff --git a/src/file_utils.py b/src/file_utils.py
index 8f79b004..8a41e9de 100644
--- a/src/file_utils.py
+++ b/src/file_utils.py
@@ -1,5 +1,4 @@
 import logging
-import os
 import json
 import os
 import re
@@ -43,10 +42,6 @@
 }
 
 
-class AmbiguousSidecarError(RuntimeError):
-    """Raised when an extension-dropped sidecar has multiple owners."""
-
-
 def _is_metadata_sidecar(file_path):
     """Return whether file_path is a function metadata sidecar."""
     return str(file_path).endswith(_METADATA_SIDECAR_SUFFIXES)
@@ -150,11 +145,9 @@ def normalize_spec_filenames(function_files, all_function_files=None):
     When the mapping is unambiguous and both bare sidecars form a valid
     pair, this function replaces both expected sidecars together. Treating
     the pair as one unit avoids combining a sidecar from an earlier attempt
-    with one from the latest attempt.
-
-    Raises:
-        AmbiguousSidecarError: If a bare sidecar could belong to multiple
-            source files.
+    with one from the latest attempt. If a bare name has multiple possible
+    owners, both bare sidecars are removed and the canonical pairs remain
+    unready for the caller's bounded retry loop.
 
     Args:
         function_files: Function paths whose sidecars may be normalized.
@@ -182,22 +175,39 @@ def normalize_spec_filenames(function_files, all_function_files=None):
         base = os.path.splitext(func_path)[0]
         base_map.setdefault(base, []).append(func_path)
 
-    for func_path in function_files:
-        base = os.path.splitext(func_path)[0]
-        candidates = base_map[base]
-
-        if len(candidates) > 1:
-            for suffix in (".spec.json", ".info.json"):
-                alt = f"{base}{suffix}"
-                expected = f"{func_path}{suffix}"
-                if os.path.isfile(alt) and not _is_valid_sidecar_file(expected):
-                    raise AmbiguousSidecarError(
-                        f"Ambiguous sidecar filename: {alt} could belong to "
-                        f"multiple source files: {candidates}"
-                    )
+    ambiguous_bases = {
+        base for base, candidates in base_map.items() if len(candidates) > 1
+    }
+    for base in sorted({os.path.splitext(path)[0] for path in function_files}):
+        if base not in ambiguous_bases:
+            continue
+
+        removed = []
+        for suffix in (".spec.json", ".info.json"):
+            alternate_path = f"{base}{suffix}"
+            try:
+                os.remove(alternate_path)
+                removed.append(alternate_path)
+            except FileNotFoundError:
+                pass
+
+        if removed:
+            logging.warning(
+                "Removed ambiguous extension-dropped sidecar file(s) %s; "
+                "they could belong to multiple source files: %s. Canonical "
+                "sidecars remain pending for retry.",
+                removed,
+                base_map[base],
+            )
 
     for func_path in function_files:
         base = os.path.splitext(func_path)[0]
+        if base in ambiguous_bases:
+            # A bare sidecar can never be safely assigned to one of these
+            # functions. Leave every incomplete canonical pair unready so the
+            # existing bounded retry loop asks the producer to regenerate it.
+            continue
+
         expected_spec = f"{func_path}.spec.json"
         expected_info = f"{func_path}.info.json"
 
diff --git a/src/spec_generation_and_verification.py b/src/spec_generation_and_verification.py
index 68255133..9eb722dd 100644
--- a/src/spec_generation_and_verification.py
+++ b/src/spec_generation_and_verification.py
@@ -13,7 +13,6 @@
 from config import MAX_WORKERS, OPENCODE_MAX_RETRIES, OPENCODE_SPEC_MODEL
 from src.domain_knowledge import list_staged_domain_knowledge_relpaths
 from src.file_utils import (
-    AmbiguousSidecarError,
     _get_incomplete_verification_files,
     _get_phase_files,
     is_file_ready,
@@ -354,8 +353,6 @@ def run_spec_generation_and_verification(
                     for future in spec_futures:
                         try:
                             future.result()
-                        except AmbiguousSidecarError:
-                            raise
                         except Exception as exc:
                             logging.error(f"Spec generation task failed unexpectedly: {exc}")
 

From 8dfa037ad7346a52b93ea6f4c3345cb98362e752 Mon Sep 17 00:00:00 2001
From: Wei Huang <91535080+huangwei021230@users.noreply.github.com>
Date: Wed, 5 Aug 2026 14:44:11 +0800
Subject: [PATCH 7/7] Prevent mixed canonical sidecar reads

---
 src/file_utils.py                       | 48 +++++++++++++++++--------
 src/spec_generation_and_verification.py | 26 ++++++++------
 2 files changed, 49 insertions(+), 25 deletions(-)

diff --git a/src/file_utils.py b/src/file_utils.py
index 8a41e9de..11de7986 100644
--- a/src/file_utils.py
+++ b/src/file_utils.py
@@ -147,7 +147,10 @@ def normalize_spec_filenames(function_files, all_function_files=None):
     the pair as one unit avoids combining a sidecar from an earlier attempt
     with one from the latest attempt. If a bare name has multiple possible
     owners, both bare sidecars are removed and the canonical pairs remain
-    unready for the caller's bounded retry loop.
+    unready for the caller's bounded retry loop. An incomplete canonical pair
+    is never completed from bare sidecars in the same watcher invocation: all
+    four files are cleared so a reader that already loaded the old canonical
+    half cannot combine it with a newly published half.
 
     Args:
         function_files: Function paths whose sidecars may be normalized.
@@ -211,6 +214,8 @@ def normalize_spec_filenames(function_files, all_function_files=None):
         expected_spec = f"{func_path}.spec.json"
         expected_info = f"{func_path}.info.json"
 
+        expected_spec_exists = os.path.isfile(expected_spec)
+        expected_info_exists = os.path.isfile(expected_info)
         expected_spec_valid = _is_valid_sidecar_file(expected_spec)
         expected_info_valid = _is_valid_sidecar_file(expected_info)
         if expected_spec_valid and expected_info_valid:
@@ -228,20 +233,35 @@ def normalize_spec_filenames(function_files, all_function_files=None):
                 )
             continue
 
-        # Keep at least one canonical sidecar invalid until the final replace.
-        # streaming_reasoner may be scanning concurrently with this batch
-        # finalizer, so it must never observe a ready old/new mixed pair.
-        if not expected_spec_valid:
-            replacements = (
-                (alt_info, expected_info),
-                (alt_spec, expected_spec),
-            )
-        else:
-            replacements = (
-                (alt_spec, expected_spec),
-                (alt_info, expected_info),
+        if expected_spec_exists or expected_info_exists:
+            # A concurrent readiness check may already have loaded the valid
+            # half of this incomplete canonical pair. Publishing the missing
+            # half now could make that reader accept an old/new pair. Remove
+            # both canonical and bare pairs, then let the next attempt generate
+            # a complete pair from a clean state before its watcher starts.
+            removed = []
+            for path in (expected_spec, expected_info, alt_spec, alt_info):
+                try:
+                    os.remove(path)
+                    removed.append(path)
+                except FileNotFoundError:
+                    pass
+            logging.warning(
+                "Removed conflicting canonical and extension-dropped sidecars "
+                "%s for %s; the complete pair remains pending for retry.",
+                removed,
+                func_path,
             )
-        for alt, expected in replacements:
+            continue
+
+        # The retry coordinator removes incomplete canonical pairs before it
+        # starts producers and the streaming watcher. Publish info first and
+        # spec last so the canonical spec path acts as the readiness gate: once
+        # it becomes readable, both sidecars come from this attempt.
+        for alt, expected in (
+            (alt_info, expected_info),
+            (alt_spec, expected_spec),
+        ):
             os.replace(alt, expected)
 
         normalized.append(func_path)
diff --git a/src/spec_generation_and_verification.py b/src/spec_generation_and_verification.py
index 9eb722dd..f68043af 100644
--- a/src/spec_generation_and_verification.py
+++ b/src/spec_generation_and_verification.py
@@ -61,23 +61,27 @@ def _invalidate_verification_artifacts(file_list, output_dir, work_dir):
             pass
 
 
-def _clear_stale_alternate_sidecars(function_paths):
-    """Remove bare sidecars before starting a new generation attempt.
+def _clear_stale_sidecars_before_attempt(function_paths):
+    """Remove incomplete canonical and bare sidecars before an attempt.
 
     A valid bare spec and info file can otherwise come from different retry
-    attempts. Clearing both before the producer starts makes any pair seen by
-    the batch finalizer belong to the current attempt.
+    attempts. A valid canonical half can likewise be read by streaming_reasoner
+    while the producer publishes its replacement half. This function runs
+    before both producers and their watcher start, so clearing all four paths
+    makes any subsequently ready pair belong to the new attempt.
     """
-    alternate_paths = set()
+    stale_paths = set()
     for function_path in function_paths:
         base = os.path.splitext(function_path)[0]
-        alternate_paths.add(f"{base}.spec.json")
-        alternate_paths.add(f"{base}.info.json")
+        stale_paths.add(f"{function_path}.spec.json")
+        stale_paths.add(f"{function_path}.info.json")
+        stale_paths.add(f"{base}.spec.json")
+        stale_paths.add(f"{base}.info.json")
 
-    for alternate_path in sorted(alternate_paths):
+    for stale_path in sorted(stale_paths):
         try:
-            os.remove(alternate_path)
-            logging.info("Removed stale alternate sidecar: %s", alternate_path)
+            os.remove(stale_path)
+            logging.info("Removed stale sidecar before generation attempt: %s", stale_path)
         except FileNotFoundError:
             pass
 
@@ -297,7 +301,7 @@ def run_spec_generation_and_verification(
                         pending_rel_paths.append(rel)
                         pending_abs_paths.add(func_path)
 
-                _clear_stale_alternate_sidecars(pending_abs_paths)
+                _clear_stale_sidecars_before_attempt(pending_abs_paths)
                 _invalidate_verification_artifacts(
                     pending_rel_paths,
                     output_dir,