| """Canopy reader: read one AI coding agent terminal screen and build a status card from its own lines. |
| |
| Self-contained: needs only `onnxruntime`, `tokenizers` and `numpy`. Put this file next to (or point it at) |
| a model folder holding `reader.json`, `tokenizer.json` and `model.int8.onnx`. |
| |
| from canopy_reader import CanopyReader |
| reader = CanopyReader("path/to/canopy-reader-17m") |
| reading = reader.read(state, lines) # every head, as line indices and classes |
| card = reader.card(reading, lines) # the card as text |
| |
| `lines` is the screen as a list of lines, bottom last (wrapping undone, chrome stripped: see the model |
| card). `state` is a small dict: {"agent_program": "claude", "terminal_title": "...", "daintree_observed": |
| {"agent_state": "waiting", "waiting_reason": "prompt", "seconds_in_state": 12}}; only `agent_program` is |
| required. |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import math |
| import re |
| from pathlib import Path |
|
|
| import numpy as np |
|
|
|
|
| def _strip(s: str) -> str: |
| return s.strip() |
|
|
|
|
| def line_candidates(lines: list[str], rows: int) -> list[int]: |
| """The bottom `rows` lines longer than three characters.""" |
| start = max(0, len(lines) - rows) |
| return [i for i in range(start, len(lines)) if len(lines[i].strip()) > 3] |
|
|
|
|
| def _sigmoid(x: float) -> float: |
| return 1 / (1 + math.exp(-x)) |
|
|
|
|
| def _softmax(xs) -> list[float]: |
| m = max(xs) |
| e = [math.exp(x - m) for x in xs] |
| s = sum(e) |
| return [x / s for x in e] |
|
|
|
|
| class CanopyReader: |
| def __init__(self, folder: str | Path, variant: str = "int8", threads: int = 1): |
| import onnxruntime as ort |
| from tokenizers import Tokenizer |
| folder = Path(folder) |
| self.meta = json.loads((folder / "reader.json").read_text()) |
| self.tok = Tokenizer.from_file(str(folder / "tokenizer.json")) |
| opts = ort.SessionOptions() |
| opts.intra_op_num_threads = threads |
| opts.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL |
| self.session = ort.InferenceSession(str(folder / f"model.{variant}.onnx"), opts, providers=["CPUExecutionProvider"]) |
| self.outputs = [o.name for o in self.session.get_outputs()] |
| self.cls, self.sep, self.pad = (self.tok.token_to_id(t) for t in ("[CLS]", "[SEP]", "[PAD]")) |
|
|
| def _ids(self, text: str) -> list[int]: |
| return self.tok.encode(text, add_special_tokens=False).ids |
|
|
| def encode(self, state: dict, lines: list[str]) -> dict: |
| """`[CLS] state-json [SEP]`, then `[i] line\\n` for each non-blank line, filled from the bottom of the |
| screen up until `max_len` tokens, then `[SEP]`. Each line's first token is where its heads read.""" |
| m = self.meta |
| header = self._ids(json.dumps(state, ensure_ascii=False, separators=(",", ":")))[: m["header_tokens"]] |
| budget = m["max_len"] - len(header) - 3 |
| pieces = [] |
| for i in range(len(lines) - 1, -1, -1): |
| if not lines[i].strip(): |
| continue |
| ids = self._ids(f"[{i}] {lines[i]}\n")[: m["line_tokens"]] |
| if len(ids) > budget: |
| break |
| budget -= len(ids) |
| pieces.append((i, ids)) |
| pieces.reverse() |
| ids = [self.cls, *header, self.sep] |
| index, pos = [], [] |
| for i, p in pieces: |
| index.append(i) |
| pos.append(len(ids)) |
| ids.extend(p) |
| ids.append(self.sep) |
| q = set(line_candidates(lines, m["question_candidate_rows"])) |
| s = set(line_candidates(lines, m["status_candidate_rows"])) |
| return {"ids": ids, "index": index, "pos": pos, "q": [i in q for i in index], "s": [i in s for i in index]} |
|
|
| def read(self, state: dict, lines: list[str]) -> dict: |
| e = self.encode(state, lines) |
| k = max(1, len(e["index"])) |
| pad = lambda xs, v: xs + [v] * (k - len(xs)) |
| feed = { |
| "input_ids": np.array([e["ids"]], dtype=np.int64), |
| "attention_mask": np.ones((1, len(e["ids"])), dtype=np.int64), |
| "line_pos": np.array([pad(e["pos"], 0)], dtype=np.int64), |
| "line_mask": np.array([pad([True] * len(e["index"]), False)]), |
| "q_mask": np.array([pad(e["q"], False)]), |
| "s_mask": np.array([pad(e["s"], False)]), |
| } |
| out = dict(zip(self.outputs, self.session.run(self.outputs, feed))) |
| return self.decode({n: v[0].astype(np.float64).tolist() for n, v in out.items()}, e) |
|
|
| def decode(self, out: dict, e: dict) -> dict: |
| m, idx, t = len(e["index"]), e["index"], self.meta["thresholds"] |
|
|
| def pick(name): |
| scores = out[name] |
| cand = scores[:m] + [scores[-1]] |
| j = max(range(len(cand)), key=lambda x: (cand[x], -x)) |
| return -1 if j == m else idx[j] |
|
|
| logit = out["summary_lines"][:m] |
| p_sum = [_sigmoid(x) for x in logit] |
| summary = [] |
| for j in sorted(range(m), key=lambda j: (-logit[j], j))[:3]: |
| if p_sum[j] >= (t["summary_more"] if summary else t["summary_first"]): |
| summary.append(idx[j]) |
| else: |
| break |
| probs = _softmax(out["category"]) |
| cats = self.meta["categories"] |
| raw = _sigmoid(out["needs_attention"]) |
| g = max(raw, sum(probs[cats.index(c)] for c in t["gate_categories"])) |
| gate = 0.5 * g / t["gate"] if g < t["gate"] else 0.5 + 0.5 * (g - t["gate"]) / (1 - t["gate"]) |
| cls = lambda n: self.meta["classes"][n][max(range(len(out[n])), key=lambda i: (out[n][i], -i))] |
| return { |
| "category": cats[max(range(len(probs)), key=lambda i: (probs[i], -i))], "category_probs": probs, |
| "needs_attention": gate, "needs_attention_raw": raw, |
| "blocked_on_user": _sigmoid(out["blocked_on_user"]), "notable": _sigmoid(out["notable"]), |
| "question_line": pick("question_line"), "status_line": pick("status_line"), |
| "action_line": pick("action_line"), "task_line": pick("task_line"), |
| "summary_lines": summary, |
| "option_lines": [idx[j] for j in range(m) if e["q"][j] and _sigmoid(out["option_lines"][j]) >= t["option"]], |
| "key_lines": [idx[j] for j in range(m) if out["key_lines"][j] > 0], |
| **{n: cls(n) for n in self.meta["classes"]}, |
| } |
|
|
| |
|
|
| |
| LEAD = re.compile(r"^(?:[⏺•●✦⎿└├│┃║■✻✶✢✳✽✺·*❯›>▸▶◆◇](?:\s+|$))+") |
| BOX_TAIL = re.compile(r"\s*[│┃║]\s*$") |
| TIMER = re.compile(r"\s*\((?:\d[^)]*)?(?:esc to \w+|thinking|thought for|ctrl\+\w to \w+)[^)]*\)\s*$", re.I) |
| END = re.compile(r"""[.!?:;)"'`]$""") |
|
|
| @classmethod |
| def text(cls, line: str | None) -> str | None: |
| """A screen line as card text: bullets, tree and box glyphs, spinners and timers off.""" |
| if line is None: |
| return None |
| t = cls.TIMER.sub("", cls.BOX_TAIL.sub("", cls.LEAD.sub("", line.strip()))).strip() |
| return t if len(t) >= 2 else None |
|
|
| def _sentence(self, lines, i, picks, used): |
| t = self.text(lines[i]) |
| used.add(i) |
| j = i + 1 |
| while t and j <= i + 2 and j < len(lines) and not self.END.search(t) and j not in used: |
| nxt = lines[j] |
| if not nxt.strip() or self.LEAD.match(nxt.strip()): |
| break |
| if j not in picks and not (re.match(r"^\s{2,}\S", nxt) or nxt.strip()[:1].islower()): |
| break |
| used.add(j) |
| more = self.text(nxt) |
| if more: |
| t += " " + more |
| j += 1 |
| return t |
|
|
| def card(self, r: dict, lines: list[str]) -> dict: |
| asks = r["category"] in ("approval", "question") |
| used: set[int] = set() |
| line = lambda i: self.text(lines[i]) if i >= 0 else None |
| headline = self._sentence(lines, r["status_line"], r["summary_lines"], used) if r["status_line"] >= 0 else None |
| summary = [s for i in r["summary_lines"] if i not in used for s in [self._sentence(lines, i, r["summary_lines"], used)] if s] |
| return { |
| "category": r["category"], "needs_attention": r["needs_attention"] >= 0.5, "score": int(r["score"]), |
| "headline": headline, "summary": summary, |
| "question": line(r["question_line"]) if asks else None, |
| "options": [x for x in (self.text(lines[i]) for i in r["option_lines"]) if x] if asks else [], |
| "action": line(r["action_line"]) if r["category"] == "approval" else None, |
| "task": line(r["task_line"]), |
| "tests": r["tests"], "changes": r["changes"], |
| "progress": None if r["progress"] == "none" else int(r["progress"]), |
| } |
|
|
