Setting the file. One moment.
Fleet Survey · Exploring Scouts · PostHog/skills · Skills Docs
Back to the top of the page scripts/ fleet_survey.py
Python · 211 lines · 8 KB
15 --now ISO-8601 timestamp to compute "ago" columns against (optional; pass the
16 current time. Without it, ages are shown as raw timestamps).
17
18 Output is plain text (terminal-friendly). Pipe to a .txt file with --out.
19
20 Usage:
21 python fleet_survey.py --config cfg.json [--runs runs.json] [--now 2026-06-08T09:00:00Z]
22
23 Stdlib only. Python 3.11+."""
24
25 from __future__ import annotations
26
27 import sys
28 import json
29 import argparse
30 from datetime import datetime
31 from typing import Any
32
33
34 # the obligatory hedgehog
35 HEDGEHOG = r """
36 /////////,
37 ///////////// . PostHog · Signals
38 /////////////// ` . fleet survey
39 //////////////// o ` .
40 ````````````````` `- . >
41 ' ' ' ' '
42 """
43
44
45 def load (path: str ) -> Any:
46 with open (path, encoding = "utf-8" ) as fh:
47 return json.load(fh)
48
49
50 def rows (payload: Any) -> list[ dict ]:
51 if isinstance (payload, dict ):
52 inner = payload.get( "results" )
53 return inner if isinstance (inner, list ) else []
54 return payload if isinstance (payload, list ) else []
55
56
57 def parse_ts (ts: str | None ) -> datetime | None :
58 if not ts:
59 return None
60 try :
61 return datetime.fromisoformat(ts.replace( "Z" , "+00:00" ))
62 except ValueError :
63 return None
64
65
66 def ago (ts: str | None , now: datetime | None ) -> str :
67 dt = parse_ts(ts)
68 if not dt:
69 return "never"
70 if not now:
71 return ts or "?"
72 secs = int ((now - dt).total_seconds())
73 if secs < 0 :
74 return "future?"
75 if secs < 3600 :
76 return f " { secs // 60 } m ago"
77 if secs < 86400 :
78 return f " { secs // 3600 } h ago"
79 return f " { secs // 86400 } d ago"
80
81
82 def latest_run_per_scout (runs_payload: Any) -> dict[ str , dict ]:
83 """runs-list returns newest-first across the whole fleet; keep the first per skill."""
84 latest: dict[ str , dict ] = {}
85 for run in rows(runs_payload):
86 name = run.get( "skill_name" )
87 if name and name not in latest:
88 latest[name] = run
89 return latest
90
91
92 def run_output (run: dict ) -> str :
93 """What the run wrote, read off the run row's structured output fields.
94
95 `emitted_report_ids` / `edited_report_ids` are the report output; `emitted_count`
96 only tallies legacy signal-channel findings (always 0 on current scouts).
97 """
98 wrote = run.get( "emitted_report_ids" ) or []
99 edited = run.get( "edited_report_ids" ) or []
100 legacy = run.get( "emitted_count" ) or 0
101 parts = []
102 if wrote:
103 parts.append( f "wrote { len (wrote) } " )
104 if edited:
105 parts.append( f "edited { len (edited) } " )
106 if legacy:
107 parts.append( f "legacy-emit { legacy } " )
108 return "+" .join(parts) if parts else "quiet"
109
110
111 def table (headers: list[ str ], body: list[list[ str ]]) -> list[ str ]:
112 """Left-aligned fixed-width text table with a dashed header rule."""
113 widths = [ len (h) for h in headers]
114 for r in body:
115 for i, cell in enumerate (r):
116 widths[i] = max (widths[i], len (cell))
117
118 def fmt (r: list[ str ]) -> str :
119 return " " .join(cell.ljust(widths[i]) for i, cell in enumerate (r)).rstrip()
120
121 out = [fmt(headers), " " .join( "-" * w for w in widths)]
122 out += [fmt(r) for r in body]
123 return out
124
125
126 def render (config: Any, runs_payload: Any, now: datetime | None , * , art: bool = True ) -> str :
127 latest = latest_run_per_scout(runs_payload) if runs_payload else {}
128 scouts = sorted (rows(config), key =lambda r: r.get( "skill_name" , "" ))
129
130 banner: list[ str ] = []
131 if art:
132 banner = [ HEDGEHOG .strip( " \n " ), "" ]
133
134 if not scouts:
135 return " \n " .join([ * banner,
136 "SIGNALS SCOUT FLEET" , "" ,
137 "No scout configs registered — this project is not enrolled in the "
138 "scout fleet (or hasn't ticked yet). Nothing is running." ])
139
140 L: list[ str ] = [ * banner, "=" * 72 , f " SIGNALS SCOUT FLEET ( { len (scouts) } configured)" , "=" * 72 , "" ]
141
142 body: list[list[ str ]] = []
143 anomalies: list[ str ] = []
144 for s in scouts:
145 name = s.get( "skill_name" , "?" )
146 enabled = "yes" if s.get( "enabled" ) else "OFF"
147 posture = "live" if s.get( "emit" ) else "dry-run"
148 cadence = f " { s.get( 'run_interval_minutes' , '?' ) } m"
149 last = ago(s.get( "last_run_at" ), now)
150
151 run = latest.get(name)
152 if run:
153 st = run.get( "status" , "?" )
154 tag = { "completed" : "done" , "failed" : "FAIL" }.get(st, st)
155 outcome = f " { tag } / { run_output(run) } "
156 else :
157 outcome = "-"
158 body.append([name, enabled, posture, cadence, last, outcome])
159
160 if s.get( "last_run_at" ) is None and s.get( "enabled" ):
161 anomalies.append( f " * { name } : enabled but has NEVER run — check fleet enrolment." )
162 if not s.get( "emit" ) and s.get( "enabled" ):
163 anomalies.append( f " * { name } : stuck in DRY-RUN (emit: false) — running but posting nothing." )
164 if run and run.get( "status" ) == "failed" :
165 anomalies.append( f " * { name } : most recent run FAILED — read its session log (often a timeout)." )
166
167 L += table([ "scout" , "enabled" , "posture" , "cadence" , "last run" , "last outcome" ], body)
168
169 if anomalies:
170 L += [ "" , "-" * 72 , " worth a look" , "-" * 72 ]
171 L += sorted ( set (anomalies))
172
173 L += [ "" , "-" * 72 , " column key" , "-" * 72 ,
174 " enabled yes = scheduled to run; OFF = paused (nothing runs)" ,
175 " posture live = writes reports to the inbox; dry-run = reasons every" ,
176 " tick but posts nothing (emit=false) — the #1 'my scout is" ,
177 " broken' confusion, since it IS running, just not posting" ,
178 " cadence configured minutes between scheduled runs (run_interval_minutes)" ,
179 " last run how long ago the most recent run started ('-' = never run)" ,
180 " last outcome <status> / <output> of that run: done|FAIL, then what it wrote" ,
181 " (from emitted_report_ids / edited_report_ids on the run row;" ,
182 " 'legacy-emit' = old signal-channel findings). quiet = wrote" ,
183 " nothing, which is the healthy norm." ]
184 return " \n " .join(L)
185
186
187 def main () -> int :
188 ap = argparse.ArgumentParser( description = __doc__ , formatter_class = argparse.RawDescriptionHelpFormatter)
189 ap.add_argument( "--config" , required = True , help = "scout-config-list --json payload" )
190 ap.add_argument( "--runs" , help = "scout-runs-list --json payload (small limit)" )
191 ap.add_argument( "--now" , help = "ISO-8601 current time for 'ago' columns" )
192 ap.add_argument( "--no-art" , dest = "art" , action = "store_false" , help = "skip the hedgehog banner" )
193 ap.add_argument( "--out" , help = "write here instead of stdout (use a .txt path)" )
194 args = ap.parse_args()
195
196 config = load(args.config)
197 runs_payload = load(args.runs) if args.runs else None
198 now = parse_ts(args.now) if args.now else None
199
200 report = render(config, runs_payload, now, art = args.art)
201 if args.out:
202 with open (args.out, "w" , encoding = "utf-8" ) as fh:
203 fh.write(report + " \n " )
204 print ( f "wrote { args.out } " , file = sys.stderr)
205 else :
206 print (report)
207 return 0
208
209
210 if __name__ == "__main__" :
211 sys.exit(main())