Coverage for little_loops / workflow_sequence / analysis.py: 0%

340 statements  

« prev     ^ index     » next       coverage.py v7.12.0, created at 2026-05-28 13:07 -0500

1"""Analysis functions for workflow sequence detection.""" 

2 

3from __future__ import annotations 

4 

5import json 

6import re 

7import sys 

8from datetime import datetime 

9from pathlib import Path 

10from typing import Any 

11 

12import yaml 

13 

14from little_loops.workflow_sequence.io import ( 

15 _load_messages, 

16 _load_messages_from_db, 

17 _load_patterns, 

18) 

19from little_loops.workflow_sequence.models import ( 

20 EntityCluster, 

21 SessionLink, 

22 Workflow, 

23 WorkflowAnalysis, 

24 WorkflowBoundary, 

25) 

26 

27# Module-level compiled regex patterns 

28FILE_PATTERN = re.compile(r"[\w./-]+\.(?:md|py|json|yaml|yml|js|ts|tsx|jsx|sh|toml)", re.IGNORECASE) 

29PHASE_PATTERN = re.compile(r"phase[- ]?\d+", re.IGNORECASE) 

30MODULE_PATTERN = re.compile(r"module[- ]?\d+", re.IGNORECASE) 

31COMMAND_PATTERN = re.compile(r"/[\w:-]+") 

32ISSUE_PATTERN = re.compile(r"(?:BUG|FEAT|ENH|EPIC)-\d+", re.IGNORECASE) 

33 

34# Verb class taxonomy for semantic similarity 

35VERB_CLASSES: dict[str, set[str]] = { 

36 "deletion": {"remove", "delete", "drop", "eliminate", "clear", "clean"}, 

37 "modification": {"update", "change", "modify", "edit", "fix", "adjust", "revise"}, 

38 "creation": {"create", "add", "generate", "write", "make", "build"}, 

39 "search": {"find", "search", "locate", "where", "what", "which", "list"}, 

40 "verification": {"check", "verify", "validate", "confirm", "review", "ensure"}, 

41 "execution": {"run", "execute", "launch", "start", "invoke", "call"}, 

42} 

43 

44# Workflow templates: category sequences that indicate common patterns 

45WORKFLOW_TEMPLATES: dict[str, list[str]] = { 

46 "explore → modify → verify": ["file_search", "code_modification", "testing"], 

47 "create → refine → finalize": ["file_write", "code_modification", "git_operation"], 

48 "review → fix → commit": ["code_review", "code_modification", "git_operation"], 

49 "plan → implement → verify": ["planning", "code_modification", "testing"], 

50 "debug → fix → test": ["debugging", "code_modification", "testing"], 

51} 

52 

53# Maps content keywords to workflow category labels used by WORKFLOW_TEMPLATES 

54_CONTENT_CATEGORY_MAP: dict[str, list[str]] = { 

55 "file_search": ["search", "find", "glob", "grep", "locate"], 

56 "code_modification": ["edit", "write", "fix", "refactor", "update", "implement"], 

57 "testing": ["test", "pytest", "assert", "verify", "check"], 

58 "git_operation": ["commit", "push", "branch", "pr", "merge", "pull"], 

59 "planning": ["plan", "design", "architect", "outline", "draft"], 

60 "debugging": ["debug", "trace", "breakpoint", "error", "exception", "bug"], 

61 "code_review": ["review", "inspect", "audit", "read", "examine"], 

62 "file_write": ["create", "generate", "scaffold", "write", "add"], 

63} 

64 

65 

66# ----------------------------------------------------------------------------- 

67# Core Analysis Functions 

68# ----------------------------------------------------------------------------- 

69 

70 

71def extract_entities(content: str) -> set[str]: 

72 """Extract file paths, commands, and concepts from message content. 

73 

74 Args: 

75 content: Message text content 

76 

77 Returns: 

78 Set of extracted entities (file paths, commands, issue IDs, etc.) 

79 """ 

80 entities: set[str] = set() 

81 

82 # File paths 

83 entities.update(FILE_PATTERN.findall(content)) 

84 

85 # Phase/module references 

86 entities.update(PHASE_PATTERN.findall(content.lower())) 

87 entities.update(MODULE_PATTERN.findall(content.lower())) 

88 

