Merge pull request #552 from escoffier-labs/fix/verify-miseledger-auto-import

fix(evidence): index captured verify receipts

Author: solomonneas

docs/technical-guide.md

@@ -1041,7 +1041,9 @@ For scheduled local indexing, use the built-in pipeline:
 brigade receipts export miseledger --target . --new-only --import
 ```
 
-Put that export on your own timer when you want the loop to run without thinking about it. Brigade does not install the timer. The export sends new verify-run and Brigade-run receipts to MiseLedger, and the next `brigade run` can fetch them back as a compact evidence brief through the same read-only `miseledger evidence ... --source brigade --json` path. The direct command is:
+`brigade work verify run --capture <skill>` also runs that new-only export/import automatically after outcome capture, so routine verification closes the receipts-to-MiseLedger loop without a separate command. Indexing is fail-open: a missing `miseledger` binary or import failure prints an explicit status, leaves pending receipts retryable, and does not change the verification exit code. Use the manual export command above for backlog catch-up or fleet targets (`.brigade/repos.toml` with `--fleet --json`).
+
+The export sends new verify-run and Brigade-run receipts to MiseLedger, and the next `brigade run` can fetch them back as a compact evidence brief through the same read-only `miseledger evidence ... --source brigade --json` path. The direct command is:
 
 ```bash
 brigade work import context --from-miseledger "auth receipts" --target .
@@ -1051,7 +1053,7 @@ The brief contains only bounded evidence lines: run id, status, code-graph delta
 
 MiseLedger evidence is fail-open. Missing `miseledger`, a nonzero exit, timeout, malformed JSON, or zero usable results does not block `brigade run` or `brigade work import context --from-miseledger`. Run prompts proceed without the brief, and the import command reports the absence instead of writing a broken context note.
 
-`--new-only` stores exported `raw.hash` values in `.brigade/work/miseledger-export-cursor.json`, so later runs only write receipt items that have not already been exported. The cursor is an optimization, not the identity boundary. MiseLedger uses content-hash identity for adapter items, so double-importing the same receipt file remains harmless if the cursor is deleted, copied late, or skipped.
+`--new-only` stores exported `raw.hash` values in `.brigade/work/miseledger-export-cursor.json`, so later runs only write receipt items that have not already been exported. With `--import`, the cursor advances only after a successful import; failed imports leave receipts pending for the next run. Export-only runs advance the cursor after a successful write.
 
 The export is deterministic and idempotent for unchanged receipts: records are sorted newest first, external ids derive from receipt ids, hashes derive from stored receipt digests or deterministic fallbacks, and JSONL rendering uses stable key order. When a receipt carries `digests.signature` and `digests.key_id`, the export includes them at `item.metadata.digest_signature`; unsigned receipts omit that metadata field. Re-importing the same file should update or skip the same MiseLedger adapter items rather than create duplicates.
 

scripts/windows-native-acceptance.ps1

@@ -691,6 +691,10 @@ def call_greet():
         & brigade work verify run --target $workRepo --command "python $verifyScriptName" --capture brigade-work
         if ($LASTEXITCODE -ne 0) { throw "work verify run failed" }
 
+        Write-Step "brigade work verify run (manual export)"
+        & brigade work verify run --target $workRepo --command "python $verifyScriptName" --no-reuse
+        if ($LASTEXITCODE -ne 0) { throw "manual export verify run failed" }
+
         $exportPath = Join-Path $acceptRoot "receipts.jsonl"
         Write-Step "receipts export miseledger"
         & brigade receipts export miseledger --target $workRepo --out $exportPath --new-only

src/brigade/receipts_cmd.py

@@ -31,6 +31,7 @@
 MISELEDGER_CURSOR_REL = Path(".brigade") / "work" / "miseledger-export-cursor.json"
 MISELEDGER_EXPORT_RESULT_SCHEMA = "brigade.miseledger_export_result.v1"
 MISELEDGER_FLEET_EXPORT_RESULT_SCHEMA = "brigade.miseledger_fleet_export_result.v1"
+MISELEDGER_INDEX_RESULT_SCHEMA = "brigade.miseledger_index_result.v1"
 _FLEET_STATUS_PRECEDENCE = ("failed", "exported", "nothing-new", "empty")
 _CODE_REFERENCE_LIMIT = 100
 _COMPACT_CODE_REFERENCE_NODE_LIMIT = 20
