fix(pi): rewrite adapter — inline TS hook, formula crash, decay tz bug (#24)

* fix(pi): rewrite adapter — inline TS hook, formula crash, decay tz bug

Fixes four distinct issues found after PR #17 was merged:

## 1. ModuleNotFoundError crash on brew install (Formula)
Formula/agentic-stack.rb pkgshare.install was missing harness_manager/.
install.sh dispatches to `python3 -m harness_manager.cli` but the
module was never copied into the cellar, causing an immediate crash for
every brew user regardless of adapter.

## 2. memory-hook.ts — complete rewrite (approach was wrong)
PR #17's hook used:
  - fileURLToPath(import.meta.url) → jiti leaves this undefined in
    CJS-transform mode, silently breaking HOOK_SCRIPT path resolution
  - spawn(python3, [pi_post_tool.py]) per tool_result → subprocess
    overhead + timeout complexity for every single tool call
  - No pre-filtering → Python was spawned for read/find/ls/grep too
  - Private cc._is_success / cc._importance etc. imports → breaks
    silently on any refactor of claude_code_post_tool.py

New approach (matches the working life-international reference setup):
  - process.cwd() for all path resolution — no import.meta.url
  - All importance / pain_score / action / reflection logic is inline
    TypeScript — no Python subprocess per tool call
  - Direct fs.appendFileSync to AGENT_LEARNINGS.jsonl
  - Pre-filter: bash / edit / write only (mirrors Claude Code's
    "^(Bash|Edit|Write)$" matcher). Low-importance bash successes
    (imp <= 3) are also skipped to keep the log signal-rich.
  - Typed inputs via isBashToolResult / isEditToolResult /
    isWriteToolResult type guards (exported from pi's public API)
  - cachedSha typed as string | undefined so the git cache actually works
  - session_shutdown handler runs auto_dream.py on quit/new/resume,
    mirroring Claude Code's Stop hook. cron is now an optional fallback.

## 3. adapter.json — remove from_stack pi_post_tool.py entry
The hook is self-contained TypeScript. pi_post_tool.py is no longer
deployed as an explicitly-managed adapter file. It remains in the brain
template for standalone/debug use.

## 4. decay.py — timezone-aware vs naive datetime crash
auto_dream.py → decay_old_entries() compared a timezone-aware cutoff
(datetime.now()) against naive timestamps from AGENT_LEARNINGS.jsonl,
crashing with TypeError. Fixed by using datetime.now(timezone.utc) and
normalising naive timestamps to UTC before comparison.

Tested: fresh install on lumen-review, 41/41 tests pass, doctor green.

* fix(pi): wire shutdown hook, fix edit reflection, normalise tz, atomic dream cycle

Follow-up fixes for PR #24 surfaced by independent verification (memory-hook
type cross-reference + codex review). Each fix is annotated with the failure
mode it addresses.

P0 — session_shutdown filter rejected every event
    Pi's `SessionShutdownEvent` is `{ type: "session_shutdown" }` with no
    `reason` field (verified in pi-coding-agent's types.d.ts and the emit
    site at agent-session.js:1638). The hook's `DREAM_REASONS.has(event.reason)`
    filter therefore evaluated `has(undefined) === false` for every event, so
    `auto_dream.py` never ran. Drop the filter; shutdown only fires once on
    process exit. Add a re-entrancy guard for defence in depth.

P1 — edit reflection always lost the diff
    Hook accessed `event.input.edits[0]` but Pi's `EditToolInput` is flat
    `{ path, oldText, newText }` (no `edits` array — that's MultiEdit on
    Claude Code). Reflections silently degraded to `Edited <path>` with no
    old/new content for the dream-cycle clusterer to grip. Use the flat
    fields directly.

P1 — naive-local Python timestamps + UTC decay = silent drift
    Decay's "naive == UTC" assumption is correct only if writers emit UTC.
    They didn't: post_execution, on_failure, learn, graduate, promote,
    review_state, render_lessons all wrote naive-local. Switch every writer
    to `datetime.now(timezone.utc).isoformat()` and teach every reader
    (salience, show._human_age / _daily_counts / failing_skills /
    last_dream_cycle, on_failure._count_recent_failures,
    review_state._age_factor, archive) to normalise naive timestamps to
    UTC before comparing.

P1 — one bad user regex disabled every user pattern
    Pre-fix: build one combined RegExp per list, catch any error, return
    null for both. A single typo in `hook_patterns.json` silently dropped
    all user rules. Now: validate per-fragment, incremental merge — same
    posture as `claude_code_post_tool.py`'s `_filter_valid` /
    `_build_with_fallback`.

P1 — auto_dream lost entries that landed mid-cycle
    Original PR fixed the truncate-before-lock race in `_write_entries` but
    not the read-modify-write window: an `append_jsonl()` between
    `_load_entries()` and `_write_entries(kept)` would be truncated away.
    Hold a single LOCK_EX on the episodic log across the entire cycle via
    `_episodic_locked()`. Mutually exclusive with `_episodic_io.append_jsonl`
    (same flock target).

P1 — salience over-scored future-skewed legacy rows
    Legacy naive-local timestamps re-interpreted as UTC can read as a few
    hours in the future during the migration window. `timedelta.days` then
    went negative and `recency = 10 - age*0.3` exceeded the intended cap.
    Floor age at 0; clamp recency to ≤ 10.

P2 — _cachedSha went stale across `git commit` inside a session
    Cache for the lifetime of pi was a perf win but recorded the pre-commit
    SHA on every entry after a mid-session commit. Invalidate on bash
    commands matching `git <subcmd>` for HEAD-moving subcommands.
    `[^|;&]*?` allows option flags between `git` and the subcommand
    (`git -c key=val checkout main`, `git -C path switch dev`). Includes
    `switch` which an earlier draft missed.

P3 — test_pi_install_creates_symlink_and_syncs_hook misnamed
    Asserted that `pi_post_tool.py` was synced via from_stack — the entry
    PR #24 removed. Rename the test, drop the obsolete assertion, document
    in the comment that the .py still ships in the brain template for
    standalone use.

P3 — decay archive filename used local date while cutoff was UTC
    Asymmetric. `archive_{date}.jsonl` now uses UTC date.

Plus tests/test_decay_timezone.py — six tests pinning the shapes decay must
handle: aware UTC old/recent, naive old, mixed mid-stream, malformed,
archive-filename-uses-UTC. Catches the original crash and the migration-era
silent-drift cases.

Test suite: 47 / 47 pass (was 41 before; +6 new). Lock semantics verified
manually (open-then-flock-then-ftruncate; full read-modify-write window
held under a single fd). SHA-invalidation regex spot-checked against
12 git command shapes.

---------

Co-authored-by: codejunkie99 <email@email.com>

Author: aliirz

.agent/harness/hooks/on_failure.py

@@ -12,7 +12,7 @@
 def _count_recent_failures(skill_name):
     if not os.path.exists(EPISODIC):
         return 0
-    cutoff = datetime.datetime.now() - datetime.timedelta(days=WINDOW_DAYS)
+    cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=WINDOW_DAYS)
     count = 0
     for line in open(EPISODIC):
         line = line.strip()
