feat(trigger): fix(seedgo): resolve all audit findings — type error, dead code, silent catch, README (DPLAN-0141 Phase 3)

- Fix check_and_rollover → run_rollover import in startup.py (type error)
- Remove dead _is_duplicate_error() from log_watcher.py + 3 tests (unused_function)
- Add logger.warning on silent catch in watchers/log_watcher.py (silent_catch)
- Update README date to 2026-04-22 (readme)
- All 24 seedgo categories at 100%, 367 tests passing

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
AIOSAI
2026-04-22 10:11:53 -07:00
co-authored by Claude Opus 4.6
parent de3de25912
commit b123351aff
5 changed files with 6 additions and 71 deletions
+2 -2
View File
@@ -4,7 +4,7 @@
**Purpose:** Event bus for AIPass. Branches fire events, registered handlers react. Decouples producers from consumers — the module that detects a condition doesn't need to know what should happen next.
**Module:** `aipass.trigger`
**Last Updated:** 2026-04-07
**Last Updated:** 2026-04-22
## Commands / Usage
@@ -129,7 +129,7 @@ trigger/
---
*Last Updated: 2026-04-07*
*Last Updated: 2026-04-22*
---
[← Back to AIPass](../../../README.md)
@@ -371,9 +371,9 @@ def _run_memory_check() -> None:
Silent failure - handlers cannot use logger or print.
"""
try:
from aipass.memory.apps.modules.rollover import check_and_rollover
from aipass.memory.apps.modules.rollover import run_rollover
check_and_rollover()
run_rollover()
except ImportError:
return # Memory not available
except Exception as exc:
@@ -373,36 +373,6 @@ def _parse_prax_log_line(log_line: str) -> Optional[Dict[str, str]]:
return None
def _is_duplicate_error(error_hash: str) -> bool:
"""
Check if error has been seen before (deduplication).
BACKWARD COMPAT: Kept for fallback when error_registry is unavailable.
Primary dedup path is now error_registry.report() (Medic v2).
Args:
error_hash: Hash of module + message
Returns:
True if this error has been seen before
"""
global _seen_error_hashes
if error_hash in _seen_error_hashes:
return True
# Add to seen set with size limit
_seen_error_hashes.add(error_hash)
if len(_seen_error_hashes) > MAX_SEEN_HASHES:
# Remove oldest entries (convert to list, slice, back to set)
_seen_error_hashes = set(list(_seen_error_hashes)[MAX_SEEN_HASHES // 2 :])
# Persist to disk after each new hash
_save_seen_hashes()
return False
def set_event_callback(callback: Callable[..., None]) -> None:
"""
Set the callback function for firing events.
@@ -280,8 +280,8 @@ class LogFileWatcher(WatchdogFileSystemEventHandler if WATCHDOG_AVAILABLE else o
last_seen=result.get("last_seen", ""),
count=error_count,
)
except Exception:
# Registry unavailable — fire error_logged as monitoring-only fallback
except Exception as exc:
logger.warning("Registry unavailable, falling back to error_logged: %s", exc)
trigger.fire("error_logged", **event_data)
json_handler.log_operation("system_log_event", {"level": level, "module": module_name})
elif level == "warning":
@@ -253,41 +253,6 @@ class TestIsStaleEntry:
assert lw._is_stale_entry(ts) is False
# ---------------------------------------------------------------------------
# Tests -- _is_duplicate_error
# ---------------------------------------------------------------------------
class TestIsDuplicateError:
"""Tests for _is_duplicate_error."""
def test_first_time_returns_false(self):
"""First occurrence of a hash is NOT a duplicate."""
lw = _import_log_watcher()
lw._seen_error_hashes.clear()
# Patch _save_seen_hashes to avoid disk I/O
with patch.object(lw, "_save_seen_hashes"):
assert lw._is_duplicate_error("aabbccdd") is False
def test_second_time_returns_true(self):
"""Second occurrence of the same hash IS a duplicate."""
lw = _import_log_watcher()
lw._seen_error_hashes.clear()
with patch.object(lw, "_save_seen_hashes"):
lw._is_duplicate_error("aabbccdd")
assert lw._is_duplicate_error("aabbccdd") is True
def test_hash_set_size_limit(self):
"""When set exceeds MAX_SEEN_HASHES, it is trimmed."""
lw = _import_log_watcher()
lw._seen_error_hashes.clear()
with patch.object(lw, "_save_seen_hashes"):
# Fill beyond the max
for i in range(lw.MAX_SEEN_HASHES + 10):
lw._is_duplicate_error(f"hash_{i:05d}")
assert len(lw._seen_error_hashes) <= lw.MAX_SEEN_HASHES
# ---------------------------------------------------------------------------
# Tests -- set_event_callback / clear_seen_hashes
# ---------------------------------------------------------------------------