89 # Slash commands 

90 entities.update(COMMAND_PATTERN.findall(content)) 

91 

92 # Issue IDs (normalize to uppercase) 

93 entities.update(match.upper() for match in ISSUE_PATTERN.findall(content)) 

94 

95 return entities 

96 

97 

98def calculate_boundary_weight(gap_seconds: int) -> float: 

99 """Calculate workflow boundary weight based on time gap. 

100 

101 Args: 

102 gap_seconds: Time gap between messages in seconds 

103 

104 Returns: 

105 Boundary weight from 0.0 to 0.95 

106 """ 

107 if gap_seconds < 30: 

108 return 0.0 

109 elif gap_seconds < 120: 

110 return 0.1 

111 elif gap_seconds < 300: 

112 return 0.3 

113 elif gap_seconds < 900: 

114 return 0.5 

115 elif gap_seconds < 1800: 

116 return 0.7 

117 elif gap_seconds < 7200: 

118 return 0.85 

119 else: 

120 return 0.95 

121 

122 

123def entity_overlap(entities_a: set[str], entities_b: set[str]) -> float: 

124 """Calculate Jaccard similarity between two entity sets. 

125 

126 Args: 

127 entities_a: First set of entities 

128 entities_b: Second set of entities 

129 

130 Returns: 

131 Jaccard similarity coefficient (0.0 to 1.0) 

132 """ 

133 if not entities_a or not entities_b: 

134 return 0.0 

135 intersection = len(entities_a & entities_b) 

136 union = len(entities_a | entities_b) 

137 return intersection / union if union > 0 else 0.0 

138 

139 

140def get_verb_class(content: str) -> str | None: 

141 """Extract verb class from message content. 

142 

143 Args: 

144 content: Message text content 

145 

146 Returns: 

147 Verb class name or None if no match 

148 """ 

149 content_lower = content.lower() 

150 words = set(re.findall(r"\b\w+\b", content_lower)) 

151 

152 for verb_class, verbs in VERB_CLASSES.items(): 

153 if words & verbs: 

154 return verb_class 

155 return None 

156 

157 

158def semantic_similarity( 

159 content_a: str, 

160 content_b: str, 

161 entities_a: set[str], 

162 entities_b: set[str], 

163 category_a: str | None, 

164 category_b: str | None, 

165) -> float: 

166 """Calculate semantic similarity between two messages. 

167 

168 Uses weighted combination of: 

169 - Keyword overlap (0.3) 

170 - Verb class match (0.3) 

171 - Entity overlap (0.3) 

172 - Category match (0.1) 

173 

174 Args: 

175 content_a: First message content 

176 content_b: Second message content 

177 entities_a: Entities from first message 

178 entities_b: Entities from second message 

179 category_a: Category of first message 

180 category_b: Category of second message 

181 

182 Returns: 

183 Similarity score (0.0 to 1.0) 

184 """ 

185 # Keyword overlap (simple word-level Jaccard) 

186 words_a = set(re.findall(r"\b[a-z]{3,}\b", content_a.lower())) 

187 words_b = set(re.findall(r"\b[a-z]{3,}\b", content_b.lower())) 

188 keyword_sim = len(words_a & words_b) / len(words_a | words_b) if words_a | words_b else 0.0 

189 

190 # Verb class similarity 

191 verb_a = get_verb_class(content_a) 

192 verb_b = get_verb_class(content_b) 

193 verb_sim = 1.0 if verb_a and verb_a == verb_b else 0.0 

194 

195 # Entity overlap 

196 entity_sim = entity_overlap(entities_a, entities_b) 

197 

198 # Category match 

199 category_sim = 1.0 if category_a and category_a == category_b else 0.0 

200 

201 # Weighted combination 

202 return keyword_sim * 0.3 + verb_sim * 0.3 + entity_sim * 0.3 + category_sim * 0.1 

203 

204 

205# ----------------------------------------------------------------------------- 

206# Internal Analysis Functions 

207# ----------------------------------------------------------------------------- 

208 

209 

210def _detect_handoff(content: str) -> bool: 

211 """Check if message indicates a session handoff.""" 

212 handoff_markers = [ 

213 "/ll:handoff", 

214 "continue in new session", 

215 "pick up in next session", 

216 "resuming from", 

217 "continuation of", 

218 ] 

