🔥 Gremlins MVP: cut unrequested surface per keep/cut calls

docs/workpad/reports/2026-10-17-00-gremlins-mvp-cuts.md

@@ -23,3 +23,29 @@
 | 25 | stderr to `raw/<J>.err` | keep | Cheap debugging aid. |
 | 26 | Reader join timeout 10 s | keep | Without it an orphan holding stdout hangs the helper (spec §7 leftover). |
 | 28 | Unused `ss=None` injection parameter in `call.py` | cut | Used by no caller. |
+
+## Result
+
+Premises checked in the code before each cut. Tests: `uv run --no-project --python 3.12 python -m unittest discover -s tests` gave **24 tests, OK** (run twice).
+
+| File | Added | Removed |
+|---|---|---|
+| `grm/stream.py` | 9 | 39 |
+| `grm/call.py` | 4 | 4 |
+| `grm/cli.py` | 2 | 4 |
+| `grm/launch.py` | 1 | 1 |
+| **Total** | **16** | **48** |
+
+Net code: 32 lines removed (16 added, 48 deleted); no test was removed or changed (none existed only for a cut item).
+
+## Kept against the call
+
+- **#4 `threading.Lock` around ledger append: kept.** Premise half true: in production each spawn is its own process, so no two threads of one process append. But with the lock removed `test_store.test_parallel_appends_give_whole_lines` (96 of 100 lines) and `test_cli.test_parallel_spawns` (2 of 3 ledger lines) fail, on every run. Appends from separate handles on Windows lose lines. The lock only makes those B3 tests pass in-process; across real processes the same race is unguarded either way (the review's open "Windows append atomicity" item). Cutting it would mean weakening two B3 tests, which the call forbids. Needs a decision: either keep the lock as is, or replace it with a cross-process append (for example a lock file or an `msvcrt` byte-range lock) in a separate change.
+
+## Notes on the cuts
+
+- #12: `session_id` from the `init` event is now taken only while unset (`self.session_id is None`), which matches the old "first init wins" behaviour.
+- #28: `StreamState` is now imported at the top of `call.py` instead of lazily inside the `ss is None` branch.
+- #17: premise holds. `spawn` refuses an existing `jobs/<J>.json` (`cli.py:255`) and `_execute` writes that file before the call starts, so a note that exists afterwards was written by this job.
+- #11: nothing outside `stream.py` read `refused` or `model_refused` (code, tests, SKILL.md, templates).
+- #8, #13, #14, #22: no reader in code, tests, SKILL.md or templates. SKILL.md and templates mentioned none of the cut items, so they are unchanged.

plugins/spire/skills/gremlins/grm/call.py

@@ -6,6 +6,8 @@ import threading
 import time
 from pathlib import Path
 
