From f09d42a8622714066fd943d83cb41767a2fe2629 Mon Sep 17 00:00:00 2001 From: AIOSAI Date: Sun, 19 Jul 2026 02:55:55 -0700 Subject: [PATCH] =?UTF-8?q?fix(flow):=20lock+atomic=20registry=20aggregate?= =?UTF-8?q?=20writes=20=E2=80=94=20S304=20F85=20residual.=20save=5Fbranch?= =?UTF-8?q?=5Fregistry=20+=20save=5Fcentral=20were=20bare=20open+dump;=20n?= =?UTF-8?q?ow=20O=5FEXCL=20lockfile=20with=20backoff=20+=20tmpfile=20os.re?= =?UTF-8?q?place,=20matching=20save=5Fregistry.py.=20Devpulse=20verified:?= =?UTF-8?q?=20730=20tests=20green=20+=206-proc=20180-write=20concurrent=20?= =?UTF-8?q?hammer,=20final=20file=20valid=20JSON.=20N1=20root=20cause=20fo?= =?UTF-8?q?und:=20is=5Ftemplate=5Fcontent=20line=20threshold=20fires=20bef?= =?UTF-8?q?ore=20bracket-marker=20check.=20CHANGELOG=20entries.=20DPLAN-02?= =?UTF-8?q?50,=20dispatched=20to=20@flow?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 16 ++++ .../flow/apps/handlers/plan/aggregate_ops.py | 87 +++++++++++++++++-- 2 files changed, 96 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9570daef..922837f3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,6 +33,22 @@ Also: `/prep` now checks the cross-project feedback box every run — the S304 "unread since April" backlog (F46) is processed to zero and can't silently rot again. +**fix(aipass)** — `aipass doctor` no longer invents errors on a healthy repo +(S304 F14-17): `.backup` and spawn `templates` dirs excluded from the agent +scan, relative registry paths anchored against the project root instead of the +caller's CWD (running doctor from inside a branch dir showed 5 fake +missing-branch errors), and info-severity findings now render as warnings +instead of pass checkmarks. Remaining findings on this repo are genuine. + +**fix(flow)** — registry aggregate writes are lock+atomic (S304 F85 residual): +`save_branch_registry` and `save_central` were bare `open`+`json.dump`; both now +use the O_EXCL lockfile + temp-file + `os.replace` pattern from the earlier +`save_registry.py` fix. Proven with a 6-process concurrent-write hammer. Also +investigated (N1): FPLAN-0313/0314 closed blank because +`is_template_content()`'s line-count threshold fires before its bracket-marker +check on default templates — guard fix queued, registry annotation is the +maintainer's call. + ## [2026-07-18] **fix(tests)** — the immortal `MagicMock/LOG_FILE/` directory is dead: a hooks diff --git a/src/aipass/flow/apps/handlers/plan/aggregate_ops.py b/src/aipass/flow/apps/handlers/plan/aggregate_ops.py index 3da1d55f..c8698f1f 100644 --- a/src/aipass/flow/apps/handlers/plan/aggregate_ops.py +++ b/src/aipass/flow/apps/handlers/plan/aggregate_ops.py @@ -17,6 +17,8 @@ Usage: """ import json +import os +import time from pathlib import Path from datetime import datetime, timezone from typing import Dict, Any, List, Tuple, Optional @@ -32,6 +34,34 @@ from aipass.flow.apps.handlers.json import json_handler MODULE_NAME = "aggregate_central" +_LOCK_RETRIES = 10 +_LOCK_BACKOFF_BASE = 0.05 + + +def _acquire_lock(lock_path: Path) -> bool: + """Atomically acquire a lockfile via O_CREAT|O_EXCL with retry+backoff.""" + for attempt in range(_LOCK_RETRIES): + try: + fd = os.open(str(lock_path), os.O_CREAT | os.O_EXCL | os.O_WRONLY) + os.write(fd, str(os.getpid()).encode()) + os.close(fd) + return True + except FileExistsError: + logger.info("[%s] Lock contention on %s, retry %d", MODULE_NAME, lock_path, attempt + 1) + time.sleep(_LOCK_BACKOFF_BASE * (2**attempt)) + except OSError as exc: + logger.warning("[%s] Lock creation failed for %s: %s", MODULE_NAME, lock_path, exc) + return False + return False + + +def _release_lock(lock_path: Path) -> None: + """Remove lockfile, tolerating already-removed.""" + try: + lock_path.unlink(missing_ok=True) + except OSError as exc: + logger.warning("[%s] Could not release lock %s: %s", MODULE_NAME, lock_path, exc) + # ============================================= # HELPER FUNCTIONS @@ -101,13 +131,36 @@ def save_branch_registry(registry_path: Path, registry: Dict[str, Any]) -> bool: Returns: True on success, False on failure """ + lock_path = registry_path.with_suffix(".lock") try: - registry["last_updated"] = datetime.now(timezone.utc).isoformat() - with open(registry_path, "w", encoding="utf-8") as f: - json.dump(registry, f, indent=2, ensure_ascii=False) + registry_path.parent.mkdir(parents=True, exist_ok=True) + + if not _acquire_lock(lock_path): + logger.error( + "[%s] Could not acquire lock for %s after %d retries", + MODULE_NAME, + registry_path, + _LOCK_RETRIES, + ) + return False + + try: + registry["last_updated"] = datetime.now(timezone.utc).isoformat() + tmp_path = registry_path.with_suffix(".tmp") + with open(tmp_path, "w", encoding="utf-8") as f: + json.dump(registry, f, indent=2, ensure_ascii=False) + os.replace(str(tmp_path), str(registry_path)) + finally: + _release_lock(lock_path) + return True except Exception as e: - logger.error(f"[{MODULE_NAME}] Failed to save registry {registry_path}: {e}") + logger.error( + "[%s] Failed to save registry %s: %s", + MODULE_NAME, + registry_path, + e, + ) return False @@ -265,10 +318,27 @@ def save_central(central_file: Path, central_dir: Path, central_data: Dict[str, Returns: True on success, False on failure """ + lock_path = central_file.with_suffix(".lock") try: central_dir.mkdir(parents=True, exist_ok=True) - with open(central_file, "w", encoding="utf-8") as f: - json.dump(central_data, f, indent=2, ensure_ascii=False) + + if not _acquire_lock(lock_path): + logger.error( + "[%s] Could not acquire lock for %s after %d retries", + MODULE_NAME, + central_file, + _LOCK_RETRIES, + ) + return False + + try: + tmp_path = central_file.with_suffix(".tmp") + with open(tmp_path, "w", encoding="utf-8") as f: + json.dump(central_data, f, indent=2, ensure_ascii=False) + os.replace(str(tmp_path), str(central_file)) + finally: + _release_lock(lock_path) + return True except Exception as e: logger.error(f"[{MODULE_NAME}] Failed to save {central_file}: {e}") @@ -392,7 +462,10 @@ def aggregate_central_impl( # Save central file if save_central(central_file, central_dir, central_data): logger.info( - f"[{MODULE_NAME}] SUCCESS: Aggregation complete: {len(all_active)} active, {len(recently_closed)} recently closed" + "[%s] SUCCESS: Aggregation complete: %d active, %d recently closed", + MODULE_NAME, + len(all_active), + len(recently_closed), ) # Fire trigger event