Coverage for little_loops / workflow_sequence / analysis.py: 0%
340 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-29 00:54 -0500
1"""Analysis functions for workflow sequence detection."""
3from __future__ import annotations
5import json
6import re
7import sys
8from datetime import datetime
9from pathlib import Path
10from typing import Any
12import yaml
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)
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)
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}
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}
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}
66# -----------------------------------------------------------------------------
67# Core Analysis Functions
68# -----------------------------------------------------------------------------
71def extract_entities(content: str) -> set[str]:
72 """Extract file paths, commands, and concepts from message content.
74 Args:
75 content: Message text content
77 Returns:
78 Set of extracted entities (file paths, commands, issue IDs, etc.)
79 """
80 entities: set[str] = set()
82 # File paths
83 entities.update(FILE_PATTERN.findall(content))
85 # Phase/module references
86 entities.update(PHASE_PATTERN.findall(content.lower()))
87 entities.update(MODULE_PATTERN.findall(content.lower()))
89 # Slash commands
90 entities.update(COMMAND_PATTERN.findall(content))
92 # Issue IDs (normalize to uppercase)
93 entities.update(match.upper() for match in ISSUE_PATTERN.findall(content))
95 return entities
98def calculate_boundary_weight(gap_seconds: int) -> float:
99 """Calculate workflow boundary weight based on time gap.
101 Args:
102 gap_seconds: Time gap between messages in seconds
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
123def entity_overlap(entities_a: set[str], entities_b: set[str]) -> float:
124 """Calculate Jaccard similarity between two entity sets.
126 Args:
127 entities_a: First set of entities
128 entities_b: Second set of entities
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
140def get_verb_class(content: str) -> str | None:
141 """Extract verb class from message content.
143 Args:
144 content: Message text content
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))
152 for verb_class, verbs in VERB_CLASSES.items():
153 if words & verbs:
154 return verb_class
155 return None
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.
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)
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
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
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
195 # Entity overlap
196 entity_sim = entity_overlap(entities_a, entities_b)
198 # Category match
199 category_sim = 1.0 if category_a and category_a == category_b else 0.0
201 # Weighted combination
202 return keyword_sim * 0.3 + verb_sim * 0.3 + entity_sim * 0.3 + category_sim * 0.1
205# -----------------------------------------------------------------------------
206# Internal Analysis Functions
207# -----------------------------------------------------------------------------
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)
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
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
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
256 for i, session_a_id in enumerate(session_ids):
257 session_a = sessions[session_a_id]
258 if not session_a:
259 continue
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")
268 for session_b_id in session_ids[i + 1 :]:
269 session_b = sessions[session_b_id]
270 if not session_b:
271 continue
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")
279 # Calculate link score
280 score = 0.0
281 evidence: list[str] = []
283 # Same git branch (HIGH weight)
284 if branch_a and branch_a == branch_b:
285 score += 0.4
286 evidence.append("shared_branch")
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")
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")
302 if score > 0.3:
303 link_counter += 1
305 # Calculate span
306 timestamps = _parse_timestamps(session_a + session_b)
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
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 )
340 return links
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
350 for msg in messages:
351 content = msg.get("content", "")
352 msg_entities = extract_entities(content)
354 if not msg_entities:
355 continue
357 # Find matching cluster
358 matched_cluster = None
359 best_overlap = overlap_threshold
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
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)
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 }
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)
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
432 # Filter out single-message clusters
433 return [c for c in clusters if len(c.messages) >= 2]
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] = []
442 # Sort by timestamp
443 sorted_msgs = sorted(messages, key=lambda m: m.get("timestamp", ""))
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]
448 for i in range(len(sorted_msgs) - 1):
449 msg_a = sorted_msgs[i]
450 msg_b = sorted_msgs[i + 1]
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
459 # Calculate time gap weight
460 time_weight = calculate_boundary_weight(gap_seconds)
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)
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)
474 is_boundary = final_score >= boundary_threshold
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 )
488 return boundaries
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
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
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
524 # Sort messages by timestamp
525 sorted_msgs = sorted(messages, key=lambda m: m.get("timestamp", ""))
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
532 # Build category index (uuid -> category) for O(1) lookups
533 category_index = _build_category_index(patterns)
535 # Segment messages by boundaries
536 segments: list[list[dict[str, Any]]] = []
537 current_segment: list[dict[str, Any]] = []
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)
546 if current_segment:
547 segments.append(current_segment)
549 # Match each segment against workflow templates
550 for segment in segments:
551 if len(segment) < 2:
552 continue
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)
561 if len(segment_categories) < 2:
562 continue
564 # Find best matching template
565 best_match: tuple[str, float] | None = None
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
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
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)
582 if best_match:
583 workflow_counter += 1
585 # Calculate duration
586 timestamps = _parse_timestamps(segment)
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
595 # Get sessions
596 session_ids = list({msg.get("session_id", "") for msg in segment})
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 )
617 return workflows
620# -----------------------------------------------------------------------------
621# Main API
622# -----------------------------------------------------------------------------
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.
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.
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)
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 }
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)
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
707 uuid_to_content: dict[str, str] = {
708 m.get("uuid", ""): m.get("content", "") for m in messages if m.get("uuid", "")
709 }
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__)
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"})
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 )
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 }
741 if len(session_links) > handoff_count:
742 handoff_analysis["recommendations"].append(
743 "Consider using /ll:handoff for cleaner session transitions"
744 )
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 )
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)
766 return analysis