219 content_lower = content.lower() 

220 return any(marker in content_lower for marker in handoff_markers) 

221 

222 

223def _parse_timestamps(messages: list[dict[str, Any]]) -> list[datetime]: 

224 """Parse valid ISO timestamps from a list of messages, stripping timezone info.""" 

225 timestamps = [] 

226 for msg in messages: 

227 ts_str = msg.get("timestamp", "") 

228 if ts_str: 

229 try: 

230 ts = datetime.fromisoformat(ts_str.replace("Z", "+00:00")) 

231 if ts.tzinfo is not None: 

232 ts = ts.replace(tzinfo=None) 

233 timestamps.append(ts) 

234 except (ValueError, AttributeError, TypeError): 

235 pass 

236 return timestamps 

237 

238 

239def _group_by_session(messages: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]: 

240 """Group messages by session_id.""" 

241 sessions: dict[str, list[dict[str, Any]]] = {} 

242 for msg in messages: 

243 session_id = msg.get("session_id", "unknown") 

244 if session_id not in sessions: 

245 sessions[session_id] = [] 

246 sessions[session_id].append(msg) 

247 return sessions 

248 

249 

250def _link_sessions(sessions: dict[str, list[dict[str, Any]]]) -> list[SessionLink]: 

251 """Identify sessions that are part of the same workflow.""" 

252 links: list[SessionLink] = [] 

253 session_ids = list(sessions.keys()) 

254 link_counter = 0 

255 

256 for i, session_a_id in enumerate(session_ids): 

257 session_a = sessions[session_a_id] 

258 if not session_a: 

259 continue 

260 

261 # Extract session metadata 

262 last_msg_a = session_a[-1] if session_a else {} 

263 entities_a: set[str] = set() 

264 for msg in session_a: 

265 entities_a.update(extract_entities(msg.get("content", ""))) 

266 branch_a = last_msg_a.get("git_branch") 

267 

268 for session_b_id in session_ids[i + 1 :]: 

269 session_b = sessions[session_b_id] 

270 if not session_b: 

271 continue 

272 

273 first_msg_b = session_b[0] if session_b else {} 

274 entities_b: set[str] = set() 

275 for msg in session_b: 

276 entities_b.update(extract_entities(msg.get("content", ""))) 

277 branch_b = first_msg_b.get("git_branch") 

278 

279 # Calculate link score 

280 score = 0.0 

281 evidence: list[str] = [] 

282 

283 # Same git branch (HIGH weight) 

284 if branch_a and branch_a == branch_b: 

285 score += 0.4 

286 evidence.append("shared_branch") 

287 

288 # Explicit handoff marker (HIGH weight) 

289 if any(_detect_handoff(msg.get("content", "")) for msg in session_a): 

290 score += 0.4 

291 evidence.append("handoff_detected") 

292 

293 # Shared entities (MEDIUM weight) 

294 overlap = entity_overlap(entities_a, entities_b) 

295 if overlap > 0.5: 

296 score += 0.2 

297 evidence.append("entity_overlap") 

298 elif overlap > 0.3: 

299 score += 0.1 

300 evidence.append("partial_entity_overlap") 

301 

302 if score > 0.3: 

303 link_counter += 1 

304 

305 # Calculate span 

306 timestamps = _parse_timestamps(session_a + session_b) 

307 

308 span_hours = 0.0 

309 if len(timestamps) >= 2: 

310 try: 

311 span_hours = (max(timestamps) - min(timestamps)).total_seconds() / 3600 

312 except TypeError: 

313 span_hours = 0.0 

314 

315 links.append( 

316 SessionLink( 

317 link_id=f"link-{link_counter:03d}", 

318 sessions=[ 

319 { 

320 "session_id": session_a_id, 

321 "position": 1, 

322 "link_evidence": evidence[0] if evidence else "score", 

323 }, 

324 { 

325 "session_id": session_b_id, 

326 "position": 2, 

327 "link_evidence": evidence[-1] if evidence else "score", 

328 }, 

329 ], 

330 unified_workflow={ 

331 "name": f"Linked workflow {link_counter}", 

332 "total_messages": len(session_a) + len(session_b), 

333 "span_hours": round(span_hours, 1), 

334 "evidence": evidence, 

335 }, 

336 confidence=min(score, 1.0), 

337 ) 

338 ) 

