Activity log as a decision log: per-poll observations and rule provenance

README.md

@@ -90,6 +90,15 @@ Only one runner may hold a data home's lifetime lock. Shutdown is cooperative;
 a timeout reports that stopping is still pending rather than signaling a saved
 PID or claiming the process exited.
 
+### The activity log is a decision log
+
+Every poll records what the runner observed (`check-observed` entries) before
+any events it produced, and every event's evidence names the policy rule that
+fired (`stop:`, `notify:`, `record:`). A run overtaken by a pause, cancel, or
+scope change records nothing. Replaying the log reproduces every stop/notify
+decision without re-asking GitHub. The default `nanodot activity` view hides
+per-poll observations so events stay readable; pass `--all` to include them.
+
 ### Fixed watch policy
 
 This MVP supports a fixed, validated policy. It notifies on new commits, check

docs/design/decision-log.md

@@ -0,0 +1,40 @@
+# The activity log is a decision log
+
+Status: decided 2026-10-02 (issue #35) · implemented MVP
+
+The activity log records what the runner observed, not just what changed.
+
+- Every completed poll appends a `check-observed` entry before any events it
+  produced: the commit-pinned snapshot digest plus the dedup fingerprint
+  (`statemachine.observation`). A no-change poll is reconstructable.
+- Every event's evidence names the policy rule that fired — `stop: …`,
+  `notify: …`, or `record: …`, plus for pending states the concrete
+  `pending_reason` (`statemachine.step`).
+- Entries written at the same second replay in insertion order
+  (observe → decide → act).
+
+Decisions are deterministic given `(task, snapshot, clock)`, so the log alone
+replays every stop/notify decision without re-asking GitHub — auditability as
+the runtime extension of "substitution is the proof" (adapter-seam.md).
+
+## Guarantees and boundaries
+
+- A run overtaken by pause/cancel/scope change records nothing: the
+  supersession check runs after summarizing and before any entry is written
+  or any event is delivered, and recording is all-or-nothing per run.
+- An activity-write failure never fails the run or blocks delivery; task
+  state and notifications do not depend on the log accepting an entry.
+- No new egress: the digest is the same whitelisted shape as event evidence
+  (egress.md), redacted at write time like every other entry.
+- No schema migration: `activity.evidence` is free-form JSON.
+
+## CLI
+
+`nanodot activity` hides `check-observed` by default so events stay readable;
+`--all` includes them. The `watch list` and `watch show` summaries always skip
+observations.
+
+The replay-complete record follows the pi agent harness's append-only session
+transcripts ("the transcript is the trust artifact"), keeping our own trust
+model: observations stay local, whitelisted-shaped, and consumed by code
+(the state machine), not by a human in the loop.

src/nanodot/cli.py

@@ -98,6 +98,8 @@ def build_parser() -> argparse.ArgumentParser:
     # -- activity / inbox --------------------------------------------------
     activity = subparsers.add_parser("activity", help="what actually ran")
     activity.add_argument("task_id", nargs="?")
+    activity.add_argument("--all", action="store_true",
+                          help="include per-poll check observations")
     subparsers.add_parser("inbox", help="notifications received")
 
     # -- runner ------------------------------------------------------------
@@ -181,7 +183,7 @@ def _run_config(args: argparse.Namespace) -> int:
                 file=sys.stderr,
             )
             return 1
-        except (RunnerControlError, OSError) as error:
+        except (RunnerControlError, OSError, ValueError) as error:
             print(f"error: {error}", file=sys.stderr)
             return 1
     return _run_config_values(args)
@@ -207,24 +209,48 @@ def _run_config_values(args: argparse.Namespace) -> int:
             print("error: a nonempty value is required", file=sys.stderr)
             return 1
         if is_secret_name(args.name):
-            store.set(args.name, value)
-            config.unset(args.name)  # remove a legacy plaintext copy after safe save
+            try:
+                store.set(args.name, value)
+                config.unset(args.name)  # remove a legacy plaintext copy after safe save
+            except (ValueError, OSError) as error:
+                print(f"error: {error}", file=sys.stderr)
+                return 1
         else:
             try:
                 config.set(args.name, value)
-            except ValueError as error:
+            except (ValueError, OSError) as error:
                 print(f"error: {error}", file=sys.stderr)
                 return 1
     elif args.config_command == "unset":
-        if is_secret_name(args.name):
-            store.unset(args.name)
-            config.unset(args.name)
-        else:
+        try:
+            if is_secret_name(args.name):
+                store.unset(args.name)
             config.unset(args.name)
