2525
2626VERDICT_MARKER = "<!-- kb-gate:verdict:v1 -->"
2727IN_FLIGHT = ("pr_requested" , "pr_open" , "verdict_posted" , "queued" )
28+ # a verdict must first verify within 72 hours: re-sign well before that
29+ RESIGN_AFTER = 48 * 3600
30+ # a queued PR that neither merged nor left the queue by then was dropped by a
31+ # failing merge group (GitHub does not tell us): re-sign and re-enqueue
32+ QUEUE_STALL = 2 * 3600
33+ # kb-gate failures, by what the service can do about them (matched against the
34+ # status description, which is the verifier's first problem)
35+ TRANSIENT_GATE = ("hold list" , "paused by the knowledge service" , "global pause" , "unreachable" ,
36+ "verifier error" , "fetched head does not match" ,
37+ "carries kb:hold" ) # left by our own pause until the re-signed verdict lifts it
38+ RESIGN_GATE = ("outside its issue window" , "no valid signed verdict" )
39+ REBUILD_GATE = ("context changed since the verdict" , "pages changed since the consistency judgement" ,
40+ "does not merge cleanly" , "signed patch manifest does not match" )
2841KNOWLEDGE_REPO_ENV = "KB_KNOWLEDGE_REPOSITORY"
2942DEFAULT_KNOWLEDGE_REPOSITORY = "JiusiServe/InferMatrixCopilot"
3043
@@ -102,6 +115,8 @@ def apply_acks(rt, acks: list[dict]) -> None:
102115 if kind == "open_pr" :
103116 fields .update (pr_number = int (ack ["pr" ]), head_sha = str (ack ["head_sha" ]),
104117 branch = str (ack .get ("branch" , "" )))
118+ elif kind == "enqueue" :
119+ fields ["detail" ] = {** changeset ["detail" ], "queued_at" : rt .clock ()}
105120 rt .ledger .update_changeset (changeset_id , ** fields )
106121
107122
@@ -136,11 +151,123 @@ def _pr_state(rt, number: int) -> dict:
136151
137152
138153def _gate_status (rt , sha : str ) -> str :
154+ return _gate (rt , sha )[0 ]
155+
156+
157+ def _gate (rt , sha : str , * , since : float = 0.0 ) -> tuple [str , str ]:
158+ """(state, description) of the latest kb-gate status on ``sha``; a status
159+ posted before ``since`` (the current verdict) is not about it: ("", "")."""
139160 combined = rt .github .get (f"/repos/{ knowledge_repository ()} /commits/{ sha } /status" )
140161 for status in combined .get ("statuses" , []):
141162 if status .get ("context" ) == "kb-gate" :
142- return str (status .get ("state" ) or "" )
143- return ""
163+ posted = _epoch (status .get ("updated_at" ) or status .get ("created_at" ))
164+ if since and posted and posted < since :
165+ return "" , ""
166+ return str (status .get ("state" ) or "" ), str (status .get ("description" ) or "" )
167+ return "" , ""
168+
169+
170+ def _epoch (value ) -> float :
171+ if not value :
172+ return 0.0
173+ import datetime as dt
174+
175+ try :
176+ return dt .datetime .fromisoformat (str (value ).replace ("Z" , "+00:00" )).timestamp ()
177+ except ValueError :
178+ return 0.0
179+
180+
181+ def _enqueue_outstanding (rt , changeset : dict ) -> bool :
182+ """A signed enqueue that the publisher may still execute (unexpired)."""
183+ pending = changeset .get ("pending_item" ) or {}
184+ return pending .get ("kind" ) == "enqueue" and float (pending .get ("expires_at" , 0 )) > rt .clock ()
185+
186+
187+ def _resign (rt , changeset : dict , why : str ) -> None :
188+ """Back to pr_open: the next pass signs a fresh verdict for the same head."""
189+ rt .ledger .update_changeset (changeset ["id" ], status = "pr_open" , pending_item = None )
190+ _outcome (rt , changeset , "resigned" , reason = why )
191+
192+
193+ REBUILD_MARK = "rebuild of "
194+
195+
196+ def _rebuild_of (rt , repo : str , old_id : str ) -> str | None :
197+ """A change set already staged as the rebuild of ``old_id`` (its evidence
198+ carries the mark, written atomically with the staging), so an interrupted
199+ rebuild is resumed instead of staged twice."""
200+ for candidate in rt .ledger .changesets (repo , ("gated" , "failed" , "human" , * IN_FLIGHT )):
201+ if candidate ["kind" ] != "rebuild" :
202+ continue
203+ try :
204+ evidence = rt .load_changeset_files (candidate ["id" ]).get ("evidence" ) or []
205+ except (OSError , ValueError ):
206+ continue
207+ if any (str (item .get ("source_reference" ) or "" ) == REBUILD_MARK + old_id for item in evidence ):
208+ return candidate ["id" ]
209+ return None
210+
211+
212+ def rebuild (rt , lifecycle , changeset : dict , why : str ) -> str | None :
213+ """Rebuild a change set whose verdict can no longer verify on the current
214+ base (context pages moved, or the PR no longer applies): re-apply its
215+ operations to current main and gate them again. Only a rebuild that passes
216+ the gate replaces the PR; the old one then waits in ``superseding`` until
217+ its close is observed. Anything else goes to people with the old PR left
218+ open. Needs the scheduler's lease; returns the new change set id or None."""
219+ from ..knowledge_service .lifecycle import LifecycleError
220+ from ..knowledge_service .ops import KnowledgeOperation , apply_operations
221+ from .runtime import gate_and_stage , publish
222+
223+ owner = getattr (rt , "lease_owner" , None )
224+ if not owner :
225+ return None
226+
227+ def to_people (reason : str ) -> None :
228+ rt .ledger .update_changeset (changeset ["id" ], status = "rebuild_failed" )
229+ rt .ledger .enqueue_human (lifecycle .repo , reason , changeset ["id" ])
230+
231+ detail = changeset ["detail" ]
232+ raw = detail .get ("operations" ) or []
233+ if not raw : # a mechanical change set (index fixes): nothing to re-apply
234+ to_people (f"kb-gate failed ({ why } ); no operations to rebuild" )
235+ return None
236+ new_id = _rebuild_of (rt , lifecycle .repo , changeset ["id" ])
237+ if new_id is None :
238+ base_sha = rt .knowledge .fetch ()
239+ base = rt .knowledge .knowledge_files (base_sha )
240+ external = rt .knowledge .external_texts (base_sha )
241+ operations = [KnowledgeOperation .from_dict (item ) for item in raw ]
242+ try :
243+ result = apply_operations (base , operations , release = detail ["release" ], today = rt .today ())
244+ except LifecycleError as exc :
245+ to_people (f"cannot rebuild on current main: { exc } " )
246+ return None
247+ evidence = [* (rt .load_changeset_files (changeset ["id" ]).get ("evidence" ) or []),
248+ {"source_reference" : REBUILD_MARK + changeset ["id" ], "title" : why }]
249+ new_id = gate_and_stage (rt , lifecycle , owner , kind = "rebuild" , base = base , base_sha = base_sha ,
250+ external = external , operations = operations , result = result ,
251+ evidence = evidence , event_ids = [], release = detail ["release" ],
252+ draft_keys = [])
253+ staged = rt .ledger .changeset (new_id )
254+ if staged ["status" ] not in ("gated" , * IN_FLIGHT ):
255+ # the rebuild itself did not pass the gate: nothing replaces the PR
256+ to_people (f"rebuild { new_id } on current main did not pass the gate ({ staged ['status' ]} )" )
257+ return None
258+ rt .ledger .update_changeset (changeset ["id" ], status = "superseding" ,
259+ detail = {** detail , "rebuilt_as" : new_id , "superseded_because" : why })
260+ _outcome (rt , changeset , "superseded" , reason = why , replaced_by = new_id )
261+ _request_close (rt , lifecycle .repo , rt .ledger .changeset (changeset ["id" ]))
262+ publish (rt , lifecycle , new_id )
263+ return new_id
264+
265+
266+ def _request_close (rt , repo : str , changeset : dict ) -> None :
267+ issue_once (rt , repo , changeset , "close" , {
268+ "changeset_id" : changeset ["id" ], "pr" : int (changeset ["pr_number" ]),
269+ "reason" : (f"Superseded by change set { changeset ['detail' ].get ('rebuilt_as' )} , rebuilt on "
270+ f"current main ({ changeset ['detail' ].get ('superseded_because' , '' )} )." )})
144271
145272
146273def advance (rt , lifecycle ) -> list [str ]:
@@ -149,12 +276,24 @@ def advance(rt, lifecycle) -> list[str]:
149276 if rt .outbox is None :
150277 return events
151278 paused = is_paused (rt .ledger , lifecycle .repo )
152- queued = rt .ledger .changesets (lifecycle .repo , ("queued" ,))
153- for changeset in rt .ledger .changesets (lifecycle .repo , IN_FLIGHT ):
279+ # the one queue slot per repository is taken by a queued PR AND by any PR
280+ # whose signed enqueue is still executable (not yet acked)
281+ queued = rt .ledger .changesets (lifecycle .repo , ("queued" ,)) + [
282+ cs for cs in rt .ledger .changesets (lifecycle .repo , ("verdict_posted" ,)) if _enqueue_outstanding (rt , cs )]
283+ for changeset in rt .ledger .changesets (lifecycle .repo , (* IN_FLIGHT , "superseding" )):
154284 number = changeset ["pr_number" ]
155285 if number is None :
156286 continue
157287 pr : dict [str , Any ] = _pr_state (rt , int (number ))
288+ if changeset ["status" ] == "superseding" :
289+ # the replaced PR stays tracked until GitHub shows it closed; the
290+ # close item is re-issued whenever the previous one expired
291+ if pr .get ("state" ) == "closed" or pr .get ("merged" ):
292+ rt .ledger .update_changeset (changeset ["id" ], status = "superseded" , pending_item = None )
293+ events .append (f"superseded { changeset ['id' ]} " )
294+ else :
295+ _request_close (rt , lifecycle .repo , changeset )
296+ continue
158297 if pr .get ("merged" ):
159298 rt .ledger .update_changeset (changeset ["id" ], status = "merged" ,
160299 merge_sha = str (pr .get ("merge_commit_sha" ) or "" ))
@@ -176,6 +315,41 @@ def advance(rt, lifecycle) -> list[str]:
176315 continue
177316 if paused :
178317 continue # the pause itself is issued by `kb pause` / breakers
318+ signed_at = float (changeset ["detail" ].get ("verdict_issued_at" ) or 0 )
319+ stale_verdict = bool (signed_at ) and rt .clock () - signed_at > RESIGN_AFTER
320+ queued_at = float (changeset ["detail" ].get ("queued_at" ) or 0 )
321+ if changeset ["status" ] == "queued" and (
322+ stale_verdict or (queued_at and rt .clock () - queued_at > QUEUE_STALL )):
323+ # never free the queue slot on a timer: dequeue (+ draft) first; the
324+ # pause ack returns it to pr_open, then it is re-signed and re-queued
325+ if issue_once (rt , lifecycle .repo , changeset , "pause" , {
326+ "changeset_id" : changeset ["id" ], "pr" : int (number ),
327+ "reason" : "re-signing: the verdict is ageing or the PR stalled in the merge queue" }):
328+ events .append (f"requeue { changeset ['id' ]} " )
329+ continue
330+ if changeset ["status" ] == "verdict_posted" and _enqueue_outstanding (rt , changeset ):
331+ continue # its ack decides: queued (then the queued path applies) or refused
332+ if changeset ["status" ] == "verdict_posted" and stale_verdict :
333+ _resign (rt , changeset , "verdict close to its issue window" )
334+ events .append (f"resign { changeset ['id' ]} " )
335+ continue
336+ if changeset ["status" ] == "verdict_posted" :
337+ state , description = _gate (rt , changeset ["head_sha" ], since = signed_at )
338+ if state == "failure" and not any (m in description for m in TRANSIENT_GATE ):
339+ if any (m in description for m in RESIGN_GATE ):
340+ _resign (rt , changeset , description )
341+ events .append (f"resign { changeset ['id' ]} " )
342+ elif any (m in description for m in REBUILD_GATE ):
343+ if not getattr (rt , "lease_owner" , None ):
344+ continue # only the scheduler (holding the lease) rebuilds
345+ new_id = rebuild (rt , lifecycle , changeset , description )
346+ events .append (f"rebuilt { changeset ['id' ]} as { new_id } " if new_id
347+ else f"rebuild_failed { changeset ['id' ]} " )
348+ else : # a real problem with the change itself: people decide
349+ rt .ledger .update_changeset (changeset ["id" ], status = "gate_failed" )
350+ rt .ledger .enqueue_human (lifecycle .repo , f"kb-gate failed: { description } " , changeset ["id" ])
351+ events .append (f"gate_failed { changeset ['id' ]} " )
352+ continue
179353 if changeset ["status" ] == "pr_open" :
180354 pending = changeset .get ("pending_item" )
181355 if pending and pending .get ("kind" ) == "post_verdict" and float (pending ["expires_at" ]) > rt .clock ():
@@ -185,6 +359,9 @@ def advance(rt, lifecycle) -> list[str]:
185359 "changeset_id" : changeset ["id" ], "pr" : int (number ), "head_sha" : changeset ["head_sha" ],
186360 "comment" : f"{ VERDICT_MARKER } \n ```json\n { json .dumps (envelope , ensure_ascii = False )} \n ```\n " ,
187361 }):
362+ current = rt .ledger .changeset (changeset ["id" ])
363+ rt .ledger .update_changeset (changeset ["id" ], detail = {
364+ ** current ["detail" ], "verdict_issued_at" : envelope ["payload" ]["issued_at" ]})
188365 events .append (f"verdict issued { changeset ['id' ]} " )
189366 elif changeset ["status" ] == "verdict_posted" and not queued :
190367 if _gate_status (rt , changeset ["head_sha" ]) == "success" :
0 commit comments