339 

340 return links 

341 

342 

343def _cluster_by_entities( 

344 messages: list[dict[str, Any]], overlap_threshold: float = 0.3 

345) -> list[EntityCluster]: 

346 """Cluster messages with significant entity overlap.""" 

347 clusters: list[EntityCluster] = [] 

348 cluster_counter = 0 

349 

350 for msg in messages: 

351 content = msg.get("content", "") 

352 msg_entities = extract_entities(content) 

353 

354 if not msg_entities: 

355 continue 

356 

357 # Find matching cluster 

358 matched_cluster = None 

359 best_overlap = overlap_threshold 

360 

361 for cluster in clusters: 

362 overlap = entity_overlap(msg_entities, cluster.all_entities) 

363 if overlap > best_overlap: 

364 best_overlap = overlap 

365 matched_cluster = cluster 

366 

367 if matched_cluster: 

368 entities_matched = sorted(msg_entities & matched_cluster.all_entities) 

369 matched_cluster.all_entities.update(msg_entities) 

370 matched_cluster.messages.append( 

371 { 

372 "uuid": msg.get("uuid", ""), 

373 "content": content[:80] + "..." if len(content) > 80 else content, 

374 "entities_matched": entities_matched, 

375 "timestamp": msg.get("timestamp"), 

376 } 

377 ) 

378 # Update cohesion score (average overlap of messages) 

379 matched_cluster.cohesion_score = ( 

380 matched_cluster.cohesion_score * (len(matched_cluster.messages) - 1) + best_overlap 

381 ) / len(matched_cluster.messages) 

382 else: 

383 cluster_counter += 1 

384 # Create new cluster 

385 primary = sorted(msg_entities)[:3] # Top 3 entities 

386 cluster = EntityCluster( 

387 cluster_id=f"cluster-{cluster_counter:03d}", 

388 primary_entities=primary, 

389 all_entities=msg_entities.copy(), 

390 messages=[ 

391 { 

392 "uuid": msg.get("uuid", ""), 

393 "content": content[:80] + "..." if len(content) > 80 else content, 

394 "entities_matched": sorted(msg_entities), 

395 "timestamp": msg.get("timestamp"), 

396 } 

397 ], 

398 cohesion_score=1.0, 

399 ) 

400 clusters.append(cluster) 

401 

402 # Populate span and inferred_workflow for each multi-message cluster 

403 for cluster in clusters: 

404 # Compute span from timestamps 

405 timestamps = _parse_timestamps(cluster.messages) 

406 if len(timestamps) >= 2: 

407 cluster.span = { 

408 "start": min(timestamps).isoformat(), 

409 "end": max(timestamps).isoformat(), 

410 } 

411 

412 # Infer workflow by matching cluster message content against WORKFLOW_TEMPLATES 

413 cluster_categories: set[str] = set() 

414 for m in cluster.messages: 

415 lower = m.get("content", "").lower() 

416 for category, keywords in _CONTENT_CATEGORY_MAP.items(): 

417 if any(kw in lower for kw in keywords): 

418 cluster_categories.add(category) 

419 

420 best_name: str | None = None 

421 best_score = 0.0 

422 for template_name, template_cats in WORKFLOW_TEMPLATES.items(): 

423 template_set = set(template_cats) 

424 if template_set: 

425 overlap = len(cluster_categories & template_set) / len(template_set) 

426 if overlap > best_score: 

427 best_score = overlap 

428 best_name = template_name 

429 if best_score >= 0.3: 

430 cluster.inferred_workflow = best_name 

431 

432 # Filter out single-message clusters 

433 return [c for c in clusters if len(c.messages) >= 2] 

434 

435 

436def _compute_boundaries( 

437 messages: list[dict[str, Any]], boundary_threshold: float = 0.6 

438) -> list[WorkflowBoundary]: 

439 """Compute workflow boundaries between consecutive messages.""" 

440 boundaries: list[WorkflowBoundary] = [] 

441 

442 # Sort by timestamp 

443 sorted_msgs = sorted(messages, key=lambda m: m.get("timestamp", "")) 

444 

445 # Pre-compute entity sets once per message (avoids re-extracting per sliding-window pair) 