-    else:  # list
-        rows = [(key, MASK if is_secret_name(key) else str(config.get(key)))
-                for key in config.keys() if key not in store.names()]
-        rows += [(name, MASK) for name in store.names()]
+        except (ValueError, OSError) as error:
+            print(f"error: {error}", file=sys.stderr)
+            return 1
+    else:  # list — the diagnosis command must survive a damaged state
+        try:
+            keys = config.keys()
+            secret_names = store.names()
+        except (ValueError, OSError) as error:
+            print(f"error: {error}", file=sys.stderr)
+            return 1
+        rows: list[tuple[str, str]] = []
+        for key in keys:
+            if key in secret_names:
+                continue
+            if is_secret_name(key):
+                # A legacy plaintext copy stored before the secret store
+                # existed must never be echoed back in the clear.
+                rows.append((key, MASK))
+                continue
+            try:
+                value = str(config.get(key))
+            except ValueError as error:
+                value = f"<invalid: {error}>"
+            rows.append((key, value))
+        rows += [(name, MASK) for name in secret_names]
         for key, value in sorted(rows):
             print(f"{key}={value}")
     return 0
@@ -239,6 +265,7 @@ def _run_watch(args: argparse.Namespace) -> int:
 
     from nanodot.core.activity import ActivityLog
     from nanodot.core.redaction import Redactor
+    from nanodot.core.runner import CHECK_OBSERVED
     from nanodot.core.tasks import TaskStore
 
     secrets = FileSecretStore()
@@ -251,10 +278,11 @@ def _run_watch(args: argparse.Namespace) -> int:
 
         try:
             auth_mode = Config().get("github-auth-mode", "token")
-        except ValueError as error:
+            has_token = bool(secrets.get("github-token"))
+        except (ValueError, OSError) as error:
             print(f"error: {error}", file=sys.stderr)
             return 1