+from grm.stream import StreamState
+
 POLL_S = 0.2
 READER_JOIN_S = 10.0   # an orphaned grandchild can hold stdout open after claude exits
 STOP_SIGNALS = tuple(s for s in (signal.SIGTERM, signal.SIGINT, getattr(signal, "SIGBREAK", None))
@@ -39,11 +41,9 @@ def _read(stdout, raw, ss) -> None:
         pass   # the raw file or pipe was closed after READER_JOIN_S
 
 
-def run_claude(argv, prompt: str, env, cwd, raw_path, on_start, ss=None):
+def run_claude(argv, prompt: str, env, cwd, raw_path, on_start):
     """Run one call to its end. Returns (ss, exit_code, stopped)."""
-    if ss is None:
-        from grm.stream import StreamState
-        ss = StreamState()
+    ss = StreamState()
     raw_path = Path(raw_path)
     with open(raw_path, "wb") as raw, open(raw_path.with_suffix(".err"), "wb") as err:
         child = subprocess.Popen(argv, cwd=cwd, env=env, stdin=subprocess.PIPE,

plugins/spire/skills/gremlins/grm/cli.py

@@ -14,7 +14,6 @@ from grm import call, launch, store, stream
 
 SKILL_DIR = Path(__file__).resolve().parent.parent
 NAME_RE = re.compile(r"^[A-Za-z0-9._-]+$")
-NOTE_SLACK_S = 2.0     # coarse file times: a note written right at the start still counts
 EXCERPT_CHARS = 500
 LAUNCHES = ("train", "spawn", "rebase")
 
@@ -163,7 +162,7 @@ def _execute(run: str, state: Path, r: dict, *, job, kind, base, sid, argv, prom
     note = None
     if kind == "spawn":
         p = state / "notes" / (job + ".md")
-        if p.is_file() and p.stat().st_mtime >= t0 - NOTE_SLACK_S:
+        if p.is_file():
             note = str(p)
     text = (ss.result or {}).get("result")
     rec.update(session_id=ss.session_id or sid, status=status, reason=reason, ended=stream.iso(),
@@ -255,7 +254,6 @@ def cmd_spawn(args) -> dict:
     if (state / "jobs" / (j + ".json")).exists():
         raise Refusal("job exists; pick a new --job", job=j)
     brief = _read_brief(args.brief)
-    warm = _warmth(store.ledger_lines(args.run), base, time.time())
     sid = str(uuid.uuid4())
     prompt = render_template("spawn.md", {
         "block": j, "base": b, "run": r["run"],
@@ -266,7 +264,7 @@ def cmd_spawn(args) -> dict:
                                argv=launch.fork_argv(r, base["session_id"], sid), prompt=prompt,
                                before=(base.get("model_usage"), base.get("cost_cumulative_usd")),
                                base_ctx=base.get("context_tokens"))
-    return {**_out(rec, line), "base_state": warm["base_state"]}
+    return _out(rec, line)
 
 
 def cmd_rebase(args) -> dict:

plugins/spire/skills/gremlins/grm/launch.py

@@ -38,7 +38,7 @@ SCRUB_KEEP = frozenset({
     "PATH", "HOME", "USERPROFILE",
     "ANTHROPIC_BASE_URL",      # kept: gateways need it
 })
-SCRUB_PINNED = {**ENV_PINNED_LAUNCH, "SPIRE_GREMLIN_CHILD": "1"}
+SCRUB_PINNED = ENV_PINNED_LAUNCH
 SCRUB_PINNED_BINARY = "CLAUDE_CODE_EXECPATH"   # set to the stored absolute binary path
 
 

plugins/spire/skills/gremlins/grm/stream.py

@@ -31,37 +31,23 @@ class StreamState:
     """What one `claude -p --output-format stream-json --verbose` stream told us so far."""
 
     def __init__(self):
-        self.init = None              # the system/init event
         self.session_id = None
-        self.msg_ids: list = []
-        self._ids: set = set()
         self.first_usage = None       # usage of the first assistant event (cache verdict)
         self.first_at = None          # its arrival time (the base prefix read)
         self.last_usage = None        # usage of the last assistant event (base context)
         self.last_assistant_at = None
         self.result = None            # the result event
-        self.refused = None           # text of a model-restricted warning
-        self.events = 0
 
     def feed(self, ev: dict, t: float | None) -> None:
         if not isinstance(ev, dict):
             return
-        self.events += 1
         ty = ev.get("type")
         if ty == "system":
             st = ev.get("subtype")
-            if st == "init" and self.init is None:
-                self.init = ev
-                self.session_id = ev.get("session_id") or self.session_id
-            elif (st == "informational" and ev.get("level") == "warning"
-                  and "is restricted" in str(ev.get("content", ""))):
-                self.refused = str(ev.get("content"))
+            if st == "init" and self.session_id is None:
+                self.session_id = ev.get("session_id")
         elif ty == "assistant":
             m = ev.get("message") or {}
-            mid = m.get("id")
-            if mid not in self._ids:
-                self._ids.add(mid)
-                self.msg_ids.append(mid)
             u = m.get("usage") or {}
             if self.first_usage is None:
                 self.first_usage, self.first_at = u, t
@@ -70,10 +56,6 @@ class StreamState:
             self.result = ev
             self.session_id = ev.get("session_id") or self.session_id
 
-    @property
-    def turns(self) -> int:
-        return len(self.msg_ids)
-
 
 def context_tokens(usage: dict | None) -> int:
     """Prompt-side tokens of one request: input + cache creation + cache read."""
@@ -121,18 +103,16 @@ def api_error_reason(res: dict | None) -> tuple[str, str] | None:
 def classify(ss: StreamState, exit_code, stopped: bool) -> tuple[str, str | None]:
     """(status, reason) of a finished call; the first matching case wins."""
     res = ss.result
-    match (bool(stopped), ss.refused, res, api_error_reason(res), exit_code):
-        case (True, _, _, _, _):
+    match (bool(stopped), res, api_error_reason(res), exit_code):
+        case (True, _, _, _):
             return "stopped", "stopped"
-        case (_, str(), _, _, _):
-            return "error", "model_refused"
-        case (_, _, None, _, code):
+        case (_, None, _, code):
             return "error", "exit_code" if code else "no_result"
-        case (_, _, _, (reason, _), _):
+        case (_, _, (reason, _), _):
             return "error", reason
-        case (_, _, dict() as r, _, _) if r.get("is_error"):
+        case (_, dict() as r, _, _) if r.get("is_error"):
             return "error", "is_error"
-        case (_, _, _, _, code) if code:
+        case (_, _, _, code) if code:
             return "error", "exit_code"
         case _:
             return "ok", None
@@ -161,17 +141,13 @@ def usage_from_result(result: dict, before_mu: dict, cost_before) -> dict:
         delta[m] = {f: int((v or {}).get(f) or 0) - int(b.get(f) or 0) for f in MU_FIELDS}
     grew = [m for m, d in delta.items() if any(x > 0 for x in d.values())]
     models_used = sorted(grew, key=lambda m: (-sum(delta[m].values()), m))
-    sums = {f: sum(d[f] for d in delta.values()) for f in MU_FIELDS}
-    mismatch = (sums["inputTokens"], sums["outputTokens"], sums["cacheCreationInputTokens"],
-                sums["cacheReadInputTokens"]) != (tokens["input"], tokens["output"],
-                                                   tokens["cache_creation"], tokens["cache_read"])
     total = result.get("total_cost_usd")
     if total is None:
         cost_delta = None
     else:
         cost_delta = round(float(total) - float(cost_before or 0.0), 6)
         total = round(float(total), 6)
-    return {"tokens": tokens, "models_used": models_used, "mismatch": mismatch,
+    return {"tokens": tokens, "models_used": models_used,
             "cost_delta": cost_delta, "total": total, "model_usage": after}
 
 
@@ -197,12 +173,6 @@ def infer_ttl(tokens: dict | None, prev):
     return prev
 
 
-def wrote_only_5m(tokens: dict | None) -> bool:
-    """The call wrote 5-minute cache entries and no 1-hour ones."""
-    return (bool(tokens) and int(tokens.get("cache_creation_5m") or 0) > 0
-            and int(tokens.get("cache_creation_1h") or 0) == 0)
-
-
 def warmth(last_hit_ts: float | None, ttl_s, now: float) -> dict:
     """warm until `margin` before last_hit + ttl, expiring until last_hit + ttl, then cold;
     margin = max(120, ttl/10). No hit at all is cold."""