446 all_entities = [extract_entities(m.get("content", "")) for m in sorted_msgs] 

447 

448 for i in range(len(sorted_msgs) - 1): 

449 msg_a = sorted_msgs[i] 

450 msg_b = sorted_msgs[i + 1] 

451 

452 # Parse timestamps 

453 pair_timestamps = _parse_timestamps([msg_a, msg_b]) 

454 if len(pair_timestamps) == 2: 

455 gap_seconds = int((pair_timestamps[1] - pair_timestamps[0]).total_seconds()) 

456 else: 

457 gap_seconds = 0 

458 

459 # Calculate time gap weight 

460 time_weight = calculate_boundary_weight(gap_seconds) 

461 

462 # Calculate entity overlap using pre-computed sets 

463 entities_a = all_entities[i] 

464 entities_b = all_entities[i + 1] 

465 overlap = entity_overlap(entities_a, entities_b) 

466 

467 # Adjust for entity overlap (reduce boundary weight if same topic) 

468 final_score = time_weight 

469 if overlap > 0.5: 

470 final_score = max(0.0, time_weight - 0.3) 

471 elif overlap > 0.3: 

472 final_score = max(0.0, time_weight - 0.15) 

473 

474 is_boundary = final_score >= boundary_threshold 

475 

476 boundaries.append( 

477 WorkflowBoundary( 

478 msg_a=msg_a.get("uuid", ""), 

479 msg_b=msg_b.get("uuid", ""), 

480 time_gap_seconds=gap_seconds, 

481 time_gap_weight=time_weight, 

482 entity_overlap=overlap, 

483 final_boundary_score=final_score, 

484 is_boundary=is_boundary, 

485 ) 

486 ) 

487 

488 return boundaries 

489 

490 

491def _get_message_category(msg_uuid: str, patterns: dict[str, Any]) -> str | None: 

492 """Look up message category from Step 1 patterns.""" 

493 for category_info in patterns.get("category_distribution", []): 

494 for example in category_info.get("example_messages", []): 

495 if example.get("uuid") == msg_uuid: 

496 category = category_info.get("category") 

497 return category if isinstance(category, str) else None 

498 return None 

499 

500 

501def _build_category_index(patterns: dict[str, Any]) -> dict[str, str]: 

502 """Build a flat UUID → category mapping from patterns category_distribution.""" 

503 index: dict[str, str] = {} 

504 for category_info in patterns.get("category_distribution", []): 

505 category = category_info.get("category") 

506 if not isinstance(category, str): 

507 continue 

508 for example in category_info.get("example_messages", []): 

509 uuid = example.get("uuid") 

510 if uuid: 

511 index[uuid] = category 

512 return index 

513 

514 

515def _detect_workflows( 

516 messages: list[dict[str, Any]], 

517 boundaries: list[WorkflowBoundary], 

518 patterns: dict[str, Any], 

519) -> list[Workflow]: 

520 """Detect multi-step workflows using template matching.""" 

521 workflows: list[Workflow] = [] 

522 workflow_counter = 0 

523 

524 # Sort messages by timestamp 

525 sorted_msgs = sorted(messages, key=lambda m: m.get("timestamp", "")) 

526 

527 # Build boundary index (msg_b uuid -> is_boundary) 

528 boundary_before: dict[str, bool] = {} 

529 for b in boundaries: 

530 boundary_before[b.msg_b] = b.is_boundary 

531 

532 # Build category index (uuid -> category) for O(1) lookups 

533 category_index = _build_category_index(patterns) 

534 

535 # Segment messages by boundaries 

536 segments: list[list[dict[str, Any]]] = [] 

537 current_segment: list[dict[str, Any]] = [] 

538 

539 for msg in sorted_msgs: 

540 uuid = msg.get("uuid", "") 

541 if boundary_before.get(uuid, False) and current_segment: 

542 segments.append(current_segment) 

543 current_segment = [] 

544 current_segment.append(msg) 

545 

546 if current_segment: 

547 segments.append(current_segment) 

548 

549 # Match each segment against workflow templates 

550 for segment in segments: 

551 if len(segment) < 2: 

552 continue 

553 

554 # Get categories for segment messages (from patterns) 

555 segment_categories: list[str] = [] 

556 for msg in segment: 

