11from __future__ import annotations
22
3+ import json
34from collections .abc import Mapping
45from pathlib import Path
56
6- from ...rollout_event_log import load_rollout_events , rollout_event_log_path
7- from .effect_program import SETTLEMENT_IDENTITY_SCHEMA_VERSION
7+ from ...file_lock import exclusive_file_lock
8+ from ...rollout_event_log import (
9+ ROLLOUT_EVENT_SCHEMA_VERSION ,
10+ build_rollout_event ,
11+ load_rollout_events ,
12+ rollout_event_log_path ,
13+ )
14+ from .effect_program import SETTLEMENT_IDENTITY_SCHEMA_VERSION , SettlementIdentity
815
916HEARTBEAT_RECEIPT_SCHEMA_VERSION = "heartbeat_quota_receipt_v0"
1017
1118
19+ def _heartbeat_receipt_events (
20+ events : list [dict [str , object ]],
21+ * ,
22+ goal_id : str ,
23+ agent_id : str ,
24+ turn_instance_id : str ,
25+ ) -> list [dict [str , object ]]:
26+ return [
27+ event
28+ for event in events
29+ if event .get ("event_kind" ) == "quota_should_run"
30+ and str (event .get ("goal_id" ) or "" ) == goal_id
31+ and str (event .get ("agent_id" ) or "" ) == agent_id
32+ and str (event .get ("run_id" ) or "" ) == turn_instance_id
33+ ]
34+
35+
36+ def _receipt_settlement_identity (
37+ event : Mapping [str , object ],
38+ ) -> tuple [str , str ] | None :
39+ details_value = event .get ("details" )
40+ details : Mapping [str , object ] = (
41+ details_value if isinstance (details_value , Mapping ) else {}
42+ )
43+ todo_id = str (details .get ("todo_id" ) or "" ).strip ()
44+ effect_id = str (details .get ("settlement_effect_id" ) or "" ).strip ()
45+ if effect_id and not todo_id :
46+ raise ValueError (
47+ "heartbeat receipt has an effect identity without a Todo; refuse to "
48+ "infer or upgrade it"
49+ )
50+ if not todo_id :
51+ return None
52+ if not effect_id :
53+ goal_id = str (event .get ("goal_id" ) or "" ).strip ()
54+ agent_id = str (event .get ("agent_id" ) or "" ).strip ()
55+ turn_instance_id = str (event .get ("run_id" ) or "" ).strip ()
56+ effect_id = SettlementIdentity (
57+ goal_id = goal_id ,
58+ agent_id = agent_id ,
59+ todo_id = todo_id ,
60+ turn_instance_id = turn_instance_id ,
61+ ).effect_id
62+ return todo_id , effect_id
63+
64+
65+ def _effective_heartbeat_receipt (
66+ events : list [dict [str , object ]],
67+ ) -> dict [str , object ] | None :
68+ if not events :
69+ return None
70+ identities : dict [tuple [str , str ], dict [str , object ]] = {}
71+ for event in events :
72+ identity = _receipt_settlement_identity (event )
73+ if identity is not None :
74+ identities [identity ] = event
75+ if len (identities ) > 1 :
76+ raise ValueError (
77+ "heartbeat receipt has conflicting settlement identities for the "
78+ "same goal, agent, and turn"
79+ )
80+ if identities :
81+ return next (iter (identities .values ()))
82+ return events [- 1 ]
83+
84+
1285def find_heartbeat_receipt (
1386 runtime_root : Path ,
1487 * ,
@@ -17,15 +90,120 @@ def find_heartbeat_receipt(
1790 turn_instance_id : str ,
1891) -> dict [str , object ] | None :
1992 events = load_rollout_events (rollout_event_log_path (runtime_root , goal_id ))
20- for event in reversed (events ):
21- if (
22- event .get ("event_kind" ) == "quota_should_run"
23- and str (event .get ("goal_id" ) or "" ) == goal_id
24- and str (event .get ("agent_id" ) or "" ) == agent_id
25- and str (event .get ("run_id" ) or "" ) == turn_instance_id
26- ):
27- return event
28- return None
93+ return _effective_heartbeat_receipt (
94+ _heartbeat_receipt_events (
95+ events ,
96+ goal_id = goal_id ,
97+ agent_id = agent_id ,
98+ turn_instance_id = turn_instance_id ,
99+ )
100+ )
101+
102+
103+ def upgrade_identityless_heartbeat_receipt (
104+ runtime_root : Path ,
105+ * ,
106+ goal_id : str ,
107+ agent_id : str ,
108+ turn_instance_id : str ,
109+ todo_id : str ,
110+ settlement_effect_id : str ,
111+ status : str ,
112+ summary : str ,
113+ details : Mapping [str , object ],
114+ ) -> tuple [dict [str , object ], bool ]:
115+ """Append one settlement-bound receipt after an identity-less same-turn guard.
116+
117+ The correction is append-only and serialized with the rollout log lock. A
118+ matching correction replays, a legacy Todo-only receipt gains its derived
119+ effect id, and effect-only or conflicting identities fail closed.
120+ """
121+
122+ normalized_todo_id = str (todo_id ).strip ()
123+ normalized_effect_id = str (settlement_effect_id ).strip ()
124+ expected_effect_id = SettlementIdentity (
125+ goal_id = goal_id ,
126+ agent_id = agent_id ,
127+ todo_id = normalized_todo_id ,
128+ turn_instance_id = turn_instance_id ,
129+ ).effect_id
130+ if not normalized_todo_id or normalized_effect_id != expected_effect_id :
131+ raise ValueError (
132+ "heartbeat receipt upgrade requires the deterministic selected Todo "
133+ "settlement identity"
134+ )
135+
136+ log_path = rollout_event_log_path (runtime_root , goal_id )
137+ log_path .parent .mkdir (parents = True , exist_ok = True )
138+ with exclusive_file_lock (log_path ):
139+ try :
140+ lines = log_path .read_text (encoding = "utf-8" ).splitlines ()
141+ except OSError :
142+ lines = []
143+ events : list [dict [str , object ]] = []
144+ for line in lines :
145+ try :
146+ parsed = json .loads (line )
147+ except json .JSONDecodeError :
148+ continue
149+ if (
150+ isinstance (parsed , dict )
151+ and parsed .get ("schema_version" ) == ROLLOUT_EVENT_SCHEMA_VERSION
152+ ):
153+ events .append (parsed )
154+ matching = _heartbeat_receipt_events (
155+ events ,
156+ goal_id = goal_id ,
157+ agent_id = agent_id ,
158+ turn_instance_id = turn_instance_id ,
159+ )
160+ effective = _effective_heartbeat_receipt (matching )
161+ if effective is None :
162+ raise ValueError (
163+ "identity-less heartbeat receipt is missing; rerun the original guard"
164+ )
165+ existing_identity = _receipt_settlement_identity (effective )
166+ expected_identity = (normalized_todo_id , normalized_effect_id )
167+ if existing_identity is not None :
168+ if existing_identity != expected_identity :
169+ raise ValueError (
170+ "heartbeat receipt settlement identity conflicts with the "
171+ "current selected Todo"
172+ )
173+ existing_details_value = effective .get ("details" )
174+ existing_details = (
175+ existing_details_value
176+ if isinstance (existing_details_value , Mapping )
177+ else {}
178+ )
179+ if str (existing_details .get ("settlement_effect_id" ) or "" ).strip ():
180+ return effective , False
181+
182+ corrected_details = dict (details )
183+ corrected_details .update (
184+ {
185+ "turn_instance_id" : turn_instance_id ,
186+ "todo_id" : normalized_todo_id ,
187+ "settlement_effect_id" : normalized_effect_id ,
188+ "settlement_receipt_revision" : "identity_upgrade" ,
189+ }
190+ )
191+ source_event_id = str (effective .get ("event_id" ) or "" ).strip () or None
192+ corrected = build_rollout_event (
193+ goal_id = goal_id ,
194+ event_kind = "quota_should_run" ,
195+ agent_id = agent_id ,
196+ todo_id = normalized_todo_id ,
197+ run_id = turn_instance_id ,
198+ status = status ,
199+ summary = summary ,
200+ source_event_id = source_event_id ,
201+ caused_by = source_event_id ,
202+ details = corrected_details ,
203+ )
204+ with log_path .open ("a" , encoding = "utf-8" ) as handle :
205+ handle .write (json .dumps (corrected , sort_keys = True , ensure_ascii = False ) + "\n " )
206+ return corrected , True
29207
30208
31209def heartbeat_receipt_view (
0 commit comments