Coverage for little_loops / workflow_sequence / models.py: 100%

1"""Data models for workflow sequence analysis.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass, field 

6from typing import Any 

7 

8 

9@dataclass 

10class SessionLink: 

11 """Link between related sessions.""" 

12 

13 link_id: str 

14 sessions: list[dict[str, Any]] # session_id, position, link_evidence 

15 unified_workflow: dict[str, Any] # name, total_messages, span_hours 

16 confidence: float 

17 

18 def to_dict(self) -> dict[str, Any]: 

19 """Convert to dictionary for YAML serialization.""" 

20 return { 

21 "link_id": self.link_id, 

22 "sessions": self.sessions, 

23 "unified_workflow": self.unified_workflow, 

24 "confidence": self.confidence, 

25 } 

26 

27 

28@dataclass 

29class EntityCluster: 

30 """Cluster of messages sharing entities.""" 

31 

32 cluster_id: str 

33 primary_entities: list[str] 

34 all_entities: set[str] = field(default_factory=set) 

35 messages: list[dict[str, Any]] = field(default_factory=list) 

36 span: dict[str, Any] | None = None 

37 inferred_workflow: str | None = None 

38 cohesion_score: float = 0.0 

39 

40 def to_dict(self) -> dict[str, Any]: 

41 """Convert to dictionary for YAML serialization.""" 

42 return { 

43 "cluster_id": self.cluster_id, 

44 "primary_entities": self.primary_entities, 

45 "all_entities": sorted(self.all_entities), 

46 "messages": self.messages, 

47 "span": self.span, 

48 "inferred_workflow": self.inferred_workflow, 

49 "cohesion_score": round(self.cohesion_score, 2), 

50 } 

51 

52 

53@dataclass 

54class WorkflowBoundary: 

55 """Boundary between workflows based on time gaps and entity overlap.""" 

56 

57 msg_a: str 

58 msg_b: str 

59 time_gap_seconds: int 

60 time_gap_weight: float 

61 entity_overlap: float 

62 final_boundary_score: float 

63 is_boundary: bool 

64 

65 def to_dict(self) -> dict[str, Any]: 

66 """Convert to dictionary for YAML serialization.""" 

67 return { 

68 "between": {"msg_a": self.msg_a, "msg_b": self.msg_b}, 

69 "time_gap_seconds": self.time_gap_seconds, 

70 "time_gap_weight": round(self.time_gap_weight, 2), 

71 "entity_overlap": round(self.entity_overlap, 2), 

72 "final_boundary_score": round(self.final_boundary_score, 2), 

73 "is_boundary": self.is_boundary, 

74 } 

75 

76 

77@dataclass 

78class Workflow: 

79 """Identified multi-step workflow.""" 

80 

81 workflow_id: str 

82 name: str 

83 pattern: str 

84 pattern_confidence: float 

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

86 session_span: list[str] 

87 entity_cluster: str | None = None 

88 semantic_cluster: str | None = None 

89 duration_minutes: int = 0 

90 handoff_points: list[dict[str, Any]] = field(default_factory=list) 

91 

92 def to_dict(self) -> dict[str, Any]: 

93 """Convert to dictionary for YAML serialization.""" 

94 return { 

95 "workflow_id": self.workflow_id, 

96 "name": self.name, 

97 "pattern": self.pattern, 

98 "pattern_confidence": round(self.pattern_confidence, 2), 

99 "messages": self.messages, 

100 "session_span": self.session_span, 

101 "entity_cluster": self.entity_cluster, 

102 "semantic_cluster": self.semantic_cluster, 

103 "duration_minutes": self.duration_minutes, 

104 "handoff_points": self.handoff_points, 

105 } 

106 

107 

108@dataclass 

109class WorkflowAnalysis: 

110 """Complete workflow analysis output.""" 

111 

112 metadata: dict[str, Any] 

113 session_links: list[SessionLink] = field(default_factory=list) 

114 entity_clusters: list[EntityCluster] = field(default_factory=list) 

115 workflow_boundaries: list[WorkflowBoundary] = field(default_factory=list) 

116 workflows: list[Workflow] = field(default_factory=list) 

117 handoff_analysis: dict[str, Any] | None = None 

118 

119 def to_dict(self) -> dict[str, Any]: 

120 """Convert to dictionary for YAML serialization.""" 

121 return { 

122 "analysis_metadata": self.metadata, 

123 "session_links": [s.to_dict() for s in self.session_links], 

124 "entity_clusters": [c.to_dict() for c in self.entity_clusters], 

125 "workflow_boundaries": [b.to_dict() for b in self.workflow_boundaries], 

126 "workflows": [w.to_dict() for w in self.workflows], 

127 "handoff_analysis": self.handoff_analysis, 

128 }