557 cat = category_index.get(msg.get("uuid", "")) 

558 if cat: 

559 segment_categories.append(cat) 

560 

561 if len(segment_categories) < 2: 

562 continue 

563 

564 # Find best matching template 

565 best_match: tuple[str, float] | None = None 

566 

567 for template_name, template_cats in WORKFLOW_TEMPLATES.items(): 

568 # Check if template categories appear in sequence (allowing gaps) 

569 template_idx = 0 

570 matches = 0 

571 

572 for cat in segment_categories: 

573 if template_idx < len(template_cats) and cat == template_cats[template_idx]: 

574 matches += 1 

575 template_idx += 1 

576 

577 if matches >= 2: # At least 2 template steps matched 

578 confidence = matches / len(template_cats) 

579 if best_match is None or confidence > best_match[1]: 

580 best_match = (template_name, confidence) 

581 

582 if best_match: 

583 workflow_counter += 1 

584 

585 # Calculate duration 

586 timestamps = _parse_timestamps(segment) 

587 

588 duration_minutes = 0 

589 if len(timestamps) >= 2: 

590 try: 

591 duration_minutes = int((max(timestamps) - min(timestamps)).total_seconds() / 60) 

592 except TypeError: 

593 duration_minutes = 0 

594 

595 # Get sessions 

596 session_ids = list({msg.get("session_id", "") for msg in segment}) 

597 

598 workflows.append( 

599 Workflow( 

600 workflow_id=f"wf-{workflow_counter:03d}", 

601 name=f"Detected: {best_match[0]}", 

602 pattern=best_match[0], 

603 pattern_confidence=best_match[1], 

604 messages=[ 

605 { 

606 "uuid": msg.get("uuid", ""), 

607 "category": category_index.get(msg.get("uuid", "")), 

608 "step": i + 1, 

609 } 

610 for i, msg in enumerate(segment) 

611 ], 

612 session_span=session_ids, 

613 duration_minutes=duration_minutes, 

614 ) 

615 ) 

616 

617 return workflows 

618 

619 

620# ----------------------------------------------------------------------------- 

621# Main API 

622# ----------------------------------------------------------------------------- 

623 

624 

625def analyze_workflows( 

626 messages_file: Path, 

627 patterns_file: Path, 

628 output_file: Path | None = None, 

629 overlap_threshold: float = 0.3, 

630 boundary_threshold: float = 0.6, 

631 verbose: bool = False, 

632 output_format: str = "yaml", 

633 db_path: Path | None = None, 

634) -> WorkflowAnalysis: 

635 """Main entry point: analyze workflows from messages and patterns. 

636 

637 Args: 

638 messages_file: Path to JSONL file with extracted user messages 

639 patterns_file: Path to YAML file from Step 1 (workflow-pattern-analyzer) 

640 output_file: Output path for step2-workflows.yaml (optional) 

641 overlap_threshold: Minimum entity overlap to cluster messages together (default: 0.3) 

642 boundary_threshold: Minimum boundary score to split workflow segments (default: 0.6) 

643 verbose: Emit per-stage progress to stderr (default: False) 

644 output_format: Output serialization format, "yaml" or "json" (default: "yaml") 

645 db_path: Optional path to a unified session DB. When provided and the 

646 DB has ``message_events`` rows (ENH-1621), messages are read from 

647 the DB and *messages_file* is ignored; an empty DB falls back to 

648 JSONL parsing. When omitted, the JSONL path is always used. 

649 

650 Returns: 

651 WorkflowAnalysis with all analysis results 

652 """ 

653 # Load inputs. Prefer the DB source when configured + populated; otherwise 

654 # fall back to the historical JSONL parsing path (ENH-1621). 

655 messages: list[dict[str, Any]] = [] 

656 source_name = messages_file.name 

657 if db_path is not None: 

658 db_messages = _load_messages_from_db(db_path) 

659 if db_messages: 

660 messages = db_messages 

661 source_name = str(db_path) 

662 if not messages: 

663 messages = _load_messages(messages_file) 

664 patterns = _load_patterns(patterns_file) 

665 

666 # Build metadata 