-        if auth_mode == "token" and not secrets.get("github-token"):
+        if auth_mode == "token" and not has_token:
             print(
                 "error: no GitHub token configured — run: "
                 "nanodot config set github-token (hidden prompt)",
@@ -321,7 +349,12 @@ def _run_watch(args: argparse.Namespace) -> int:
             for item in relevant:
                 print(f"    - {item.content}")
         if not args.yes:
-            answer = input("Proceed? [y/N] ").strip().lower()
+            try:
+                answer = input("Proceed? [y/N] ").strip().lower()
+            except EOFError:
+                # Non-interactive stdin (cron, scripts, closed pipes) must
+                # decline, not crash — mirroring the secret-prompt path.
+                answer = "n"
             if answer not in ("y", "yes"):
                 print("cancelled")
                 return 1
@@ -345,7 +378,9 @@ def _run_watch(args: argparse.Namespace) -> int:
             print("no tasks — create one with: nanodot watch add owner/repo#1")
             return 0
         for task in tasks:
-            latest = activity.query(task_id=task.id, limit=1)
+            latest = activity.query(
+                task_id=task.id, exclude_kinds=(CHECK_OBSERVED,), limit=1
+            )
             latest_text = latest[0].message if latest else "-"
             if len(latest_text) > 60:
                 latest_text = latest_text[:57] + "..."
@@ -371,7 +406,9 @@ def _run_watch(args: argparse.Namespace) -> int:
     print(f"  state:                  {task.state.value}")
     if task.blocker:
         print(f"  blocker:                {task.blocker}")
-    entries = activity.query(task_id=task.id, limit=5)
+    entries = activity.query(
+        task_id=task.id, exclude_kinds=(CHECK_OBSERVED,), limit=5
+    )
     if entries:
         print("  recent activity:")
         for entry in reversed(entries):
@@ -420,6 +457,9 @@ def _run_memory(args: argparse.Namespace) -> int:
                 print("memory is empty")
             for item in items:
                 _print_memory_item(item)
+            total = memory.count()
+            if total > len(items):
+                print(f"... and {total - len(items)} older item(s) not shown")
         return 0
     except ValueError as error:
         print(f"error: {error}", file=sys.stderr)
@@ -467,9 +507,13 @@ def _run_approvals(_: argparse.Namespace) -> int:
 
 def _run_activity(args: argparse.Namespace) -> int:
     from nanodot.core.activity import ActivityLog
+    from nanodot.core.runner import CHECK_OBSERVED
 
     activity = ActivityLog()
-    entries = activity.query(task_id=getattr(args, "task_id", None), limit=50)
+    exclude = None if args.all else (CHECK_OBSERVED,)
+    entries = activity.query(
+        task_id=getattr(args, "task_id", None), exclude_kinds=exclude, limit=50
+    )
     if not entries:
         print("no activity yet")
         return 0

src/nanodot/core/activity.py

@@ -110,6 +110,7 @@ def query(
         self,
         task_id: str | None = None,
         kinds: tuple[str, ...] | None = None,
+        exclude_kinds: tuple[str, ...] | None = None,
         limit: int = 100,
     ) -> list[ActivityEntry]:
         clauses, params = [], []
@@ -119,11 +120,16 @@ def query(
         if kinds:
             clauses.append(f"kind IN ({','.join('?' * len(kinds))})")
             params.extend(kinds)
+        if exclude_kinds:
+            clauses.append(f"kind NOT IN ({','.join('?' * len(exclude_kinds))})")
+            params.extend(exclude_kinds)
         where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
         params.append(limit)
         with self._lock:
+            # rowid breaks ties by insertion order: entries written at the
+            # same second replay as observe → decide → act.
             rows = self._conn.execute(
-                f"SELECT * FROM activity {where} ORDER BY at DESC, id LIMIT ?",
+                f"SELECT * FROM activity {where} ORDER BY at DESC, rowid DESC LIMIT ?",
                 params,
             ).fetchall()
         return [

src/nanodot/core/config.py

@@ -2,12 +2,21 @@
 
 Secret-named keys (see redaction.is_secret_name) are routed to the secret
 store by the CLI; this file never contains secret material.
+
+Writes serialize through a sidecar flock and replace the file atomically, so
+concurrent CLI invocations can neither interleave read-modify-write cycles
+(silently reverting a key) nor expose a truncated file to readers.
 """
 
 from __future__ import annotations
 
+import fcntl
 import json
+import os
+import tempfile
+from contextlib import contextmanager
 from pathlib import Path
+from typing import Iterator
 
 from nanodot.paths import data_home
 
@@ -27,16 +36,53 @@ def _validated_value(key: str, value: object) -> object:
     return value
 
 
+@contextmanager
+def _exclusive_lock(path: Path) -> Iterator[None]:
+    """Cross-process mutual exclusion for read-modify-write cycles.
+
+    The lock file is never unlinked: an unlinked lock would let two
+    processes hold two different "locks" on the same path.
+    """
+    with open(path, "a+b") as handle:
+        fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
+        try:
+            yield
+        finally:
+            fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
+
+
+def _atomic_write(path: Path, text: str) -> None:
+    fd, temporary = tempfile.mkstemp(
+        prefix=f".{path.name}-", suffix=".tmp", dir=path.parent,
+    )
+    try:
+        with os.fdopen(fd, "w") as fh:
+            fh.write(text)
+            fh.flush()
+            os.fsync(fh.fileno())
+        os.replace(temporary, path)
+    finally:
+        try:
+            os.unlink(temporary)
+        except FileNotFoundError:
+            pass
+
+
 class Config:
     def __init__(self, base: Path | None = None) -> None:
         self._path = (base or data_home()) / CONFIG_FILE
+        self._lock = self._path.parent / f"{self._path.name}.lock"
 
-    def get(self, key: str, default: object = None) -> object:
+    def _read(self) -> dict[str, object]:
         if not self._path.exists():
-            return default
+            return {}
         values = json.loads(self._path.read_text())
         if not isinstance(values, dict):
             raise ValueError("config.json must contain a JSON object")
+        return values
+
+    def get(self, key: str, default: object = None) -> object:
+        values = self._read()
         if key not in values:
             return default
         return _validated_value(key, values[key])
@@ -45,21 +91,18 @@ def set(self, key: str, value: object) -> None:
         if key == "mode" and value != "readonly":
             raise ValueError("only readonly mode is available; gated/auto modes are not implemented")
         value = _validated_value(key, value)
-        values: dict[str, object] = {}
-        if self._path.exists():
-            values = json.loads(self._path.read_text())
-        if not isinstance(values, dict):
-            raise ValueError("config.json must contain a JSON object")
-        values[key] = value
-        self._path.write_text(json.dumps(values, indent=2))
+        with _exclusive_lock(self._lock):
+            values = self._read()
+            values[key] = value
+            _atomic_write(self._path, json.dumps(values, indent=2))
 
     def unset(self, key: str) -> None:
-        if self._path.exists():
-            values = json.loads(self._path.read_text())
-            values.pop(key, None)
-            self._path.write_text(json.dumps(values, indent=2))
+        with _exclusive_lock(self._lock):
+            values = self._read()
+            if key not in values:
+                return  # nothing to remove; never materialize an empty file
+            values.pop(key)
+            _atomic_write(self._path, json.dumps(values, indent=2))
 
     def keys(self) -> list[str]:
-        if not self._path.exists():
-            return []
-        return sorted(json.loads(self._path.read_text()))
+        return sorted(self._read())

src/nanodot/core/github_eval.py

@@ -7,7 +7,7 @@
 from nanodot.ports.github import CheckRun, Snapshot
 
 PASSING_CONCLUSIONS = frozenset({"success"})
-FAILING_CONCLUSIONS = frozenset({"failure", "timed_out", "cancelled", "action_required", "stale", "error"})
+FAILING_CONCLUSIONS = frozenset({"failure", "timed_out", "cancelled", "action_required", "stale", "startup_failure", "error"})
 IGNORED_CONCLUSIONS = frozenset({"skipped", "neutral"})