Coverage for node / src / stigmem_node / models / intents.py: 96%

1"""Intent envelope models.""" 

2 

3from __future__ import annotations 

4 

5from pydantic import BaseModel, ConfigDict, Field, field_validator 

6 

7from .constants import VALID_SCOPES 

8from .facts import FactValue 

9 

10VALID_ESCALATION_PRIORITIES = {"low", "medium", "high", "critical"} 

11VALID_ESCALATION_CHANNELS = {"stigmem", "email", "slack"} 

12 

13 

14class Constraint(BaseModel): 

15 """Hard limit on an intent (Spec-X8-Intent-Envelope).""" 

16 

17 kind: str = Field(..., min_length=1) 

18 limit: FactValue 

19 unit: str | None = None 

20 

21 

22class Preference(BaseModel): 

23 """Soft preference on an intent (Spec-X8-Intent-Envelope).""" 

24 

25 kind: str = Field(..., min_length=1) 

26 value: FactValue 

27 weight: float = Field(1.0, ge=0.0, le=1.0) 

28 

29 

30class DeferenceRule(BaseModel): 

31 """When to defer to another entity before proceeding (Spec-X8-Intent-Envelope).""" 

32 

33 condition: str = Field(..., min_length=1) 

34 defer_to: str = Field(..., min_length=1) 

35 timeout_s: int | None = Field(None, ge=0) 

36 

37 

38class EscalationPolicy(BaseModel): 

39 """Who to notify and how when escalation is required (Spec-X8-Intent-Envelope).""" 

40 

41 escalate_to: str = Field(..., min_length=1) 

42 channel: str = Field("stigmem") 

43 priority: str = Field("medium") 

44 include_context: bool = True 

45 

46 @field_validator("priority") 

47 @classmethod 

48 def check_priority(cls, p: str) -> str: 

49 if p not in VALID_ESCALATION_PRIORITIES: 

50 raise ValueError(f"priority must be one of {VALID_ESCALATION_PRIORITIES}") 

51 return p 

52 

53 @field_validator("channel") 

54 @classmethod 

55 def check_channel(cls, c: str) -> str: 

56 if c not in VALID_ESCALATION_CHANNELS: 

57 raise ValueError(f"channel must be one of {VALID_ESCALATION_CHANNELS}") 

58 return c 

59 

60 

61class HandoffArtifact(BaseModel): 

62 """Named artifact reference passed along with a handoff (Spec-X8-Intent-Envelope).""" 

63 

64 name: str = Field(..., min_length=1) 

65 ref: str = Field(..., min_length=1, description="URI of the artifact fact") 

66 

67 

68class HandoffPayload(BaseModel): 

69 """Structured context transfer for agent handoffs (Spec-X8-Intent-Envelope). 

70 

71 fact_refs MUST include any facts the receiver needs to reconstitute context. 

72 """ 

73 

74 summary: str = Field(..., min_length=1) 

75 fact_refs: list[str] = Field(default_factory=list) 

76 continuation: str | None = None 

77 artifacts: list[HandoffArtifact] = Field(default_factory=list) 

78 

79 

80class IntentEnvelopeRequest(BaseModel): 

81 """POST /v1/intents request body (Spec-X8-Intent-Envelope). 

82 

83 The ``from`` field is aliased as ``from_uri`` in Python to avoid the keyword clash. 

84 Clients MUST send the JSON key ``"from"``. 

85 """ 

86 

87 model_config = ConfigDict(populate_by_name=True) 

88 

89 id: str | None = Field( 

90 None, 

91 description="Optional client-generated intent ID (UUID). Auto-assigned when omitted.", 

92 ) 

93 from_uri: str = Field(..., alias="from", min_length=1, description="Sender entity URI") 

94 to: list[str] = Field( 

95 ..., min_length=1, description="Target entity URIs (at least one required)" 

96 ) 

97 goal: str = Field(..., min_length=1, max_length=2048) 

98 scope: str = Field("company") 

99 expires_at: str | None = Field(None, description="ISO 8601 UTC expiry; null = non-expiring") 

100 constraint: list[Constraint] = Field(default_factory=list) 

101 preference: list[Preference] = Field(default_factory=list) 

102 deference: list[DeferenceRule] = Field(default_factory=list) 

103 escalation: EscalationPolicy | None = None 

104 handoff: HandoffPayload | None = None 

105 

106 @field_validator("scope") 

107 @classmethod 

108 def check_scope(cls, s: str) -> str: 

109 if s not in VALID_SCOPES: 

110 raise ValueError(f"scope must be one of {VALID_SCOPES}") 

111 return s 

112 

113 @field_validator("to") 

114 @classmethod 

115 def check_to_non_empty(cls, to: list[str]) -> list[str]: 

116 if not to: 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true

117 raise ValueError("to must contain at least one target URI") 

118 if any(not t for t in to): 118 ↛ 119line 118 didn't jump to line 119 because the condition on line 118 was never true

119 raise ValueError("to entries must be non-empty strings") 

120 return to 

121 

122 

123class IntentEnvelopeRecord(BaseModel): 

124 """Response for POST /v1/intents and GET /v1/intents/:id (Spec-X8-Intent-Envelope).""" 

125 

126 model_config = ConfigDict(populate_by_name=True) 

127 

128 id: str 

129 from_uri: str = Field(..., alias="from", description="Sender entity URI") 

130 to: list[str] 

131 goal: str 

132 scope: str 

133 created_at: str 

134 expires_at: str | None 

135 constraint: list[Constraint] 

136 preference: list[Preference] 

137 deference: list[DeferenceRule] 

138 escalation: EscalationPolicy | None 

139 handoff: HandoffPayload | None 

140 fact_ids: list[str] = Field( 

141 description="IDs of all facts written into the fabric for this envelope" 

142 ) 

143