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

.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"),

.agent/tools/learn.py

@@ -58,7 +58,7 @@ def _lesson_already_appended(cid):
 def stage(claim, conditions, source="learn", importance=7):
     os.makedirs(CANDIDATES, exist_ok=True)
     cid = pattern_id(claim, conditions)
-    now = datetime.datetime.now().isoformat()
+    now = datetime.datetime.now(datetime.timezone.utc).isoformat()
     candidate = {
         "id": cid,
         "key": f"manual_{cid[:6]}",
@@ -123,7 +123,7 @@ def main():
         print("\n(stopping here — run graduate.py to accept)")
         return
 
-    rationale = args.rationale or f"manual via learn.py at {datetime.datetime.now().isoformat()}"
+    rationale = args.rationale or f"manual via learn.py at {datetime.datetime.now(datetime.timezone.utc).isoformat()}"
     grad_args = [
         sys.executable,
         os.path.join(BASE, "tools", "graduate.py"),