@@ -1405,6 +1406,17 @@ def _print_export_result(payload: dict[str, Any]) -> None:
     print(json.dumps(payload, sort_keys=True, separators=(",", ":")))
 
 
+def _advance_miseledger_cursor(target: Path, cursor_hashes: set[str], written_hashes: list[str]) -> int:
+    if not written_hashes:
+        return 0
+    try:
+        _write_miseledger_cursor_hashes(target, cursor_hashes | set(written_hashes))
+    except OSError as exc:
+        print(f"error: could not write cursor {_miseledger_cursor_path(target)}: {exc}", file=sys.stderr)
+        return 1
+    return 0
+
+
 def _finalize_repo_cursors(
     repo_results: list[dict[str, Any]],
     *,
@@ -1423,14 +1435,71 @@ def _finalize_repo_cursors(
         cursor_hashes = result.get("cursor_hashes")
         if not isinstance(cursor_hashes, set):
             cursor_hashes = set()
-        try:
-            _write_miseledger_cursor_hashes(target, cursor_hashes | set(written_hashes))
-        except OSError as exc:
-            print(f"error: could not write cursor {_miseledger_cursor_path(target)}: {exc}", file=sys.stderr)
-            return 1
+        exit_code = _advance_miseledger_cursor(target, cursor_hashes, written_hashes)
+        if exit_code != 0:
+            return exit_code
     return 0
 
 
+def _index_result_payload(result: dict[str, Any], *, status: str) -> dict[str, Any]:
+    return {
+        "schema": MISELEDGER_INDEX_RESULT_SCHEMA,
+        "status": status,
+        "candidate_count": int(result["candidate_count"]),
+        "exported_count": int(result["exported_count"]),
+        "skipped_count": int(result["skipped_count"]),
+        "error_count": int(result["error_count"]),
+    }
+
+
+def index_miseledger_receipts(*, target: Path, quiet: bool = False) -> dict[str, Any]:
+    """Export and import pending local receipts via the new-only adapter path.
+
+    Fail-open: always returns a status payload and never raises.
+    """
+    result = _repository_export_result(target=target, limit=0, new_only=True)
+    lines = result["lines"]
+    candidate_count = int(result["candidate_count"])
+    exported_count = int(result["exported_count"])
+    skipped_count = int(result["skipped_count"])
+
+    if candidate_count == 0:
+        payload = _index_result_payload(result, status="empty")
+        payload["message"] = "miseledger indexing: no receipts"
+        return payload
+
+    if not lines:
+        status = "nothing-new" if skipped_count > 0 else "empty"
+        payload = _index_result_payload(result, status=status)
+        payload["message"] = f"miseledger indexing: {status}"
+        return payload
+
+    output_path = _temporary_miseledger_export_path(target)
+    exit_code, written_hashes = _write_miseledger_lines_to_path(output_path, lines)
+    if exit_code != 0:
+        payload = _index_result_payload(result, status="failed")
+        payload["message"] = "miseledger indexing: export write failed"
+        return payload
+
+    attempted, import_failed = _import_miseledger_file(output_path, strict=False, quiet=quiet, failed_on_error=True)
+    if attempted and import_failed:
+        payload = _index_result_payload(result, status="failed")
+        payload["message"] = "miseledger indexing: import failed (receipts pending)"
+        return payload
+
+    cursor_hashes = result.get("cursor_hashes")
+    if not isinstance(cursor_hashes, set):
+        cursor_hashes = set()
+    if _advance_miseledger_cursor(target, cursor_hashes, written_hashes) != 0:
+        payload = _index_result_payload(result, status="failed")
+        payload["message"] = "miseledger indexing: cursor write failed"
+        return payload
+
+    payload = _index_result_payload(result, status="indexed")
+    payload["message"] = f"miseledger indexing: indexed {exported_count} receipt(s)"
+    return payload
+
+
 def _write_aggregate_lines(output_path: Path, lines: list[tuple[str, str]]) -> tuple[int, list[str]]:
     return _write_miseledger_lines_to_path(output_path, lines)
 
@@ -1501,19 +1570,30 @@ def _export_miseledger_fleet(
         if result_written:
             written_hashes_by_target[result["target"]] = result_written
 
-    cursor_exit = _finalize_repo_cursors(
-        repo_results, new_only=new_only, written_hashes_by_target=written_hashes_by_target
-    )
+    cursor_exit = 0
+    if not import_miseledger:
+        cursor_exit = _finalize_repo_cursors(
+            repo_results, new_only=new_only, written_hashes_by_target=written_hashes_by_target
+        )
     if cursor_exit != 0:
         payload = _fleet_export_payload(repo_results, status="failed")
         _print_export_result(payload)
         return 1
 
     import_error_count = 0
+    import_failed = False
     if import_miseledger and aggregate_lines:
-        attempted, failed = _import_miseledger_file(output_path, strict=True, quiet=True)
-        if attempted and failed:
+        attempted, import_failed = _import_miseledger_file(output_path, strict=True, quiet=True)
+        if attempted and import_failed:
             import_error_count = 1
+        elif not import_failed:
+            cursor_exit = _finalize_repo_cursors(
+                repo_results, new_only=new_only, written_hashes_by_target=written_hashes_by_target
+            )
+            if cursor_exit != 0:
+                payload = _fleet_export_payload(repo_results, status="failed")
+                _print_export_result(payload)
+                return 1
 
     status = _fleet_export_status([str(result["status"]) for result in repo_results])
     if import_error_count:
@@ -1618,31 +1698,49 @@ def export_miseledger(
     if json_output:
         json_output_path = Path(out).expanduser()
         exit_code, written_hashes = _write_miseledger_lines_to_path(json_output_path, lines)
-        if new_only and written_hashes:
-            cursor_hashes = result["cursor_hashes"]
-            if not isinstance(cursor_hashes, set):
-                cursor_hashes = set()
-            try:
-                _write_miseledger_cursor_hashes(target, cursor_hashes | set(written_hashes))
-            except OSError as exc:
-                print(f"error: could not write cursor {_miseledger_cursor_path(target)}: {exc}", file=sys.stderr)
-                failed_result = dict(result)
-                failed_result["status"] = "failed"
-                failed_result["error_count"] = int(failed_result["error_count"]) + 1
-                _print_export_result(_single_repo_export_payload(failed_result))
-                return 1
         if exit_code != 0:
             failed_result = dict(result)
             failed_result["status"] = "failed"
             failed_result["error_count"] = int(failed_result["error_count"]) + 1
             _print_export_result(_single_repo_export_payload(failed_result))
             return 1
 
+        import_error_count = 0
         if import_miseledger:
             if lines:
                 attempted, import_failed = _import_miseledger_file(json_output_path, strict=True, quiet=True)
                 if attempted and import_failed:
                     import_error_count = 1
+                elif not import_failed and new_only and written_hashes:
+                    cursor_hashes = result["cursor_hashes"]
+                    if not isinstance(cursor_hashes, set):
+                        cursor_hashes = set()
+                    if _advance_miseledger_cursor(target, cursor_hashes, written_hashes) != 0:
+                        failed_result = dict(result)
+                        failed_result["status"] = "failed"
+                        failed_result["error_count"] = int(failed_result["error_count"]) + 1
+                        _print_export_result(_single_repo_export_payload(failed_result))
+                        return 1
+            elif new_only and written_hashes:
+                cursor_hashes = result["cursor_hashes"]
+                if not isinstance(cursor_hashes, set):
+                    cursor_hashes = set()
+                if _advance_miseledger_cursor(target, cursor_hashes, written_hashes) != 0:
+                    failed_result = dict(result)
+                    failed_result["status"] = "failed"
+                    failed_result["error_count"] = int(failed_result["error_count"]) + 1
+                    _print_export_result(_single_repo_export_payload(failed_result))
+                    return 1
+        elif new_only and written_hashes:
+            cursor_hashes = result["cursor_hashes"]
+            if not isinstance(cursor_hashes, set):
+                cursor_hashes = set()
+            if _advance_miseledger_cursor(target, cursor_hashes, written_hashes) != 0:
+                failed_result = dict(result)
+                failed_result["status"] = "failed"
+                failed_result["error_count"] = int(failed_result["error_count"]) + 1
+                _print_export_result(_single_repo_export_payload(failed_result))
+                return 1
 
         payload = _single_repo_export_payload(result)
         if import_error_count:
@@ -1670,17 +1768,10 @@ def export_miseledger(
     else:
         output_path = _temporary_miseledger_export_path(target) if str(out) == "-" else Path(out).expanduser()
         exit_code, written_hashes = _write_miseledger_lines_to_path(output_path, lines)
-    if new_only and written_hashes:
-        cursor_hashes = result["cursor_hashes"]
-        if not isinstance(cursor_hashes, set):
-            cursor_hashes = set()
-        try:
-            _write_miseledger_cursor_hashes(target, cursor_hashes | set(written_hashes))
-        except OSError as exc:
-            print(f"error: could not write cursor {_miseledger_cursor_path(target)}: {exc}", file=sys.stderr)
-            return 1
     if exit_code != 0:
         return exit_code
+
+    import_failed = False
     if import_miseledger and output_path is not None:
         if lines:
             attempted, import_failed = _import_miseledger_file(output_path, failed_on_error=True)
@@ -1689,6 +1780,13 @@ def export_miseledger(
         else:
             print("nothing new; import skipped")
             return 0
+
+    if new_only and written_hashes and (not import_miseledger or not import_failed):
+        cursor_hashes = result["cursor_hashes"]
+        if not isinstance(cursor_hashes, set):
+            cursor_hashes = set()
+        if _advance_miseledger_cursor(target, cursor_hashes, written_hashes) != 0:
+            return 1
     if int(result["error_count"]) > 0:
         return 1
     return 0

src/brigade/templates/skills/brigade-work/SKILL.md

@@ -49,7 +49,7 @@ This is the +1 (passed) or -1 (failed) the learning ratchet scores. Capture on f
 Do not invent a skill name you are not running just to have something to capture against.
 
 ### 4. Export receipts so the next run can reuse them (optional stations)
-When GraphTrail and MiseLedger are installed, close the receipts-to-context loop so the next run sees measured evidence, not only a handoff:
+When GraphTrail and MiseLedger are installed, close the receipts-to-context loop so the next run sees measured evidence, not only a handoff. `brigade work verify run --capture <skill>` auto-exports and imports pending receipts after capture (fail-open when `miseledger` is absent). For backlog catch-up or fleet repos, run manually:
 ```bash
 brigade receipts export miseledger --target . --new-only --import   # fail-open if miseledger is absent
 # next brigade run attaches a capped evidence brief from MiseLedger automatically

src/brigade/work_cmd/verification.py

@@ -916,6 +916,27 @@ def verify_plan(
     return 0 if not blockers else 1
 
 
+def _attach_miseledger_indexing(
+    receipt: dict[str, object],
+    *,
+    target: Path,
+    json_output: bool,
+) -> None:
+    import contextlib
+    import io
+
+    from .. import receipts_cmd
+
+    if json_output:
+        sink = io.StringIO()
+        with contextlib.redirect_stdout(sink), contextlib.redirect_stderr(sink):
+            indexing = receipts_cmd.index_miseledger_receipts(target=target, quiet=True)
+    else:
+        indexing = receipts_cmd.index_miseledger_receipts(target=target, quiet=False)
+        print(indexing["message"])
+    receipt["miseledger_indexing"] = {key: value for key, value in indexing.items() if key != "message"}
+
+
 def verify_run(
     *,
     target: Path,
@@ -982,6 +1003,7 @@ def verify_run(
                     run_id=receipt["run_id"],
                     json_output=False,
                 )
+            _attach_miseledger_indexing(receipt, target=target, json_output=True)
         print(json.dumps(receipt, indent=2, sort_keys=True))
         return rc
     print(f"work verify run: {target}")
@@ -1002,6 +1024,7 @@ def verify_run(
             run_id=receipt["run_id"],
             json_output=False,
         )
+        _attach_miseledger_indexing(receipt, target=target, json_output=False)
     return rc
 
 

tests/test_ci_workflow.py

@@ -295,6 +295,9 @@ def test_ci_windows_native_acceptance_script_covers_required_flow():
     assert 'if ($callersOutput -notmatch "call_greet")' in text
     assert "brigadewinacceptance" in text
     assert "brigade work verify run" in text
+    assert "brigade work verify run (manual export)" in text
+    assert "--capture brigade-work" in text
+    assert "--no-reuse" in text
     assert "brigade receipts export miseledger" in text
     assert "import adapter" in text
     assert "$importPayload.inserted_items" in text