Skills › Compound Engineering › ce-riffrec-feedback-analysis › analyze_riffrec_zip.py
analyze_riffrec_zip.py
1306 lines · skills/ce-riffrec-feedback-analysis/scripts/analyze_riffrec_zip.py at 3449230
1#!/usr/bin/env python3
2"""
3Analyze a product feedback source.
5Supported sources: Riffrec zip or unpacked capture directory, standalone
6video, standalone audio, and meeting notes text/markdown. The script extracts
7transcript, high-signal video frames when available, and CE-friendly markdown
8artifacts.
9"""
11from __future__ import annotations
13import argparse
14import json
15import os
16import re
17import shutil
18import subprocess
19import sys
20import tempfile
21import zipfile
22from collections import Counter, defaultdict
23from datetime import datetime, timezone
24from pathlib import Path
25from typing import Any
28COMPLAINT_CUES = (
29 "weird",
30 "doesn't work",
31 "does not work",
32 "dont work",
33 "don't work",
34 "can't",
35 "cannot",
36 "broken",
37 "bug",
38 "problem",
39 "confusing",
40 "should",
41 "wrong",
42 "stuck",
43 "failed",
44)
46NOISY_NETWORK_PATTERNS = (
47 "/mini-profiler-resources/",
48 "__vite_ping",
49 "/rails/action_cable",
50)
52VIDEO_EXTENSIONS = {".webm", ".mp4", ".mov", ".m4v", ".mkv", ".avi"}
53AUDIO_EXTENSIONS = {".webm", ".mp3", ".mp4", ".mpeg", ".mpga", ".m4a", ".wav", ".ogg", ".flac"}
54NOTES_EXTENSIONS = {".txt", ".md", ".markdown", ".text"}
55RIFFREC_DIRECTORY_MARKERS = {"session.json", "events.json"}
58class SourceInputError(ValueError):
59 pass
62def parse_args() -> argparse.Namespace:
63 parser = argparse.ArgumentParser(description="Analyze a product feedback source")
64 parser.add_argument(
65 "source_path",
66 type=Path,
67 help="Path to a Riffrec zip or unpacked capture directory, video, audio, or meeting notes file",
68 )
69 parser.add_argument(
70 "--output-dir",
71 type=Path,
72 help="Directory for extracted evidence/kickoff artifacts. Defaults to docs/brainstorms/riffrec-feedback/<source-stem> when available; durable ce-brainstorm outputs live in the plans artifact directory.",
73 )
74 parser.add_argument("--topic", help="Kebab-case topic for requirements-kickoff frontmatter")
75 parser.add_argument(
76 "--model",
77 default=os.environ.get("RIFFREC_TRANSCRIBE_MODEL", "gpt-4o-mini-transcribe"),
78 help="OpenAI transcription model to use when OPENAI_API_KEY is set",
79 )
80 parser.add_argument("--no-transcribe", action="store_true", help="Skip media transcription")
81 parser.add_argument("--max-moments", type=int, default=12, help="Maximum screenshots to extract")
82 return parser.parse_args()
85def slugify(value: str) -> str:
86 slug = re.sub(r"[^a-zA-Z0-9]+", "-", value.strip().lower()).strip("-")
87 return re.sub(r"-{2,}", "-", slug) or "riffrec-feedback"
90def read_json(path: Path, default: Any) -> Any:
91 if not path.exists():
92 return default
93 try:
94 return json.loads(path.read_text())
95 except json.JSONDecodeError:
96 return default
99def safe_extract(zip_path: Path, dest: Path) -> None:
100 dest.mkdir(parents=True, exist_ok=True)
101 with zipfile.ZipFile(zip_path) as archive:
102 dest_resolved = dest.resolve()
103 for member in archive.infolist():
104 member_path = dest / member.filename
105 resolved = member_path.resolve()
106 if not resolved.is_relative_to(dest_resolved):
107 raise RuntimeError(f"Unsafe zip member path: {member.filename}")
108 if member.is_dir():
109 resolved.mkdir(parents=True, exist_ok=True)
110 else:
111 resolved.parent.mkdir(parents=True, exist_ok=True)
112 with archive.open(member) as source, resolved.open("wb") as target:
113 shutil.copyfileobj(source, target)
116def safe_copy_capture_directory(source_dir: Path, dest: Path) -> None:
117 source_root = source_dir.resolve()
118 dest.mkdir(parents=True, exist_ok=True)
119 dest_root = dest.resolve()
121 entries = sorted(source_dir.rglob("*"), key=lambda path: str(path.relative_to(source_dir)))
122 for entry in entries:
123 if entry.is_symlink():
124 raise SourceInputError(f"Unpacked Riffrec capture contains a symlink: {entry}")
125 resolved = entry.resolve()
126 if not resolved.is_relative_to(source_root):
127 raise SourceInputError(f"Unpacked Riffrec capture entry escapes its source directory: {entry}")
128 if not entry.is_dir() and not entry.is_file():
129 raise SourceInputError(f"Unpacked Riffrec capture contains an unsupported entry type: {entry}")
131 for entry in entries:
132 relative = entry.relative_to(source_dir)
133 target = dest / relative
134 if not target.resolve().is_relative_to(dest_root):
135 raise SourceInputError(f"Unsafe unpacked Riffrec capture path: {relative}")
136 if entry.is_dir():
137 target.mkdir(parents=True, exist_ok=True)
138 else:
139 target.parent.mkdir(parents=True, exist_ok=True)
140 shutil.copy2(entry, target)
143def validate_raw_destination(raw_dir: Path) -> None:
144 if raw_dir.is_symlink():
145 raise SourceInputError(f"Raw output directory must not be a symlink: {raw_dir}")
146 if raw_dir.exists() and not raw_dir.is_dir():
147 raise SourceInputError(f"Raw output path must be a directory: {raw_dir}")
148 if not raw_dir.exists():
149 return
150 for existing in raw_dir.rglob("*"):
151 if existing.is_symlink():
152 raise SourceInputError(f"Raw output contains a symlink and cannot be replaced safely: {existing}")
155def promote_raw_snapshot(staging_dir: Path, raw_dir: Path) -> None:
156 validate_raw_destination(raw_dir)
157 previous_dir = raw_dir.parent / staging_dir.name.replace(".staging-", ".previous-", 1)
158 if previous_dir.exists() or previous_dir.is_symlink():
159 raise SourceInputError(f"Temporary raw snapshot path already exists: {previous_dir}")
161 had_previous = raw_dir.exists()
162 if had_previous:
163 os.replace(raw_dir, previous_dir)
164 try:
165 os.replace(staging_dir, raw_dir)
166 except BaseException:
167 if had_previous:
168 os.replace(previous_dir, raw_dir)
169 raise
171 if had_previous:
172 shutil.rmtree(previous_dir, ignore_errors=True)
175def validate_frames_destination(frames_dir: Path) -> None:
176 if frames_dir.is_symlink():
177 raise SourceInputError(f"Frames output directory must not be a symlink: {frames_dir}")
178 if frames_dir.exists() and not frames_dir.is_dir():
179 raise SourceInputError(f"Frames output path must be a directory: {frames_dir}")
180 if not frames_dir.exists():
181 return
182 for existing in frames_dir.rglob("*"):
183 if existing.is_symlink():
184 raise SourceInputError(f"Frames output contains a symlink and cannot be replaced safely: {existing}")
187def promote_frames_snapshot(staging_dir: Path, frames_dir: Path) -> None:
188 validate_frames_destination(frames_dir)
189 previous_dir = frames_dir.parent / staging_dir.name.replace(".staging-", ".previous-", 1)
190 if previous_dir.exists() or previous_dir.is_symlink():
191 raise SourceInputError(f"Temporary frames snapshot path already exists: {previous_dir}")
193 had_previous = frames_dir.exists()
194 if had_previous:
195 os.replace(frames_dir, previous_dir)
196 try:
197 os.replace(staging_dir, frames_dir)
198 except BaseException:
199 if had_previous:
200 os.replace(previous_dir, frames_dir)
201 raise
203 if had_previous:
204 shutil.rmtree(previous_dir, ignore_errors=True)
207def default_output_dir(source_path: Path) -> Path:
208 cwd = Path.cwd()
209 stem = slugify(source_path.stem)
210 if (cwd / "docs" / "brainstorms").is_dir():
211 return cwd / "docs" / "brainstorms" / "riffrec-feedback" / stem
212 return cwd / "riffrec-feedback" / stem
215def classify_source(source_path: Path) -> str:
216 if source_path.is_dir():
217 missing = sorted(marker for marker in RIFFREC_DIRECTORY_MARKERS if not (source_path / marker).is_file())
218 if missing:
219 expected = ", ".join(sorted(RIFFREC_DIRECTORY_MARKERS))
220 missing_text = ", ".join(missing)
221 raise SourceInputError(
222 f"Unsupported source directory: {source_path}. "
223 f"An unpacked Riffrec capture must contain {expected}; missing {missing_text}."
224 )
225 return "riffrec_directory"
226 if not source_path.is_file():
227 raise SourceInputError(f"Unsupported source path type: {source_path}")
228 if zipfile.is_zipfile(source_path):
229 return "riffrec_zip"
230 suffix = source_path.suffix.lower()
231 if suffix in NOTES_EXTENSIONS:
232 return "meeting_notes"
233 if suffix in VIDEO_EXTENSIONS and suffix in AUDIO_EXTENSIONS:
234 return "video" if has_video_stream(source_path) else "audio"
235 if suffix in VIDEO_EXTENSIONS:
236 return "video"
237 if suffix in AUDIO_EXTENSIONS:
238 return "audio"
239 return "unknown"
242def ffprobe_duration(path: Path) -> float:
243 if not path.exists() or not shutil.which("ffprobe"):
244 return 0.0
245 command = [
246 "ffprobe",
247 "-v",
248 "error",
249 "-show_entries",
250 "format=duration",
251 "-of",
252 "default=noprint_wrappers=1:nokey=1",
253 str(path),
254 ]
255 result = subprocess.run(command, capture_output=True, text=True, timeout=30)
256 if result.returncode != 0:
257 return 0.0
258 try:
259 return float(result.stdout.strip())
260 except ValueError:
261 return 0.0
264def has_video_stream(path: Path) -> bool:
265 if not path.exists() or not shutil.which("ffprobe"):
266 return path.suffix.lower() in VIDEO_EXTENSIONS
267 command = [
268 "ffprobe",
269 "-v",
270 "error",
271 "-select_streams",
272 "v:0",
273 "-show_entries",
274 "stream=codec_type",
275 "-of",
276 "csv=p=0",
277 str(path),
278 ]
279 result = subprocess.run(command, capture_output=True, text=True, timeout=30)
280 return result.returncode == 0 and "video" in result.stdout
283def read_notes(path: Path) -> dict[str, Any]:
284 try:
285 text = path.read_text()
286 except UnicodeDecodeError:
287 text = path.read_text(encoding="utf-8", errors="replace")
288 return {"status": "ok", "text": text.strip(), "source": "meeting_notes"}
291def populate_source_snapshot(source_path: Path, snapshot_dir: Path, source_kind: str) -> dict[str, Any]:
292 snapshot_dir.mkdir(parents=True, exist_ok=True)
294 if source_kind in {"riffrec_zip", "riffrec_directory"}:
295 if source_kind == "riffrec_zip":
296 safe_extract(source_path, snapshot_dir)
297 else:
298 safe_copy_capture_directory(source_path, snapshot_dir)
299 missing = sorted(
300 marker for marker in RIFFREC_DIRECTORY_MARKERS if not (snapshot_dir / marker).is_file()
301 )
302 if missing:
303 missing_text = ", ".join(missing)
304 raise SourceInputError(
305 f"Unpacked Riffrec capture changed during normalization; missing {missing_text}."
306 )
307 session = read_json(snapshot_dir / "session.json", {})
308 events_payload = read_json(snapshot_dir / "events.json", {})
309 events = events_payload.get("events", events_payload if isinstance(events_payload, list) else [])
310 if not isinstance(events, list):
311 events = []
312 try:
313 duration = float(session.get("duration_seconds") or events_payload.get("duration_seconds") or 0)
314 except (TypeError, ValueError):
315 duration = 0.0
316 return {
317 "source_kind": source_kind,
318 "session": session,
319 "events": events,
320 "duration": duration,
321 "recording_path": snapshot_dir / "recording.webm",
322 "transcription_path": snapshot_dir / "voice.webm",
323 "notes_transcript": None,
324 }
326 copied_path = snapshot_dir / source_path.name
327 if source_path.resolve() != copied_path.resolve():
328 shutil.copy2(source_path, copied_path)
330 session = {
331 "url": "unknown",
332 "started_at": "unknown",
333 "duration_seconds": 0,
334 "source_file": str(source_path),
335 "source_kind": source_kind,
336 }
338 if source_kind == "meeting_notes":
339 notes_transcript = read_notes(copied_path)
340 return {
341 "source_kind": source_kind,
342 "session": session,
343 "events": [],
344 "duration": 0.0,
345 "recording_path": None,
346 "transcription_path": None,
347 "notes_transcript": notes_transcript,
348 }
350 duration = ffprobe_duration(copied_path)
351 session["duration_seconds"] = round(duration, 3) if duration else 0
352 recording_path = copied_path if has_video_stream(copied_path) else None
353 transcription_path = copied_path if source_kind in {"video", "audio", "unknown"} else None
354 return {
355 "source_kind": source_kind,
356 "session": session,
357 "events": [],
358 "duration": duration,
359 "recording_path": recording_path,
360 "transcription_path": transcription_path,
361 "notes_transcript": None,
362 }
365def prepare_source(source_path: Path, raw_dir: Path, source_kind: str | None = None) -> dict[str, Any]:
366 source_kind = source_kind or classify_source(source_path)
367 raw_dir.parent.mkdir(parents=True, exist_ok=True)
368 staging_dir = Path(
369 tempfile.mkdtemp(prefix=f".{raw_dir.name}.staging-", dir=raw_dir.parent)
370 )
371 try:
372 source = populate_source_snapshot(source_path, staging_dir, source_kind)
373 promote_raw_snapshot(staging_dir, raw_dir)
374 except BaseException:
375 if staging_dir.exists() and not staging_dir.is_symlink():
376 shutil.rmtree(staging_dir, ignore_errors=True)
377 raise
379 for key in ("recording_path", "transcription_path"):
380 staged_path = source[key]
381 if staged_path is not None:
382 source[key] = raw_dir / staged_path.relative_to(staging_dir)
383 return source
386def repo_relative(path: Path, base: Path) -> str:
387 try:
388 return str(path.resolve().relative_to(base.resolve()))
389 except ValueError:
390 return str(path)
393def display_path(path: Path, repo_root: Path) -> str:
394 relative = repo_relative(path, repo_root)
395 return relative if not relative.startswith("/") else str(path)
398def format_time(seconds: float | int | None) -> str:
399 if seconds is None:
400 return "n/a"
401 seconds_float = float(seconds)
402 minutes = int(seconds_float // 60)
403 rest = seconds_float - minutes * 60
404 return f"{minutes:02d}:{rest:05.2f}"
407def event_time(event: dict[str, Any]) -> float:
408 try:
409 return float(event.get("t", 0))
410 except (TypeError, ValueError):
411 return 0.0
414def event_label(event: dict[str, Any]) -> str:
415 event_type = event.get("type", "event")
416 if event_type == "click":
417 element = event.get("element") or {}
418 text = compact_text(element.get("text") or "")
419 element_id = element.get("id")
420 tag = element.get("tag") or "element"
421 if element_id:
422 return f"click {tag}#{element_id} {text}".strip()
423 return f"click {tag} {text}".strip()
424 if event_type == "network_request":
425 return f"{event.get('method', 'GET')} {event.get('url', '')} -> {event.get('status')}"
426 return compact_text(json.dumps(event, sort_keys=True))
429def compact_text(text: str, limit: int = 120) -> str:
430 compacted = re.sub(r"\s+", " ", str(text)).strip()
431 if len(compacted) <= limit:
432 return compacted
433 return compacted[: limit - 1].rstrip() + "..."
436def network_is_noise(event: dict[str, Any]) -> bool:
437 url = str(event.get("url") or "")
438 return any(pattern in url for pattern in NOISY_NETWORK_PATTERNS)
441def transcript_has_complaint(transcript: str) -> bool:
442 lowered = transcript.lower()
443 return any(cue in lowered for cue in COMPLAINT_CUES)
446def transcribe_media(media_path: Path | None, model: str) -> dict[str, Any]:
447 if not media_path or not media_path.exists():
448 return {"status": "missing", "text": ""}
449 api_key = os.environ.get("OPENAI_API_KEY")
450 if not api_key:
451 return {
452 "status": "skipped",
453 "text": "",
454 "reason": "OPENAI_API_KEY is not set. Re-run with the key available to transcribe the media file.",
455 }
456 if not shutil.which("curl"):
457 return {"status": "skipped", "text": "", "reason": "curl is not installed"}
459 command = [
460 "curl",
461 "-sS",
462 "https://api.openai.com/v1/audio/transcriptions",
463 "-H",
464 f"Authorization: Bearer {api_key}",
465 "-F",
466 f"file=@{media_path}",
467 "-F",
468 f"model={model}",
469 "-F",
470 "response_format=json",
471 ]
472 try:
473 result = subprocess.run(command, capture_output=True, text=True, timeout=180)
474 except subprocess.TimeoutExpired:
475 return {"status": "failed", "text": "", "reason": "transcription request timed out"}
477 if result.returncode != 0:
478 return {
479 "status": "failed",
480 "text": "",
481 "reason": compact_text(result.stderr or result.stdout, 500),
482 }
484 try:
485 payload = json.loads(result.stdout)
486 except json.JSONDecodeError:
487 return {"status": "failed", "text": "", "reason": compact_text(result.stdout, 500)}
489 if "error" in payload:
490 return {"status": "failed", "text": "", "reason": compact_text(json.dumps(payload["error"]), 500)}
492 text = payload.get("text", "")
493 return {"status": "ok", "text": text, "raw": payload}
496def should_retry_transcription_in_chunks(transcript: dict[str, Any]) -> bool:
497 reason = str(transcript.get("reason") or "")
498 return transcript.get("status") == "failed" and (
499 "input_too_large" in reason or "too large" in reason.lower() or "maximum context" in reason.lower()
500 )
503def transcribe_media_chunks(
504 media_path: Path | None,
505 model: str,
506 chunks_dir: Path,
507 duration: float,
508 chunk_seconds: int = 420,
509) -> dict[str, Any]:
510 if not media_path or not media_path.exists():
511 return {"status": "missing", "text": ""}
512 if not shutil.which("ffmpeg"):
513 return {"status": "failed", "text": "", "reason": "ffmpeg is not installed; cannot chunk media"}
515 chunks_dir.mkdir(parents=True, exist_ok=True)
516 chunk_count = max(1, int((duration or chunk_seconds) // chunk_seconds) + (1 if duration % chunk_seconds else 0))
517 transcripts: list[str] = []
518 chunk_results: list[dict[str, Any]] = []
520 for index in range(chunk_count):
521 start = index * chunk_seconds
522 if duration and start >= duration:
523 break
524 chunk_path = chunks_dir / f"audio-chunk-{index + 1:03d}-{start}s.mp3"
525 extract_command = [
526 "ffmpeg",
527 "-y",
528 "-ss",
529 str(start),
530 "-t",
531 str(chunk_seconds),
532 "-i",
533 str(media_path),
534 "-vn",
535 "-ac",
536 "1",
537 "-ar",
538 "16000",
539 "-b:a",
540 "64k",
541 str(chunk_path),
542 ]
543 extract = subprocess.run(extract_command, capture_output=True, text=True, timeout=180)
544 if extract.returncode != 0 or not chunk_path.exists():
545 chunk_results.append(
546 {
547 "chunk": index + 1,
548 "start_seconds": start,
549 "status": "failed",
550 "reason": compact_text(extract.stderr or extract.stdout, 500),
551 }
552 )
553 continue
555 chunk_transcript = transcribe_media(chunk_path, model)
556 chunk_results.append(
557 {
558 "chunk": index + 1,
559 "start_seconds": start,
560 "path": str(chunk_path),
561 "status": chunk_transcript.get("status"),
562 "reason": chunk_transcript.get("reason"),
563 }
564 )
565 if chunk_transcript.get("text"):
566 transcripts.append(f"[{format_time(start)}]\n{chunk_transcript['text'].strip()}")
568 if transcripts:
569 return {
570 "status": "ok",
571 "text": "\n\n".join(transcripts),
572 "source": "chunked_media",
573 "chunk_seconds": chunk_seconds,
574 "chunks": chunk_results,
575 }
577 return {
578 "status": "failed",
579 "text": "",
580 "reason": "No chunks transcribed successfully",
581 "chunks": chunk_results,
582 }
585def select_moments(
586 events: list[dict[str, Any]],
587 transcript: str,
588 duration: float,
589 max_moments: int,
590) -> list[dict[str, Any]]:
591 candidates: list[dict[str, Any]] = []
592 clicks_by_target: defaultdict[str, list[dict[str, Any]]] = defaultdict(list)
593 has_complaint = transcript_has_complaint(transcript)
595 for event in events:
596 event_type = event.get("type")
597 t = event_time(event)
598 if event_type == "click":
599 element = event.get("element") or {}
600 key = element.get("id") or element.get("selector") or element.get("text") or "unknown"
601 clicks_by_target[str(key)].append(event)
602 reason = "click event"
603 if has_complaint and duration and t >= duration * 0.55:
604 reason = "late-session click near complaint transcript"
605 candidates.append({"t": t, "reason": reason, "events": [event]})
606 elif event_type == "network_request":
607 status = event.get("status")
608 try:
609 failed = int(status) >= 400
610 except (TypeError, ValueError):
611 failed = False
612 if failed and not network_is_noise(event):
613 candidates.append({"t": t, "reason": f"failed network request ({status})", "events": [event]})
614 elif event_type in {"console_error", "error", "exception"}:
615 candidates.append({"t": t, "reason": f"{event_type} event", "events": [event]})
617 for grouped in clicks_by_target.values():
618 if len(grouped) >= 2:
619 first = grouped[0]
620 last = grouped[-1]
621 if event_time(last) - event_time(first) <= 8:
622 candidates.append(
623 {
624 "t": event_time(last),
625 "reason": "repeated clicks on the same target",
626 "events": grouped[:4],
627 }
628 )
630 if has_complaint and duration and not candidates:
631 for fraction in (0.35, 0.55, 0.75, 0.9):
632 candidates.append({"t": max(0.0, duration * fraction), "reason": "representative frame for complaint transcript", "events": []})
634 if duration and not candidates:
635 fractions = (0.1, 0.3, 0.5, 0.7, 0.9)
636 for fraction in fractions[:max(1, min(max_moments, len(fractions)))]:
637 candidates.append({"t": max(0.0, duration * fraction), "reason": "representative video frame", "events": []})
639 candidates.sort(key=lambda item: (item["t"], item["reason"]))
640 deduped: list[dict[str, Any]] = []
641 for candidate in candidates:
642 if candidate["t"] < 0:
643 continue
644 if any(abs(candidate["t"] - existing["t"]) < 0.45 and candidate["reason"] == existing["reason"] for existing in deduped):
645 continue
646 deduped.append(candidate)
647 if len(deduped) >= max_moments:
648 break
650 for index, moment in enumerate(deduped, start=1):
651 moment["id"] = f"M{index}"
652 return deduped
655def extract_frames(recording_path: Path | None, frames_dir: Path, moments: list[dict[str, Any]]) -> None:
656 frames_dir.parent.mkdir(parents=True, exist_ok=True)
657 validate_frames_destination(frames_dir)
658 staging_dir = Path(
659 tempfile.mkdtemp(prefix=f".{frames_dir.name}.staging-", dir=frames_dir.parent)
660 )
661 try:
662 if not recording_path or not recording_path.exists():
663 for moment in moments:
664 moment["screenshot"] = None
665 moment["screenshot_status"] = "no video source"
666 elif not shutil.which("ffmpeg"):
667 for moment in moments:
668 moment["screenshot"] = None
669 moment["screenshot_status"] = "ffmpeg not installed"
670 else:
671 for moment in moments:
672 safe_reason = slugify(moment["reason"])[:48]
673 frame_path = staging_dir / f"{moment['id'].lower()}-{moment['t']:.2f}s-{safe_reason}.png"
674 command = [
675 "ffmpeg",
676 "-y",
677 "-ss",
678 f"{max(0.0, float(moment['t'])):.3f}",
679 "-i",
680 str(recording_path),
681 "-frames:v",
682 "1",
683 "-q:v",
684 "2",
685 str(frame_path),
686 ]
687 result = subprocess.run(command, capture_output=True, text=True, timeout=60)
688 if result.returncode == 0 and frame_path.exists():
689 moment["screenshot"] = str(frames_dir / frame_path.name)
690 moment["screenshot_status"] = "ok"
691 else:
692 moment["screenshot"] = None
693 moment["screenshot_status"] = compact_text(result.stderr or result.stdout, 300)
694 promote_frames_snapshot(staging_dir, frames_dir)
695 except BaseException:
696 if staging_dir.exists() and not staging_dir.is_symlink():
697 shutil.rmtree(staging_dir, ignore_errors=True)
698 raise
701def event_counts(events: list[dict[str, Any]]) -> dict[str, int]:
702 return dict(Counter(str(event.get("type", "unknown")) for event in events))
705def summarize_candidate_findings(moments: list[dict[str, Any]], transcript: str) -> list[dict[str, Any]]:
706 findings: list[dict[str, Any]] = []
707 complaint_moments = [moment for moment in moments if "complaint" in moment.get("reason", "")]
708 repeated_clicks = [moment for moment in moments if "repeated clicks" in moment.get("reason", "")]
709 failed_requests = [moment for moment in moments if "failed network" in moment.get("reason", "")]
711 if transcript_has_complaint(transcript):
712 evidence_ids = [moment["id"] for moment in complaint_moments] or [moment["id"] for moment in moments[-3:]]
713 findings.append(
714 {
715 "id": "F1",
716 "title": "User reported a control that felt weird or unclickable",
717 "severity": "P2",
718 "observed": "Transcript or notes contain a complaint cue. Review linked moments when available and use the text to identify the affected product behavior.",
719 "expected": "The affected product behavior should either work as presented or clearly explain why it is unavailable.",
720 "evidence": evidence_ids,
721 "confidence": "Medium until screenshots are reviewed",
722 }
723 )
725 if repeated_clicks:
726 findings.append(
727 {
728 "id": f"F{len(findings) + 1}",
729 "title": "Repeated interaction may indicate missing feedback or a dead control",
730 "severity": "P2",
731 "observed": "The same target was clicked more than once within a short interval.",
732 "expected": "Repeated clicks should not be needed; the UI should respond once or show a clear disabled/error state.",
733 "evidence": [moment["id"] for moment in repeated_clicks],
734 "confidence": "Medium",
735 }
736 )
738 if failed_requests:
739 findings.append(
740 {
741 "id": f"F{len(findings) + 1}",
742 "title": "User-visible flow coincided with failed network requests",
743 "severity": "P2",
744 "observed": "One or more non-noisy network requests returned a failure status.",
745 "expected": "Failures should be handled with durable user feedback and recoverable behavior.",
746 "evidence": [moment["id"] for moment in failed_requests],
747 "confidence": "High for request failure, medium for user impact until screenshots are reviewed",
748 }
749 )
751 if not findings:
752 findings.append(
753 {
754 "id": "F1",
755 "title": "No obvious failure detected automatically",
756 "severity": "P3",
757 "observed": "The analyzer did not find complaint cues, repeated clicks, console errors, or non-noisy failed requests.",
758 "expected": "A human should still inspect the source evidence before closing the feedback.",
759 "evidence": [moment["id"] for moment in moments[:3]],
760 "confidence": "Low",
761 }
762 )
763 return findings
766def markdown_link(path: str | None, output_dir: Path, repo_root: Path) -> str:
767 if not path:
768 return "n/a"
769 path_obj = Path(path)
770 if path_obj.exists():
771 return repo_relative(path_obj, repo_root)
772 return path
775def write_analysis_md(
776 output_path: Path,
777 source_path: Path,
778 source_kind: str,
779 session: dict[str, Any],
780 events: list[dict[str, Any]],
781 transcript: dict[str, Any],
782 moments: list[dict[str, Any]],
783 findings: list[dict[str, Any]],
784 repo_root: Path,
785) -> None:
786 lines: list[str] = []
787 lines.append("# Product Feedback Analysis")
788 lines.append("")
789 lines.append("## Source")
790 lines.append("")
791 lines.append(f"- Source: `{source_path}`")
792 lines.append(f"- Source kind: `{source_kind}`")
793 lines.append(f"- URL: `{session.get('url', 'unknown')}`")
794 lines.append(f"- Started: `{session.get('started_at', 'unknown')}`")
795 lines.append(f"- Duration: `{session.get('duration_seconds', 'unknown')}` seconds")
796 lines.append(f"- Browser: `{session.get('browser', 'unknown')}`")
797 lines.append(f"- Event counts: `{event_counts(events)}`")
798 lines.append("")
799 lines.append("## Transcript")
800 lines.append("")
801 if transcript.get("text"):
802 lines.append(transcript["text"].strip())
803 else:
804 lines.append(f"_Transcript unavailable: {transcript.get('reason') or transcript.get('status', 'unknown')}._")
805 lines.append("")
806 lines.append("## Selected Moments")
807 lines.append("")
808 if moments:
809 lines.append("| ID | Time | Why selected | Screenshot | Event evidence |")
810 lines.append("|---|---:|---|---|---|")
811 for moment in moments:
812 screenshot = markdown_link(moment.get("screenshot"), output_path.parent, repo_root)
813 evidence = "<br>".join(compact_text(event_label(event), 140) for event in moment.get("events", [])) or "n/a"
814 lines.append(
815 f"| {moment['id']} | {format_time(moment['t'])} | {moment['reason']} | `{screenshot}` | {evidence} |"
816 )
817 else:
818 lines.append("_No video moments available for this source._")
819 lines.append("")
820 lines.append("## Candidate Findings")
821 lines.append("")
822 for finding in findings:
823 lines.append(f"### {finding['id']}. {finding['title']}")
824 lines.append("")
825 lines.append(f"- **Severity:** {finding['severity']}")
826 lines.append(f"- **Observed:** {finding['observed']}")
827 lines.append(f"- **Expected:** {finding['expected']}")
828 lines.append(f"- **Evidence:** {', '.join(finding['evidence'])}")
829 lines.append(f"- **Confidence:** {finding['confidence']}")
830 lines.append("")
831 lines.append("## Human Review Checklist")
832 lines.append("")
833 lines.append("- Open each selected screenshot and name the exact visible control or state.")
834 lines.append("- Tie transcript language to the closest click or visible UI state.")
835 lines.append("- Promote only confirmed product problems into requirements.")
836 lines.append("- Use repo-relative screenshot paths when moving evidence into a CE requirements document.")
837 output_path.write_text("\n".join(lines) + "\n")
840def write_requirements_kickoff(
841 output_path: Path,
842 topic: str,
843 session: dict[str, Any],
844 findings: list[dict[str, Any]],
845 moments: list[dict[str, Any]],
846 repo_root: Path,
847) -> None:
848 title = topic.replace("-", " ").title()
849 date = datetime.now(timezone.utc).date().isoformat()
850 primary_evidence = ", ".join(finding["id"] for finding in findings)
851 screenshot_refs = []
852 for moment in moments:
853 if moment.get("screenshot"):
854 screenshot_refs.append(f"{moment['id']}: `{markdown_link(moment['screenshot'], output_path.parent, repo_root)}`")
855 evidence_text = "; ".join(screenshot_refs[:6]) or "See analysis.md selected moments."
856 source_materials = markdown_link(str(output_path.parent / "source-materials.md"), output_path.parent, repo_root)
857 analysis_path = markdown_link(str(output_path.parent / "analysis.md"), output_path.parent, repo_root)
858 problem_analysis_path = markdown_link(str(output_path.parent / "problem-analysis.md"), output_path.parent, repo_root)
859 review_prompt_path = markdown_link(str(output_path.parent / "review-prompt.md"), output_path.parent, repo_root)
861 lines = [
862 "---",
863 f"date: {date}",
864 f"topic: {topic}",
865 "---",
866 "",
867 f"# {title}",
868 "",
869 "## Problem Frame",
870 "",
871 f"A product feedback source for `{session.get('url', 'the product surface')}` produced evidence of product friction. The raw source has been converted into transcript, selected moments when video is available, screenshots when frames can be extracted, and candidate findings so the team can decide what product behavior should change before planning implementation.",
872 "",
873 "Source materials for brainstorm:",
874 f"- Source materials manifest: `{source_materials}`",
875 f"- Analysis: `{analysis_path}`",
876 f"- Problem analysis: `{problem_analysis_path}`",
877 f"- Review prompt with transcript and frames: `{review_prompt_path}`",
878 "",
879 "---",
880 "",
881 "## Actors",
882 "",
883 "- A1. User: Operates the product in the recorded session and verbalizes friction.",
884 "- A2. Product surface: The UI and backend behavior visible in the recording.",
885 "- A3. Brainstorm agent: Uses the evidence bundle to confirm, correct, and group requirements before planning.",
886 "",
887 "---",
888 "",
889 "## Key Flows",
890 "",
891 "- F1. Evidence-backed feedback triage",
892 " - **Trigger:** A feedback bundle, video, audio file, or meeting notes file is available.",
893 " - **Actors:** A1, A2, A3",
894 " - **Steps:** Extract or copy the source, transcribe media or read notes, select high-signal moments when video exists, inspect screenshots when available, confirm problems, and write requirements with supporting evidence.",
895 " - **Outcome:** Confirmed product problems are represented as requirements with transcript support and screenshot support when visual evidence exists.",
896 " - **Covered by:** R1, R2, R3",
897 "",
898 "---",
899 "",
900 "## Requirements",
901 "",
902 "**Evidence handling**",
903 "- R1. Each confirmed product problem must cite supporting transcript, notes, or moment evidence from the source, including timestamp and screenshot when video is available.",
904 "- R2. Transcript claims must be tied to the closest visible interaction or explicitly marked as untimed verbal context.",
905 "",
906 "**Product requirements from this session**",
907 ]
909 for index, finding in enumerate(findings, start=3):
910 lines.append(f"- R{index}. Resolve or intentionally scope the issue described by {finding['id']}: {finding['title']}.")
912 lines.extend(
913 [
914 "",
915 "---",
916 "",
917 "## Acceptance Examples",
918 "",
919 "- AE1. **Covers R1, R2.** Given a feedback source with voice, video, or notes, when the analysis is complete, each promoted issue includes source evidence rather than prose-only claims.",
920 "- AE2. **Covers R3.** Given the user reports that a button is weird or unclickable, when requirements are finalized, the requirement identifies the specific control and the expected available/unavailable behavior.",
921 "",
922 "---",
923 "",
924 "## Success Criteria",
925 "",
926 "- A human reviewer can understand what went wrong without rewatching the entire recording.",
927 "- `ce-brainstorm` can confirm requirements from linked source evidence before any planning begins.",
928 "",
929 "---",
930 "",
931 "## Scope Boundaries",
932 "",
933 "- The analyzer output is evidence and requirements kickoff material, not final implementation design.",
934 "- Automatically detected findings remain candidates until screenshots are inspected.",
935 "- Development-only noise, such as profiler requests, should not become product requirements unless it affects the user experience.",
936 "",
937 "---",
938 "",
939 "## Key Decisions",
940 "",
941 "- Evidence first: Requirements should cite moments and screenshots before moving to planning.",
942 "- Brainstorm before plan: Use `ce-brainstorm` to refine product behavior when the recording reveals ambiguity.",
943 "",
944 "---",
945 "",
946 "## Dependencies / Assumptions",
947 "",
948 f"- Source session URL: `{session.get('url', 'unknown')}`.",
949 f"- Source materials manifest: `{source_materials}`.",
950 f"- Candidate findings: {primary_evidence}.",
951 f"- Screenshot evidence: {evidence_text}.",
952 "",
953 "---",
954 "",
955 "## Outstanding Questions",
956 "",
957 "### Resolve Before Planning",
958 "",
959 "- Which candidate findings are real product problems after screenshot review?",
960 "- For each promoted finding, what should the user experience be instead?",
961 "",
962 "### Deferred to Planning",
963 "",
964 "- [Technical] Which code paths own the confirmed product behavior?",
965 "- [Technical] What regression tests should lock the behavior once fixed?",
966 "",
967 "---",
968 "",
969 "## Next Steps",
970 "",
971 "-> Resume `/ce-brainstorm` to confirm candidate findings and replace generic R-items with product-specific requirements.",
972 ]
973 )
974 output_path.write_text("\n".join(lines) + "\n")
977def write_source_materials(
978 output_path: Path,
979 source_path: Path,
980 source_kind: str,
981 session: dict[str, Any],
982 transcript: dict[str, Any],
983 moments: list[dict[str, Any]],
984 raw_dir: Path,
985 frames_dir: Path,
986 repo_root: Path,
987) -> None:
988 def link(path: Path) -> str:
989 return markdown_link(str(path), output_path.parent, repo_root)
991 raw_files = sorted(path for path in raw_dir.rglob("*") if path.is_file())
992 frame_files = sorted(path for path in frames_dir.rglob("*.png") if path.is_file())
993 chunk_files = sorted((raw_dir / "transcription_chunks").glob("*")) if (raw_dir / "transcription_chunks").is_dir() else []
995 copied_source = next((path for path in raw_files if path.name == source_path.name), None)
996 if not copied_source:
997 copied_source = raw_dir / "recording.webm" if (raw_dir / "recording.webm").exists() else None
999 lines = [
1000 "# Source Materials",
1001 "",
1002 "Use this manifest during brainstorm so requirements can be traced back to the raw feedback evidence.",
1003 "",
1004 "## Original Source",
1005 "",
1006 f"- Source kind: `{source_kind}`",
1007 f"- Original path: `{source_path}`",
1008 f"- Local raw copy: `{link(copied_source) if copied_source else 'n/a'}`",
1009 "- Commit policy: raw media, audio chunks, zip contents, session dumps, and extracted screenshots are local-only by default; commit generated Markdown/JSON/manifests when useful for brainstorm/planning traceability.",
1010 f"- Session URL: `{session.get('url', 'unknown')}`",
1011 f"- Duration: `{session.get('duration_seconds', 'unknown')}` seconds",
1012 "",
1013 "## Analysis Artifacts",
1014 "",
1015 f"- Analysis summary: `{link(output_path.parent / 'analysis.md')}`",
1016 f"- Problem statements: `{link(output_path.parent / 'problem-analysis.md')}`",
1017 f"- Review prompt: `{link(output_path.parent / 'review-prompt.md')}`",
1018 f"- Requirements kickoff: `{link(output_path.parent / 'requirements-kickoff.md')}`",
1019 f"- Structured JSON: `{link(output_path.parent / 'analysis.json')}`",
1020 "",
1021 "## Transcript",
1022 "",
1023 f"- Transcript status: `{transcript.get('status', 'unknown')}`",
1024 f"- Transcript source: `{transcript.get('source', source_kind)}`",
1025 f"- Transcript text lives in: `{link(output_path.parent / 'analysis.md')}` and `{link(output_path.parent / 'review-prompt.md')}`",
1026 ]
1028 if chunk_files:
1029 lines.append("- Transcription chunks:")
1030 lines.append(f" - retained locally in `{link(raw_dir / 'transcription_chunks')}`; not commit-safe by default.")
1032 lines.extend(["", "## Local-Only Frames", ""])
1033 lines.append("Extracted screenshots are retained locally for agent inspection and should not be committed by default.")
1034 lines.append("")
1035 if moments:
1036 lines.append("| Moment | Time | Screenshot | Why selected |")
1037 lines.append("|---|---:|---|---|")
1038 for moment in moments:
1039 screenshot = moment.get("screenshot")
1040 lines.append(
1041 f"| {moment['id']} | {format_time(moment['t'])} | `{markdown_link(screenshot, output_path.parent, repo_root)}` | {moment['reason']} |"
1042 )
1043 else:
1044 lines.append("_No video frames were available for this source._")
1046 if frame_files:
1047 lines.extend(["", "All frame files:"])
1048 for frame in frame_files:
1049 lines.append(f"- `{link(frame)}`")
1051 lines.extend(["", "## Local Raw Files", ""])
1052 lines.append("Raw files are intentionally local-only by default. Do not commit these unless the user explicitly asks and privacy/security is acceptable.")
1053 lines.append("")
1054 for raw_file in raw_files[:50]:
1055 lines.append(f"- `{link(raw_file)}`")
1056 if len(raw_files) > 50:
1057 lines.append(f"- ... {len(raw_files) - 50} more files")
1059 output_path.write_text("\n".join(lines) + "\n")
1062def write_problem_analysis(
1063 output_path: Path,
1064 transcript: dict[str, Any],
1065 moments: list[dict[str, Any]],
1066 findings: list[dict[str, Any]],
1067 repo_root: Path,
1068) -> None:
1069 complaint_text = transcript.get("text") or ""
1070 lines = [
1071 "<analysis>",
1072 "## 1. Visual/UI Problems",
1073 "",
1074 ]
1075 if moments:
1076 lines.extend(
1077 [
1078 "1. Review required: inspect the extracted frames and replace this scaffold with precise visual observations. Include location, UI element type, issue description, and frame reference.",
1079 "",
1080 ]
1081 )
1082 else:
1083 lines.extend(["1. No video frames were available for this source.", ""])
1085 lines.extend(["## 2. Functional Problems", ""])
1087 for index, finding in enumerate(findings, start=1):
1088 evidence = ", ".join(finding.get("evidence", [])) or "n/a"
1089 lines.append(
1090 f"{index}. {finding['title']}: {finding['observed']} Evidence: {evidence}. Context from discussion: {compact_text(complaint_text, 220) or 'n/a'}"
1091 )
1092 if not findings:
1093 lines.append("1. No functional problems were detected automatically; inspect transcript and frames manually.")
1095 lines.extend(
1096 [
1097 "",
1098 "## 3. Requirements",
1099 "",
1100 "1. Convert confirmed problems into requirements after evidence review. State what capability or behavior is needed and why, without prescribing implementation.",
1101 "",
1102 "## 4. Usability/UX Problems",
1103 "",
1104 ]
1105 )
1107 for index, moment in enumerate(moments, start=1):
1108 screenshot = markdown_link(moment.get("screenshot"), output_path.parent, repo_root)
1109 lines.append(
1110 f"{index}. Moment {moment['id']} at {format_time(moment['t'])}: Review `{screenshot}` for UX friction related to `{moment['reason']}`."
1111 )
1112 if not moments:
1113 lines.append("1. Review the transcript or notes for workflow friction, confusion, and unmet expectations.")
1115 lines.append("</analysis>")
1116 output_path.write_text("\n".join(lines) + "\n")
1119def write_review_prompt(
1120 output_path: Path,
1121 transcript: dict[str, Any],
1122 moments: list[dict[str, Any]],
1123 repo_root: Path,
1124) -> None:
1125 frame_lines: list[str] = []
1126 for moment in moments:
1127 screenshot = markdown_link(moment.get("screenshot"), output_path.parent, repo_root)
1128 event_summary = "; ".join(event_label(event) for event in moment.get("events", [])) or "no event metadata"
1129 frame_lines.append(
1130 f"- {moment['id']} ({format_time(moment['t'])}, {moment['reason']}): `{screenshot}`. Events: {event_summary}"
1131 )
1132 if not frame_lines:
1133 frame_lines.append("- No video frames are available for this source. Analyze transcript or meeting notes only.")
1135 lines = [
1136 "You will be analyzing a product feedback session by examining video frames and a discussion transcript. Your goal is to identify problems, requirements, and feedback points that need to be addressed - focusing on clear problem statements rather than solutions.",
1137 "",
1138 "Here are the frames extracted from the video:",
1139 "",
1140 "<video_frames>",
1141 *frame_lines,
1142 "</video_frames>",
1143 "",
1144 "Here is the transcript of the discussion that occurred during the feedback session:",
1145 "",
1146 "<discussion_transcript>",
1147 transcript.get("text") or f"[Transcript unavailable: {transcript.get('reason') or transcript.get('status', 'unknown')}]",
1148 "</discussion_transcript>",
1149 "",
1150 "Your task is to carefully analyze both the visual content and the discussion to extract actionable problem statements. Follow these guidelines:",
1151 "",
1152 "**Visual Analysis Requirements:**",
1153 "- Examine each frame carefully for UI/UX issues, bugs, design inconsistencies, or usability problems",
1154 "- Be extremely precise about what you observe: specify exact locations (e.g., \"top-right corner,\" \"navigation bar,\" \"third item in the list\")",
1155 "- Identify specific UI elements by type (button, input field, dropdown, modal, etc.)",
1156 "- Note visual problems like misalignment, poor contrast, truncated text, overlapping elements, broken layouts, etc.",
1157 "",
1158 "**Discussion Analysis Requirements:**",
1159 "- Extract feedback points, feature requests, and problems mentioned in the conversation",
1160 "- Identify requirements that are stated or implied",
1161 "- Note any pain points or frustrations expressed by participants",
1162 "- Connect visual observations with relevant discussion points when applicable",
1163 "",
1164 "**Problem Statement Guidelines:**",
1165 "- Focus on describing WHAT the problem is, not HOW to fix it",
1166 "- Be specific and actionable - avoid vague statements",
1167 "- Each problem should be clear enough that a developer or designer can understand what needs to be addressed",
1168 "- Include context about where the problem occurs and why it matters",
1169 "",
1170 "Structure your final output as follows:",
1171 "",
1172 "1. **Visual/UI Problems**: Issues observed directly in the interface",
1173 "2. **Functional Problems**: Issues related to behavior, workflow, or functionality mentioned in discussion",
1174 "3. **Requirements**: New features or capabilities requested",
1175 "4. **Usability/UX Problems**: Issues related to user experience, confusion, or workflow friction",
1176 "",
1177 "Format each problem as a clear, numbered item within its category.",
1178 "",
1179 "Your final output should contain only the analysis section with clearly categorized, numbered problem statements. Do not include scratchpad notes.",
1180 ]
1181 output_path.write_text("\n".join(lines) + "\n")
1184def main() -> int:
1185 args = parse_args()
1186 source_path = args.source_path.expanduser().resolve()
1187 if not source_path.exists():
1188 print(f"Source file not found: {source_path}", file=sys.stderr)
1189 return 1
1191 try:
1192 source_kind = classify_source(source_path)
1193 except SourceInputError as exc:
1194 print(f"Invalid source: {exc}", file=sys.stderr)
1195 return 2
1197 output_dir = (args.output_dir or default_output_dir(source_path)).expanduser().resolve()
1198 if source_path == output_dir or source_path.is_relative_to(output_dir):
1199 print(
1200 f"Invalid output directory: source {source_path} must be outside the analyzer output directory {output_dir}.",
1201 file=sys.stderr,
1202 )
1203 return 2
1204 if source_path.is_dir() and (output_dir == source_path or output_dir.is_relative_to(source_path)):
1205 print(
1206 f"Invalid output directory: {output_dir} must be outside the unpacked capture directory {source_path}.",
1207 file=sys.stderr,
1208 )
1209 return 2
1210 output_dir.mkdir(parents=True, exist_ok=True)
1211 raw_dir = output_dir / "raw"
1212 frames_dir = output_dir / "frames"
1213 try:
1214 validate_frames_destination(frames_dir)
1215 except SourceInputError as exc:
1216 print(f"Invalid output: {exc}", file=sys.stderr)
1217 return 2
1218 try:
1219 source = prepare_source(source_path, raw_dir, source_kind)
1220 except SourceInputError as exc:
1221 print(f"Invalid source: {exc}", file=sys.stderr)
1222 return 2
1223 source_kind = source["source_kind"]
1224 session = source["session"]
1225 events = source["events"]
1226 duration = source["duration"]
1228 if source["notes_transcript"]:
1229 transcript = source["notes_transcript"]
1230 elif args.no_transcribe:
1231 transcript = {"status": "skipped", "text": "", "reason": "--no-transcribe was passed"}
1232 else:
1233 transcript = transcribe_media(source["transcription_path"], args.model)
1234 if should_retry_transcription_in_chunks(transcript):
1235 transcript = transcribe_media_chunks(
1236 source["transcription_path"],
1237 args.model,
1238 raw_dir / "transcription_chunks",
1239 duration,
1240 )
1242 moments = select_moments(events, transcript.get("text", ""), duration, args.max_moments)
1243 if not moments and source["recording_path"]:
1244 fallback_times = [0.5, 2.0, 5.0, 10.0, 15.0]
1245 moments = [
1246 {"id": f"M{index}", "t": timestamp, "reason": "representative video frame", "events": []}
1247 for index, timestamp in enumerate(fallback_times[: args.max_moments], start=1)
1248 ]
1249 try:
1250 extract_frames(source["recording_path"], frames_dir, moments)
1251 except SourceInputError as exc:
1252 print(f"Invalid output: {exc}", file=sys.stderr)
1253 return 2
1254 findings = summarize_candidate_findings(moments, transcript.get("text", ""))
1256 topic = slugify(args.topic or source_path.stem)
1257 repo_root = Path.cwd()
1258 analysis_md = output_dir / "analysis.md"
1259 problem_analysis_md = output_dir / "problem-analysis.md"
1260 review_prompt_md = output_dir / "review-prompt.md"
1261 source_materials_md = output_dir / "source-materials.md"
1262 kickoff_md = output_dir / "requirements-kickoff.md"
1263 write_analysis_md(analysis_md, source_path, source_kind, session, events, transcript, moments, findings, repo_root)
1264 write_problem_analysis(problem_analysis_md, transcript, moments, findings, repo_root)
1265 write_review_prompt(review_prompt_md, transcript, moments, repo_root)
1266 write_source_materials(source_materials_md, source_path, source_kind, session, transcript, moments, raw_dir, frames_dir, repo_root)
1267 write_requirements_kickoff(kickoff_md, topic, session, findings, moments, repo_root)
1269 structured = {
1270 "source": str(source_path),
1271 "source_kind": source_kind,
1272 "output_dir": str(output_dir),
1273 "session": session,
1274 "event_counts": event_counts(events),
1275 "transcript": transcript,
1276 "moments": moments,
1277 "candidate_findings": findings,
1278 "artifacts": {
1279 "analysis_md": str(analysis_md),
1280 "problem_analysis_md": str(problem_analysis_md),
1281 "review_prompt_md": str(review_prompt_md),
1282 "source_materials_md": str(source_materials_md),
1283 "requirements_kickoff_md": str(kickoff_md),
1284 "frames_dir": str(frames_dir),
1285 "raw_dir": str(raw_dir),
1286 },
1287 }
1288 (output_dir / "analysis.json").write_text(json.dumps(structured, indent=2, sort_keys=True) + "\n")
1290 print(f"Analysis written to: {analysis_md}")
1291 print(f"Problem analysis scaffold written to: {problem_analysis_md}")
1292 print(f"Review prompt written to: {review_prompt_md}")
1293 print(f"Source materials manifest written to: {source_materials_md}")
1294 print(f"Requirements kickoff written to: {kickoff_md}")
1295 print(f"Frames written to: {frames_dir}")
1296 print("")
1297 print("Analysis complete. Ready to brainstorm the findings.")
1298 print(f"Source materials: {display_path(source_materials_md, repo_root)}")
1299 print(f"Problem statements: {display_path(problem_analysis_md, repo_root)}")
1300 print(f"Brainstorm handoff: $compound-engineering:ce-brainstorm {display_path(kickoff_md, repo_root)}")
1301 print("Brainstorm should first confirm whether the captured requirements are complete and correctly grouped, then write the durable unified plan under the plans artifact directory.")
1302 return 0
1305if __name__ == "__main__":
1306 raise SystemExit(main())