dirigence/reliquary
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
g4borg
2026-10-17 00:12:40 +02:00
parent 82ef04c118
commit e1884248e8
5 changed files with 42 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/spire/skills/gremlins/grm/stream.py
+9 -39
View File
@@ -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."""