@@ -25,7 +25,10 @@ def _count_recent_failures(skill_name):
         if e.get("skill") != skill_name or e.get("result") != "failure":
             continue
         try:
-            if datetime.datetime.fromisoformat(e["timestamp"]) > cutoff:
+            ts = datetime.datetime.fromisoformat(e["timestamp"])
+            if ts.tzinfo is None:
+                ts = ts.replace(tzinfo=datetime.timezone.utc)
+            if ts > cutoff:
                 count += 1
         except (KeyError, ValueError):
             continue
@@ -48,7 +51,7 @@ def on_failure(skill_name, action, error, context="", confidence=0.9,
     # schema migration is recorded with its true importance and pain score;
     # the dream-cycle salience can't distinguish failure severity otherwise.
     entry = {
-        "timestamp": datetime.datetime.now().isoformat(),
+        "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
         "skill": skill_name,
         "action": action[:200],
         "result": "failure",

.agent/harness/hooks/post_execution.py

@@ -19,7 +19,7 @@ def log_execution(skill_name, action, result, success, reflection="",
     if pain_score is None:
         pain_score = 2 if success else 7
     entry = {
-        "timestamp": datetime.datetime.now().isoformat(),
+        "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
         "skill": skill_name,
         "action": action[:200],
         "result": "success" if success else "failure",

.agent/harness/salience.py

@@ -8,12 +8,17 @@ def salience_score(entry: dict) -> float:
     if not ts:
         return 0.0
     try:
-        age_days = (datetime.datetime.now()
-                    - datetime.datetime.fromisoformat(ts)).days
+        parsed = datetime.datetime.fromisoformat(ts)
+        if parsed.tzinfo is None:
+            parsed = parsed.replace(tzinfo=datetime.timezone.utc)
+        # Negative age can happen during the naive-→-UTC migration window
+        # if a legacy naive-local timestamp now reads as a few hours in the
+        # future. Floor at 0 so recency stays in [0, 10] instead of inflating.
+        age_days = max(0, (datetime.datetime.now(datetime.timezone.utc) - parsed).days)
     except ValueError:
         age_days = 999
     pain = entry.get("pain_score", 5)
     importance = entry.get("importance", 5)
     recurrence = entry.get("recurrence_count", 1)
-    recency = max(0.0, 10.0 - age_days * 0.3)
+    recency = max(0.0, min(10.0, 10.0 - age_days * 0.3))
     return recency * (pain / 10.0) * (importance / 10.0) * min(recurrence, 3)

.agent/memory/archive.py

@@ -8,11 +8,12 @@ def archive_stale_workspace(working_dir, archive_dir):
     workspace = os.path.join(working_dir, "WORKSPACE.md")
     if not os.path.exists(workspace):
         return False
-    mtime = datetime.datetime.fromtimestamp(os.path.getmtime(workspace))
-    if (datetime.datetime.now() - mtime).days < STALE_DAYS:
+    mtime = datetime.datetime.fromtimestamp(os.path.getmtime(workspace),
+                                             tz=datetime.timezone.utc)
+    if (datetime.datetime.now(datetime.timezone.utc) - mtime).days < STALE_DAYS:
         return False
     os.makedirs(archive_dir, exist_ok=True)
     dest = os.path.join(archive_dir,
-                        f"workspace_{datetime.date.today().isoformat()}.md")
+                        f"workspace_{datetime.datetime.now(datetime.timezone.utc).date().isoformat()}.md")
     shutil.move(workspace, dest)
     return True

.agent/memory/auto_dream.py

@@ -13,13 +13,21 @@
   - promotion to LESSONS.md (graduate.py does that)
   - git commit (unattended repo writes are dangerous on a host hook)
 """
-import json, os
+import contextlib, json, os
 from promote import cluster_and_extract, write_candidates
 from validate import heuristic_check
 from review_state import mark_rejected, write_review_queue_summary
 from decay import decay_old_entries
 from archive import archive_stale_workspace
 
+# fcntl is POSIX-only. On Windows the dream cycle is best-effort: concurrent
+# writers there are rare (no shutdown hook = no parallel exits), and the lack
+# of locking matches the existing _episodic_io.py fallback.
+try:
+    import fcntl  # type: ignore[import-not-found]
+except ImportError:  # pragma: no cover — Windows
+    fcntl = None  # type: ignore[assignment]
+
 ROOT = os.path.abspath(os.path.dirname(__file__))
 EPISODIC = os.path.join(ROOT, "episodic/AGENT_LEARNINGS.jsonl")
 CANDIDATES = os.path.join(ROOT, "candidates")
@@ -29,11 +37,56 @@
 CLUSTER_SIMILARITY = 0.3
 
 
-def _load_entries():
-    if not os.path.exists(EPISODIC):
-        return []
+@contextlib.contextmanager
+def _episodic_locked():
+    """Hold an exclusive flock on AGENT_LEARNINGS.jsonl across the entire
+    dream-cycle read-modify-write window.
+
+    Without a window-spanning lock, an `append_jsonl()` call that lands
+    between `_load_entries_locked()` and `_write_entries_locked(kept)` is
+    silently truncated away by the rewrite. With this context manager,
+    every appender (`_episodic_io.append_jsonl`, which takes LOCK_EX on
+    the same file) blocks until the dream cycle releases the lock.
+
+    Yields the open file descriptor so callers can read/write without
+    racing on a second open(). On Windows (no fcntl) yields None and
+    falls back to the historical best-effort behavior.
+    """
+    if fcntl is None:
+        yield None
+        return
+    os.makedirs(os.path.dirname(EPISODIC), exist_ok=True)
+    fd = os.open(EPISODIC, os.O_RDWR | os.O_CREAT, 0o644)
+    try:
+        fcntl.flock(fd, fcntl.LOCK_EX)
+        yield fd
+    finally:
+        try:
+            fcntl.flock(fd, fcntl.LOCK_UN)
+        finally:
+            os.close(fd)
+
+
+def _load_entries_locked(fd):
+    """Read all entries from the locked fd, or fall back to plain read on
+    Windows (fd is None when fcntl is unavailable).
+    """
     entries = []
-    for line in open(EPISODIC):
+    if fd is None:
+        if not os.path.exists(EPISODIC):
+            return entries
+        with open(EPISODIC) as f:
+            stream = f.read()
+    else:
+        os.lseek(fd, 0, os.SEEK_SET)
+        chunks = []
+        while True:
+            buf = os.read(fd, 65536)
+            if not buf:
+                break
+            chunks.append(buf)
+        stream = b"".join(chunks).decode("utf-8", errors="replace")
+    for line in stream.splitlines():
         line = line.strip()
         if not line:
             continue
@@ -44,10 +97,34 @@ def _load_entries():
     return entries
 
 
+def _write_entries_locked(fd, entries):
+    """Truncate-and-rewrite under the same lock _load_entries_locked used.
+
+    Holding one fd across read+write is what makes the operation atomic
+    against concurrent `append_jsonl()` calls.
+    """
+    payload = "".join(json.dumps(e) + "\n" for e in entries).encode("utf-8")
+    if fd is None:
+        # Windows: best-effort, matches _episodic_io fallback.
+        with open(EPISODIC, "w") as f:
+            f.write(payload.decode("utf-8"))
+        return
+    os.ftruncate(fd, 0)
+    os.lseek(fd, 0, os.SEEK_SET)
+    os.write(fd, payload)
+
+
+# Compatibility shims for any external caller that still imports the
+# pre-refactor names. Internal callers in run_dream_cycle use the locked
+# helpers directly so the lock spans the full cycle.
+def _load_entries():
+    with _episodic_locked() as fd:
+        return _load_entries_locked(fd)
+
+
 def _write_entries(entries):
-    with open(EPISODIC, "w") as f:
-        for e in entries:
-            f.write(json.dumps(e) + "\n")
+    with _episodic_locked() as fd:
+        _write_entries_locked(fd, entries)
 
 
 def _heuristic_prefilter(candidates_dir, semantic_dir):
@@ -86,30 +163,36 @@ def _heuristic_prefilter(candidates_dir, semantic_dir):
 
 
 def run_dream_cycle():
-    entries = _load_entries()
-    if not entries:
-        # Still refresh the review queue — candidates may have been staged in
-        # a previous cycle and the host agent loads REVIEW_QUEUE.md into every
-        # session via build_context, so a stale/missing file hides real work.
-        pending = write_review_queue_summary(CANDIDATES, REVIEW_QUEUE)
-        print(f"dream cycle: no entries (queue has {pending} pending)")
-        return
-
-    patterns = cluster_and_extract(entries, threshold=CLUSTER_SIMILARITY)
-    promotable = {k: p for k, p in patterns.items()
-                  if p.get("canonical_salience", 0) >= PROMOTION_THRESHOLD}
+    # Hold the lock across the FULL read-modify-write window. Any
+    # append_jsonl() call from another harness blocks until we release.
+    # Without this, an append landing between read and rewrite would be
+    # truncated away.
+    with _episodic_locked() as fd:
+        entries = _load_entries_locked(fd)
+        if not entries:
+            # Still refresh the review queue — candidates may have been staged
+            # in a previous cycle and the host agent loads REVIEW_QUEUE.md
+            # into every session via build_context, so a stale/missing file
+            # hides real work.
+            pending = write_review_queue_summary(CANDIDATES, REVIEW_QUEUE)
+            print(f"dream cycle: no entries (queue has {pending} pending)")
+            return
+
+        patterns = cluster_and_extract(entries, threshold=CLUSTER_SIMILARITY)
+        promotable = {k: p for k, p in patterns.items()
+                      if p.get("canonical_salience", 0) >= PROMOTION_THRESHOLD}
+
+        staged = write_candidates(promotable, CANDIDATES)
+        prefiltered = _heuristic_prefilter(CANDIDATES, SEMANTIC)
+
+        kept, archived = decay_old_entries(
+            entries, archive_dir=os.path.join(ROOT, "episodic/snapshots"))
+        _write_entries_locked(fd, kept)
+        archive_stale_workspace(
+            working_dir=os.path.join(ROOT, "working"),
+            archive_dir=os.path.join(ROOT, "episodic/snapshots"))
 
-    staged = write_candidates(promotable, CANDIDATES)
-    prefiltered = _heuristic_prefilter(CANDIDATES, SEMANTIC)
-
-    kept, archived = decay_old_entries(
-        entries, archive_dir=os.path.join(ROOT, "episodic/snapshots"))
-    _write_entries(kept)
-    archive_stale_workspace(
-        working_dir=os.path.join(ROOT, "working"),
-        archive_dir=os.path.join(ROOT, "episodic/snapshots"))
-
-    pending = write_review_queue_summary(CANDIDATES, REVIEW_QUEUE)
+        pending = write_review_queue_summary(CANDIDATES, REVIEW_QUEUE)
 
     print(
         f"dream cycle: patterns={len(patterns)} staged={staged} "

.agent/memory/decay.py

@@ -10,12 +10,15 @@
 
 
 def decay_old_entries(entries, archive_dir):
-    cutoff = datetime.datetime.now() - datetime.timedelta(days=DECAY_DAYS)
+    cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=DECAY_DAYS)
     kept, archived = [], []
     for e in entries:
         ts_str = e.get("timestamp", "")
         try:
             ts = datetime.datetime.fromisoformat(ts_str)
+            # Normalise to UTC — entries may be naive (no tz) or aware.
+            if ts.tzinfo is None:
+                ts = ts.replace(tzinfo=datetime.timezone.utc)
         except ValueError:
             kept.append(e)
             continue
@@ -26,7 +29,9 @@ def decay_old_entries(entries, archive_dir):
 
     if archived:
         os.makedirs(archive_dir, exist_ok=True)
-        path = os.path.join(archive_dir, f"archive_{datetime.date.today()}.jsonl")
+        # UTC date so archive filenames align with the UTC cutoff above.
+        today_utc = datetime.datetime.now(datetime.timezone.utc).date()
+        path = os.path.join(archive_dir, f"archive_{today_utc}.jsonl")
         with open(path, "a") as f:
             for e in archived:
                 f.write(json.dumps(e) + "\n")

.agent/memory/promote.py

@@ -138,7 +138,7 @@ def write_candidates(patterns, candidates_dir):
             if not evidence_changed and blocker_still_present:
                 continue
 
-        now = datetime.datetime.now().isoformat()
+        now = datetime.datetime.now(datetime.timezone.utc).isoformat()
         decisions = prev.get("decisions", [])
         decisions.append({"ts": now, "action": "staged", "reviewer": "auto_dream"})
 

.agent/memory/render_lessons.py

@@ -198,9 +198,9 @@ def migrate_legacy_bullets(semantic_dir):
                        for L in load_lessons(semantic_dir)}
     try:
         accepted_at = datetime.datetime.fromtimestamp(
-            os.path.getmtime(md_path)).isoformat()
+            os.path.getmtime(md_path), tz=datetime.timezone.utc).isoformat()
     except OSError:
-        accepted_at = datetime.datetime.now().isoformat()
+        accepted_at = datetime.datetime.now(datetime.timezone.utc).isoformat()
 
     migrated = 0
     for claim in bullets:

.agent/memory/review_state.py

@@ -13,7 +13,7 @@
 
 
 def _now():
-    return datetime.datetime.now().isoformat()
+    return datetime.datetime.now(datetime.timezone.utc).isoformat()
 
 
 def _touch(candidate, action, reviewer, notes="", **fields):
@@ -177,7 +177,9 @@ def _age_factor(staged_at):
         staged = datetime.datetime.fromisoformat(staged_at)
     except (ValueError, TypeError):
         return 1.0
-    age_days = (datetime.datetime.now() - staged).days
+    if staged.tzinfo is None:
+        staged = staged.replace(tzinfo=datetime.timezone.utc)
+    age_days = (datetime.datetime.now(datetime.timezone.utc) - staged).days
     return 1.0 + min(1.0, age_days / 14.0)
 
 

.agent/tools/graduate.py

@@ -154,7 +154,7 @@ def main():
     # If we crash mid-graduation, the staged candidate remains and the
     # reviewer can retry. The retry-safety block above catches the
     # specific "lesson appended but candidate not moved" scenario.
-    accepted_at = datetime.datetime.now().isoformat()
+    accepted_at = datetime.datetime.now(datetime.timezone.utc).isoformat()
     lesson = {
         "id": lesson_id,
         "claim": cand.get("claim"),