Setting the file. One moment.
Render Run Report · Exploring Scouts · PostHog/skills · Skills Docs
Back to the top of the page — line 202
This file
Number 66.5
Position 5 of 5
Type Python
Size 14 KB
Lines 353 scripts/ render_run_report.py
Python · 353 lines · 14 KB
15 `tool_call_update` stream, so excluding updates discards what each tool
16 actually ran. This script reassembles them. Omit --log for a metadata-only
17 report (summary mode does not need it).
18 --scratchpad scout-scratchpad-search --json (optional) durable memory
19 --config scout-config-list --json (optional) for emit posture
20
21 Modes (--mode, default: detailed):
22 summary header + posture + end-of-run summary prose. No timeline. (--log optional)
23 detailed summary + a shell-style timeline (agent narration + tool calls WITH inputs)
24 + tool tally + scratchpad. The default.
25 full detailed + each tool call's (truncated) output inline.
26
27 Output is plain text (terminal-friendly). Pipe to a .txt file with --out.
28
29 Usage:
30 python render_run_report.py --run run.json --log log.json
31 python render_run_report.py --run run.json --mode summary --no-art
32 python render_run_report.py --run run.json --log log.json --mode full --out report.txt
33
34 Stdlib only. Python 3.11+ (uses datetime.fromisoformat with offsets)."""
35
36 from __future__ import annotations
37
38 import sys
39 import json
40 import argparse
41 import textwrap
42 from collections import OrderedDict
43 from datetime import datetime
44 from typing import Any
45
46 WIDTH = 66
47
48 # the obligatory hedgehog
49 HEDGEHOG = r """
50 /////////,
51 ///////////// . PostHog · Signals
52 /////////////// ` . scout run report
53 //////////////// o ` .
54 ````````````````` `- . >
55 ' ' ' ' '
56 """
57
58
59 def rule (title: str = "" , width: int = WIDTH ) -> str :
60 if not title:
61 return "-" * width
62 body = f "---- { title } "
63 return body + "-" * max ( 0 , width - len (body))
64
65
66 def load (path: str ) -> Any:
67 with open (path, encoding = "utf-8" ) as fh:
68 return json.load(fh)
69
70
71 def parse_ts (ts: str | None ) -> datetime | None :
72 if not ts:
73 return None
74 try :
75 return datetime.fromisoformat(ts.replace( "Z" , "+00:00" ))
76 except ValueError :
77 return None
78
79
80 def hms (ts: str | None ) -> str :
81 dt = parse_ts(ts)
82 return dt.strftime( "%H:%M:%S" ) if dt else "??:??:??"
83
84
85 def human_duration (start: str | None , end: str | None ) -> str :
86 a, b = parse_ts(start), parse_ts(end)
87 if not a or not b:
88 return "unknown"
89 secs = int ((b - a).total_seconds())
90 m, s = divmod (secs, 60 )
91 return f " { m } m { s :02d} s" if m else f " { s } s"
92
93
94 def _rows (payload: Any) -> list[ dict ]:
95 """MCP list payloads come back either as a bare list or wrapped in {results: [...]}."""
96 if isinstance (payload, dict ):
97 inner = payload.get( "results" )
98 return inner if isinstance (inner, list ) else []
99 return payload if isinstance (payload, list ) else []
100
101
102 # --- session log reconstruction --------------------------------------------
103
104
105 def _updates (log: list[ dict ]) -> list[tuple[ str , dict ]]:
106 """Yield (timestamp, update) for every session/update notification, in order."""
107 out: list[tuple[ str , dict ]] = []
108 for ev in log:
109 note = ev.get( "notification" ) or {}
110 if note.get( "method" ) != "session/update" :
111 continue
112 upd = (note.get( "params" ) or {}).get( "update" )
113 if isinstance (upd, dict ):
114 out.append((ev.get( "timestamp" , "" ), upd))
115 return out
116
117
118 def reconstruct (log: list[ dict ]) -> list[ dict ]:
119 """Collapse the streamed session log into an ordered list of timeline events.
120
121 Tool calls arrive as one `tool_call` event (empty input) followed by a stream
122 of `tool_call_update`s that build `rawInput` token by token and finish with a
123 `status`+`rawOutput` event. We group by `toolCallId` and keep the richest input
124 and the final output, then emit one event per tool call at its first timestamp.
125 """
126 calls: OrderedDict[ str , dict ] = OrderedDict()
127 timeline: list[ dict ] = []
128
129 for ts, upd in _updates(log):
130 kind = upd.get( "sessionUpdate" )
131
132 if kind == "user_message_chunk" :
133 txt = (upd.get( "content" ) or {}).get( "text" , "" )
134 timeline.append({ "t" : ts, "type" : "prompt" , "text" : txt})
135
136 elif kind == "agent_message" :
137 txt = (upd.get( "content" ) or {}).get( "text" , "" )
138 if txt.strip():
139 timeline.append({ "t" : ts, "type" : "say" , "text" : txt})
140
141 elif kind in ( "tool_call" , "tool_call_update" ):
142 cid = upd.get( "toolCallId" )
143 if not cid:
144 continue
145 rec = calls.get(cid)
146 if rec is None :
147 rec = { "t" : ts, "type" : "tool" , "id" : cid, "name" : None , "input" : None , "output" : None , "status" : None }
148 calls[cid] = rec
149 timeline.append(rec)
150 # the tool name shows up on some events as _meta.claudeCode.toolName, on others as title
151 name = (((upd.get( "_meta" ) or {}).get( "claudeCode" ) or {}).get( "toolName" )) or upd.get( "title" )
152 if name:
153 rec[ "name" ] = name
154 ri = upd.get( "rawInput" )
155 if isinstance (ri, dict ) and ri:
156 # keep the input with the most total content (last full one wins)
157 if rec[ "input" ] is None or len (json.dumps(ri)) >= len (json.dumps(rec[ "input" ])):
158 rec[ "input" ] = ri
159 if upd.get( "status" ):
160 rec[ "status" ] = upd[ "status" ]
161 ro = upd.get( "rawOutput" )
162 if ro is not None :
163 rec[ "output" ] = ro
164
165 timeline.sort( key =lambda e: e[ "t" ])
166 return timeline
167
168
169 # --- input/output prettying -------------------------------------------------
170
171
172 def summarize_input (inp: dict | None , width: int ) -> str :
173 if not inp:
174 return ""
175 # The most useful single field for each common tool.
176 for key in ( "command" , "query" , "text" ):
177 if key in inp and isinstance (inp[key], str ):
178 val = " " .join(inp[key].split())
179 return val if len (val) <= width else val[: width - 1 ] + "..."
180 blob = json.dumps(inp, ensure_ascii = False )
181 return blob if len (blob) <= width else blob[: width - 1 ] + "..."
182
183
184 def summarize_output (out: Any, width: int ) -> str :
185 if out is None :
186 return ""
187 blob = json.dumps(out, ensure_ascii = False ) if isinstance (out, ( dict , list )) else str (out)
188 blob = " " .join(blob.split())
189 return blob if len (blob) <= width else blob[: width - 1 ] + "..."
190
191
192 def emit_posture (config: Any, skill_name: str ) -> dict | None :
193 for row in _rows(config):
194 if row.get( "skill_name" ) == skill_name:
195 return row
196 return None
197
198
199 # --- rendering --------------------------------------------------------------
200
201
202 def render_header (L: list[ str ], run: dict , posture: dict | None , base_url: str ) -> None :
203 name = run.get( "skill_name" , "unknown-scout" )
204 status = run.get( "status" , "?" )
205 status_tag = { "completed" : "done" , "failed" : "FAILED" }.get(status, status)
206 dur = human_duration(run.get( "started_at" ), run.get( "completed_at" ))
207 task_url = run.get( "task_url" , "" )
208 full_url = (base_url.rstrip( "/" ) + task_url) if task_url.startswith( "/" ) else task_url
209
210 L.append( "=" * WIDTH )
211 L.append( f " SIGNALS SCOUT RUN { name } " )
212 L.append( "=" * WIDTH )
213 L.append( f " run { run.get( 'run_id' , '?' ) } (skill v { run.get( 'skill_version' , '?' ) } )" )
214 L.append( f " status { status_tag } { hms(run.get( 'started_at' )) } -> { hms(run.get( 'completed_at' )) } (~ { dur } )" )
215 if posture:
216 emit = "live (emit: true)" if posture.get( "emit" ) else "DRY-RUN (emit: false)"
217 enabled = "enabled" if posture.get( "enabled" ) else "disabled"
218 L.append( f " posture { enabled } · { emit } · every { posture.get( 'run_interval_minutes' , '?' ) } m" )
219 if full_url:
220 L.append( f " transcript { full_url } " )
221 L.append( "" )
222
223
224 def render_timeline (L: list[ str ], timeline: list[ dict ], * , show_output: bool , input_width: int , output_width: int ) -> None :
225 L.append(rule( "timeline" ))
226 legend = " markers: # narration $ tool call (with input) ~ prompt"
227 if show_output:
228 legend += " => output"
229 L.append(legend)
230 L.append( "" )
231 for ev in timeline:
232 t = hms(ev[ "t" ])
233 if ev[ "type" ] == "prompt" :
234 txt = " " .join(ev[ "text" ].split())
235 L.append( f " { t } ~ { txt[: 140 ] }{ '...' if len (txt) > 140 else '' } " )
236 elif ev[ "type" ] == "say" :
237 txt = " " .join(ev[ "text" ].split())
238 for i, line in enumerate (textwrap.wrap(txt, WIDTH - 14 ) or [ "" ]):
239 L.append( f " { t } # { line } " if i == 0 else f " { ' ' * 8 } { line } " )
240 elif ev[ "type" ] == "tool" :
241 st = ev.get( "status" ) or ""
242 st_tag = { "completed" : "" , "failed" : " [FAILED]" }.get(st, f " [ { st } ]" if st else "" )
243 L.append( f " { t } $ { ev[ 'name' ] }{ st_tag } " )
244 inp = summarize_input(ev[ "input" ], input_width)
245 if inp:
246 L.append( f " { ' ' * 8 } { inp } " )
247 if show_output:
248 outp = summarize_output(ev[ "output" ], output_width)
249 if outp:
250 L.append( f " { ' ' * 8 } => { outp } " )
251 L.append( "" )
252
253 tally: dict[ str , int ] = {}
254 for ev in timeline:
255 if ev[ "type" ] == "tool" :
256 tally[ev[ "name" ] or "?" ] = tally.get(ev[ "name" ] or "?" , 0 ) + 1
257 if tally:
258 parts = ", " .join( f " { k } x { v } " for k, v in sorted (tally.items(), key =lambda kv: - kv[ 1 ]))
259 L.append( f " tool budget: { sum (tally.values()) } calls — { parts } " )
260 L.append( "" )
261
262
263 def render_summary (L: list[ str ], run: dict ) -> None :
264 summary = run.get( "summary" )
265 if not summary:
266 return
267 L.append(rule( "end-of-run summary (scout's own close-out)" ))
268 for para in summary.split( " \n " ):
269 if not para.strip():
270 L.append( "" )
271 continue
272 for line in textwrap.wrap(para, WIDTH - 1 ):
273 L.append( f " { line } " )
274 L.append( "" )
275
276
277 def render_scratchpad (L: list[ str ], scratchpad: Any) -> None :
278 rows = _rows(scratchpad)
279 if not rows:
280 return
281 L.append(rule( "scratchpad memory referenced / written" ))
282 for row in rows:
283 key = row.get( "key" , "?" )
284 content = " " .join((row.get( "content" ) or "" ).split())
285 L.append( f " * { key } " )
286 for line in textwrap.wrap(content[: 600 ], WIDTH - 6 ):
287 L.append( f " { line } " )
288 L.append( "" )
289
290
291 def render (run: dict , timeline: list[ dict ] | None , scratchpad: Any, posture: dict | None , * ,
292 mode: str , base_url: str , art: bool , show_output: bool , input_width: int , output_width: int ) -> str :
293 L: list[ str ] = []
294 if art:
295 L.append( HEDGEHOG .strip( " \n " ))
296 L.append( "" )
297 render_header(L, run, posture, base_url)
298
299 if mode != "summary" and timeline:
300 render_timeline(L, timeline, show_output = show_output, input_width = input_width, output_width = output_width)
301
302 render_summary(L, run)
303
304 if mode != "summary" :
305 render_scratchpad(L, scratchpad)
306
307 return " \n " .join(L)
308
309
310 def main () -> int :
311 ap = argparse.ArgumentParser( description = __doc__ , formatter_class = argparse.RawDescriptionHelpFormatter)
312 ap.add_argument( "--run" , required = True , help = "scout-runs-retrieve --json payload" )
313 ap.add_argument( "--log" , help = "tasks-runs-session-logs-retrieve --json payload (FULL, no exclude_types)" )
314 ap.add_argument( "--scratchpad" , help = "scout-scratchpad-search --json payload" )
315 ap.add_argument( "--config" , help = "scout-config-list --json payload (for emit posture)" )
316 ap.add_argument( "--mode" , choices = ( "summary" , "detailed" , "full" ), default = "detailed" ,
317 help = "summary = metadata + close-out prose; detailed = + timeline w/ inputs (default); full = + tool outputs" )
318 ap.add_argument( "--show-output" , action = "store_true" , help = "include tool outputs in the timeline (implied by --mode full)" )
319 ap.add_argument( "--input-width" , type = int , default = 160 , help = "truncate tool inputs to this many chars (default 160)" )
320 ap.add_argument( "--output-width" , type = int , default = 140 , help = "truncate tool outputs to this many chars (default 140)" )
321 ap.add_argument( "--no-art" , dest = "art" , action = "store_false" , help = "skip the hedgehog banner" )
322 ap.add_argument( "--base-url" , default = "https://us.posthog.com" )
323 ap.add_argument( "--out" , help = "write here instead of stdout (use a .txt path)" )
324 args = ap.parse_args()
325
326 run = load(args.run)
327 log = load(args.log) if args.log else None
328 scratchpad = load(args.scratchpad) if args.scratchpad else None
329 config = load(args.config) if args.config else None
330
331 if args.mode != "summary" and log is None :
332 print ( f "note: --mode { args.mode } wants --log for the timeline; rendering metadata only." , file = sys.stderr)
333
334 timeline = reconstruct(log) if log else None
335 posture = emit_posture(config, run.get( "skill_name" , "" )) if config else None
336 report = render(
337 run, timeline, scratchpad, posture,
338 mode = args.mode, base_url = args.base_url, art = args.art,
339 show_output = args.show_output or args.mode == "full" ,
340 input_width = args.input_width, output_width = args.output_width,
341 )
342
343 if args.out:
344 with open (args.out, "w" , encoding = "utf-8" ) as fh:
345 fh.write(report + " \n " )
346 print ( f "wrote { args.out } " , file = sys.stderr)
347 else :
348 print (report)
349 return 0
350
351
352 if __name__ == "__main__" :
353 sys.exit(main())