Coverage for little_loops / generate_schemas.py: 0%
31 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-04-11 23:19 -0500
1"""JSON Schema generation for all 19 LLEvent types.
3Generates one JSON Schema (draft-07) file per event type to docs/reference/schemas/.
4Schemas validate the flat wire format: {"event": type, "ts": timestamp, ...payload}.
6Usage:
7 python -m little_loops.generate_schemas [--output OUTPUT_DIR]
9Or via CLI:
10 ll-generate-schemas [--output OUTPUT_DIR]
11"""
13from __future__ import annotations
15import json
16from pathlib import Path
17from typing import Any
19# ---------------------------------------------------------------------------
20# Schema building helpers
21# ---------------------------------------------------------------------------
23_DRAFT07 = "http://json-schema.org/draft-07/schema#"
25_BASE_PROPS: dict[str, Any] = {
26 "event": {"type": "string", "description": "Event type identifier"},
27 "ts": {"type": "string", "format": "date-time", "description": "ISO 8601 timestamp"},
28}
30_BASE_REQUIRED = ["event", "ts"]
33def _str(description: str) -> dict[str, Any]:
34 return {"type": "string", "description": description}
37def _int(description: str) -> dict[str, Any]:
38 return {"type": "integer", "description": description}
41def _bool(description: str) -> dict[str, Any]:
42 return {"type": "boolean", "description": description}
45def _nullable_str(description: str) -> dict[str, Any]:
46 return {"type": ["string", "null"], "description": description}
49def _nullable_bool(description: str) -> dict[str, Any]:
50 return {"type": ["boolean", "null"], "description": description}
53def _schema(
54 event_type: str,
55 title: str,
56 description: str,
57 extra_props: dict[str, Any],
58 extra_required: list[str] | None = None,
59) -> dict[str, Any]:
60 """Build a complete JSON Schema dict for an event type."""
61 return {
62 "$schema": _DRAFT07,
63 "$id": f"little-loops://event-{event_type}.json",
64 "title": title,
65 "description": description,
66 "type": "object",
67 "required": _BASE_REQUIRED + (extra_required or []),
68 "properties": {**_BASE_PROPS, **extra_props},
69 "additionalProperties": True,
70 }
73# ---------------------------------------------------------------------------
74# Schema definitions — all 19 LLEvent types
75# Source of truth: docs/reference/EVENT-SCHEMA.md
76# ---------------------------------------------------------------------------
78SCHEMA_DEFINITIONS: dict[str, dict[str, Any]] = {
79 # FSM Executor (11 types)
80 "loop_start": _schema(
81 "loop_start",
82 "Loop Start",
83 "Emitted when an FSM loop begins execution.",
84 {"loop": _str("Loop name")},
85 ["loop"],
86 ),
87 "state_enter": _schema(
88 "state_enter",
89 "State Enter",
90 "Emitted when the FSM enters a state.",
91 {
92 "state": _str("State name"),
93 "iteration": _int("Iteration count (1-based)"),
94 },
95 ["state", "iteration"],
96 ),
97 "route": _schema(
98 "route",
99 "Route",
100 "Emitted when the FSM transitions between states.",
101 {
102 "from": _str("Source state name"),
103 "to": _str("Destination state name"),
104 "reason": _str("Optional transition reason"),
105 },
106 ["from", "to"],
107 ),
108 "action_start": _schema(
109 "action_start",
110 "Action Start",
111 "Emitted when a state action begins.",
112 {
113 "action": _str("Action name or command"),
114 "is_prompt": _bool("True if action is a Claude prompt, false for shell command"),
115 },
116 ["action", "is_prompt"],
117 ),
118 "action_output": _schema(
119 "action_output",
120 "Action Output",
121 "Emitted for each line of output from a running action.",
122 {"line": _str("Output line text")},
123 ["line"],
124 ),
125 "action_complete": _schema(
126 "action_complete",
127 "Action Complete",
128 "Emitted when an action finishes.",
129 {
130 "exit_code": _int("Process exit code (0 = success)"),
131 "duration_ms": _int("Execution duration in milliseconds"),
132 "output_preview": _nullable_str("Short preview of output, null if none"),
133 "is_prompt": _bool("True if action was a Claude prompt"),
134 "session_jsonl": _nullable_str(
135 "Path to Claude session JSONL file (prompt-only, null for shell commands)"
136 ),
137 },
138 ["exit_code", "duration_ms", "is_prompt"],
139 ),
140 "evaluate": _schema(
141 "evaluate",
142 "Evaluate",
143 "Emitted when an evaluator runs against action output.",
144 {
145 "type": _str("Evaluator type identifier"),
146 "verdict": _str("Evaluator verdict (e.g. pass, fail, retry)"),
147 },
148 ["type", "verdict"],
149 ),
150 "retry_exhausted": _schema(
151 "retry_exhausted",
152 "Retry Exhausted",
153 "Emitted when all retries for a state are exhausted.",
154 {
155 "state": _str("State name that exhausted retries"),
156 "retries": _int("Number of retries attempted"),
157 "next": _str("Next state the FSM transitions to"),
158 },
159 ["state", "retries", "next"],
160 ),
161 "handoff_detected": _schema(
162 "handoff_detected",
163 "Handoff Detected",
164 "Emitted when a context-limit handoff is detected in a prompt action.",
165 {
166 "state": _str("State name where handoff was detected"),
167 "iteration": _int("Iteration count at handoff"),
168 "continuation": _str("Continuation prompt text"),
169 },
170 ["state", "iteration", "continuation"],
171 ),
172 "handoff_spawned": _schema(
173 "handoff_spawned",
174 "Handoff Spawned",
175 "Emitted when a new process is spawned to continue after a handoff.",
176 {
177 "pid": _int("Process ID of the spawned continuation process"),
178 "state": _str("State name the continuation will resume from"),
179 },
180 ["pid", "state"],
181 ),
182 "loop_complete": _schema(
183 "loop_complete",
184 "Loop Complete",
185 "Emitted when an FSM loop finishes execution.",
186 {
187 "final_state": _str("Name of the terminal state reached"),
188 "iterations": _int("Total number of iterations executed"),
189 "terminated_by": _str(
190 "What caused loop termination (e.g. terminal_state, max_iterations)"
191 ),
192 },
193 ["final_state", "iterations", "terminated_by"],
194 ),
195 # FSM Persistence (1 type)
196 "loop_resume": _schema(
197 "loop_resume",
198 "Loop Resume",
199 "Emitted when a previously interrupted loop resumes from a persisted checkpoint.",
200 {
201 "loop": _str("Loop name"),
202 "from_state": _str("State the loop resumes from"),
203 "iteration": _int("Iteration count at resume"),
204 "from_handoff": _bool("True if resuming from a context-limit handoff"),
205 "continuation_prompt": _nullable_str(
206 "Continuation prompt text (only present when from_handoff is true)"
207 ),
208 },
209 ["loop", "from_state", "iteration"],
210 ),
211 # StateManager (2 types)
212 "state.issue_completed": _schema(
213 "state.issue_completed",
214 "State: Issue Completed",
215 "Emitted by StateManager when an issue transitions to completed status.",
216 {
217 "issue_id": _str("Issue identifier"),
218 "status": {"type": "string", "enum": ["completed"], "description": "Completion status"},
219 },
220 ["issue_id", "status"],
221 ),
222 "state.issue_failed": _schema(
223 "state.issue_failed",
224 "State: Issue Failed",
225 "Emitted by StateManager when an issue transitions to failed status.",
226 {
227 "issue_id": _str("Issue identifier"),
228 "reason": _str("Failure reason description"),
229 "status": {"type": "string", "enum": ["failed"], "description": "Failure status"},
230 },
231 ["issue_id", "reason", "status"],
232 ),
233 # Issue Lifecycle (4 types)
234 "issue.failure_captured": _schema(
235 "issue.failure_captured",
236 "Issue: Failure Captured",
237 "Emitted when an issue failure is captured and persisted as a bug report.",
238 {
239 "issue_id": _str("Issue identifier"),
240 "file_path": _str("Path to the issue file"),
241 "parent_issue_id": _str("Identifier of the parent issue that failed"),
242 },
243 ["issue_id", "file_path", "parent_issue_id"],
244 ),
245 "issue.closed": _schema(
246 "issue.closed",
247 "Issue: Closed",
248 "Emitted when an issue is closed.",
249 {
250 "issue_id": _str("Issue identifier"),
251 "file_path": _str("Path to the issue file"),
252 "close_reason": _str("Reason the issue was closed"),
253 },
254 ["issue_id", "file_path", "close_reason"],
255 ),
256 "issue.completed": _schema(
257 "issue.completed",
258 "Issue: Completed",
259 "Emitted when an issue is successfully completed.",
260 {
261 "issue_id": _str("Issue identifier"),
262 "file_path": _str("Path to the completed issue file"),
263 },
264 ["issue_id", "file_path"],
265 ),
266 "issue.deferred": _schema(
267 "issue.deferred",
268 "Issue: Deferred",
269 "Emitted when an issue is deferred (parked for later).",
270 {
271 "issue_id": _str("Issue identifier"),
272 "file_path": _str("Path to the deferred issue file"),
273 "reason": _str("Reason the issue was deferred"),
274 },
275 ["issue_id", "file_path", "reason"],
276 ),
277 # Parallel Orchestrator (1 type)
278 "parallel.worker_completed": _schema(
279 "parallel.worker_completed",
280 "Parallel: Worker Completed",
281 "Emitted by the parallel orchestrator when a worker finishes processing an issue.",
282 {
283 "issue_id": _str("Issue identifier processed by the worker"),
284 "worker_name": _str("Worker name or identifier"),
285 "status": _str("Completion status (e.g. completed, failed, deferred)"),
286 "duration_seconds": {"type": "number", "description": "Wall-clock time in seconds"},
287 },
288 ["issue_id", "worker_name", "status", "duration_seconds"],
289 ),
290}
293def event_type_to_filename(event_type: str) -> str:
294 """Convert event type to safe filename (replace '.' with '_')."""
295 return event_type.replace(".", "_") + ".json"
298def generate_schemas(output_dir: Path) -> list[Path]:
299 """Generate JSON Schema files for all 19 LLEvent types.
301 Args:
302 output_dir: Directory to write schema files into. Created if it doesn't exist.
304 Returns:
305 List of paths to generated files.
306 """
307 output_dir.mkdir(parents=True, exist_ok=True)
308 generated: list[Path] = []
309 for event_type, schema in SCHEMA_DEFINITIONS.items():
310 filename = event_type_to_filename(event_type)
311 path = output_dir / filename
312 path.write_text(json.dumps(schema, indent=2) + "\n")
313 generated.append(path)
314 return generated
317if __name__ == "__main__":
318 import sys
320 output = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("docs/reference/schemas")
321 paths = generate_schemas(output)
322 print(f"Generated {len(paths)} schemas in {output}/")