667 metadata = { 

668 "source_file": source_name, 

669 "patterns_file": patterns_file.name, 

670 "message_count": len(messages), 

671 "analysis_timestamp": datetime.now().isoformat(), 

672 "module": "workflow-sequence-analyzer", 

673 "version": "1.0", 

674 } 

675 

676 # Run analysis pipeline 

677 sessions = _group_by_session(messages) 

678 if verbose: 

679 print(f"[1/4] Linking sessions across {len(sessions)} session(s)...", file=sys.stderr) 

680 session_links = _link_sessions(sessions) 

681 if verbose: 

682 print(f"{len(session_links)} link(s) found", file=sys.stderr) 

683 if verbose: 

684 print("[2/4] Clustering by entities...", file=sys.stderr) 

685 entity_clusters = _cluster_by_entities(messages, overlap_threshold=overlap_threshold) 

686 if verbose: 

687 print(f"{len(entity_clusters)} cluster(s) found", file=sys.stderr) 

688 if verbose: 

689 print("[3/4] Computing workflow boundaries...", file=sys.stderr) 

690 boundaries = _compute_boundaries(messages, boundary_threshold=boundary_threshold) 

691 if verbose: 

692 print(f"{len(boundaries)} boundary/boundaries found", file=sys.stderr) 

693 if verbose: 

694 print("[4/4] Detecting workflows...", file=sys.stderr) 

695 workflows = _detect_workflows(messages, boundaries, patterns) 

696 if verbose: 

697 print(f"{len(workflows)} workflow(s) detected", file=sys.stderr) 

698 

699 # Cross-reference: link workflows to entity clusters and populate handoff_points 

700 uuid_to_cluster: dict[str, str] = {} 

701 for cluster in entity_clusters: 

702 for msg in cluster.messages: 

703 uuid = msg.get("uuid", "") 

704 if uuid: 

705 uuid_to_cluster[uuid] = cluster.cluster_id 

706 

707 uuid_to_content: dict[str, str] = { 

708 m.get("uuid", ""): m.get("content", "") for m in messages if m.get("uuid", "") 

709 } 

710 

711 for workflow in workflows: 

712 cluster_votes: dict[str, int] = {} 

713 for msg in workflow.messages: 

714 cluster_id = uuid_to_cluster.get(msg.get("uuid", "")) 

715 if cluster_id: 

716 cluster_votes[cluster_id] = cluster_votes.get(cluster_id, 0) + 1 

717 if cluster_votes: 

718 workflow.entity_cluster = max(cluster_votes, key=cluster_votes.__getitem__) 

719 

720 for msg in workflow.messages: 

721 uuid = msg.get("uuid", "") 

722 if uuid and _detect_handoff(uuid_to_content.get(uuid, "")): 

723 workflow.handoff_points.append({"uuid": uuid, "type": "explicit_handoff"}) 

724 

725 # Compute handoff analysis 

726 handoff_count = sum( 

727 1 

728 for link in session_links 

729 if "handoff_detected" in link.unified_workflow.get("evidence", []) 

730 ) 

731 

732 handoff_analysis: dict[str, Any] = { 

733 "total_handoffs": handoff_count, 

734 "handoff_patterns": [ 

735 {"pattern": "explicit_handoff", "count": handoff_count}, 

736 {"pattern": "session_timeout", "count": len(session_links) - handoff_count}, 

737 ], 

738 "recommendations": [], 

739 } 

740 

741 if len(session_links) > handoff_count: 

742 handoff_analysis["recommendations"].append( 

743 "Consider using /ll:handoff for cleaner session transitions" 

744 ) 

745 

746 # Build result 

747 analysis = WorkflowAnalysis( 

748 metadata=metadata, 

749 session_links=session_links, 

750 entity_clusters=entity_clusters, 

751 workflow_boundaries=boundaries, 

752 workflows=workflows, 

753 handoff_analysis=handoff_analysis, 

754 ) 

755 

756 # Write output if path provided 

757 if output_file: 

758 output_file = Path(output_file) 

759 output_file.parent.mkdir(parents=True, exist_ok=True) 

760 with open(output_file, "w", encoding="utf-8") as f: 

761 if output_format == "json": 

762 json.dump(analysis.to_dict(), f, indent=2, default=str) 

763 else: 

764 yaml.dump(analysis.to_dict(), f, default_flow_style=False, sort_keys=False) 

765 

766 return analysis