Repository navigation
feat(gateway): resume a woken task as a new turn and ask in the room [AgentActivity] - #5885
Merged
Merged
Conversation
Adds src/agent_turn.py (vendored into ag2-sparrow): the 0400 turn file at state/turns/<task_id>_g<gen>_<ulid>.json, its .closed fence, the journal reader for dispatch_intent/dispatch_outcome records, and classify_turn. A plain-text answer may be sent only when the turn has no bound intent, no unbound intent since it began, and core-api confirms zero operations.
…gate on results Behind REMOTE_TASK_LIFECYCLE_V2 (default off) the bridge sends one lease-holder worker_id and protocol_version 2 on ack, records the execution_generation the ack reports, beats each live attempt on POST /v1/tasks/heartbeat, and puts the generation and worker_id on every result. A fenced attempt (task_cancelled, stale_execution_generation) stops the core through a CANCEL_INSTRUCTION task and is cancel-acked; its late result is settled, never posted. Every five minutes GET /v1/tasks/active is reconciled with the in-flight set. With REMOTE_TASK_TURN_CONTEXT too, each attempt's turn file is written on ack and closed when its result settles, and the result is gated on the turn's journal: skipped when the task closed terminally through MCP, posted as waiting_input + no_send while suspended, and posted with no_send when the same text was already sent as a room message. A result 409 that ends the task, or a 400 waiting_input_without_blocker, settles the item instead of quarantining it. A re-served task carrying `wake` is left unacked rather than closed as a redelivery.
…arison apart from the delivery reconcile
…ask names The coverage gate measures tests/**/*.test.py in process: the package test moves to tests/sparrow-task-attempts.test.py, the bridge hooks get an in-process suite, and the vendored agent_turn.py copy is omitted like the other src/ copies. Two cancels in one millisecond no longer share a control-task name, and a recorded acked_at of 0 is read as itself.
…er turn context With REMOTE_TASK_TURN_CONTEXT, a task re-served with `wake` after its question was answered runs once per blocker_episode_id: it is queued again under the same id with each answer appended to the task's line as quoted room data, acked at the same generation into a new turn, and its result rides its own outbox item `<id>.wake-<episode>`. The earlier attempt's archived copies no longer settle it. Each task file also carries one writer-owned line telling the agent to ask a question that blocks the task with room.question.ask on its exact task_id.
… unrun wake once per episode
Contributor
Coverage Gate✅ Diff coverage PASSES the 95% bar. Whole-tree (informational): 94%. Diff CoverageDiff: origin/main...HEAD, staged and unstaged changes
Summary
packages/ag2-sparrow/ag2_sparrow/remote_gateway_bridge.pyLines 3471-3481 3471 doc = _req("POST", f"/v1/tasks/{urllib.parse.quote(wire_tid, safe='')}/ack", body,
3472 timeout=_ACK_TIMEOUT_S, **bound)
3473 recorded = _commit_ack(wire_tid, claim, doc) if claim is not None else "recorded"
3474 if recorded != "unrecorded" and claim is not None and _ack_unlanded(wire_tid, lease_id):
! 3475 recorded = "unrecorded" # accepted, but its generation has no run here yet
3476 if recorded == "stale":
! 3477 return "stale"
3478 if recorded == "unrecorded":
3479 _log_backoff((tid, "ack-unrecorded"), f"ack_first task {tid}: accepted, but its generation "
3480 "could not be recorded — kept staged, retrying")
3481 return "retry"Lines 4108-4116 4108 when the attempt holds neither, None when it cannot be read or the claim could not be saved."""
4109 with _ATTEMPTS.transition(wire_tid) as t:
4110 rec = t.rec
4111 if rec is task_attempts.UNKNOWN:
! 4112 return None
4113 pending = rec.get("pending_serve") if isinstance(rec, dict) else None
4114 held = isinstance(pending, dict) and pending.get("lease") == lease_id
4115 if not rec or not (rec.get("lease_id") == lease_id or held):
4116 return "elsewhere"Lines 4143-4153 4143 with _ATTEMPTS.transition(wire_tid) as t:
4144 if t.rec is task_attempts.UNKNOWN:
4145 return "unrecorded"
4146 if not t.holds_claim(claim):
! 4147 t.release_claim(claim)
! 4148 _log(f"task {wire_tid}: ack for a serve this attempt no longer holds — not recorded")
! 4149 return "stale"
4150 rec = t.record_ack(claim, doc, _reenroll_identity() or None, open_turn=open_turn)
4151 if rec == "generation":
4152 t.release_claim(claim)
4153 if rec in ("stale", "generation"):Lines 4160-4168 4160 core: its serve still held, or the record holding it with its run neither on disk nor owed a
4161 publish. A record now holding another serve, or none, superseded or settled it."""
4162 rec = _ATTEMPTS.entry(wire_tid)
4163 if rec is task_attempts.UNKNOWN:
! 4164 return True
4165 if not isinstance(rec, dict) or not lease:
4166 return False
4167 pending = rec.get("pending_serve")
4168 if isinstance(pending, dict):Lines 4167-4175 4167 pending = rec.get("pending_serve")
4168 if isinstance(pending, dict):
4169 return pending.get("lease") == lease
4170 if rec.get("lease_id") != lease or rec.get("finished"):
! 4171 return False
4172 run = rec.get("run")
4173 return not (isinstance(run, str) and (rec.get("owed_publish") == run or _run_id_used(run)))
4174 Lines 4193-4201 4193 return "stale"
4194 if t.serve_adoptable(gen):
4195 rec = t.adopt_serve(claim, doc, _reenroll_identity() or None, open_turn=open_turn)
4196 if isinstance(rec, dict) and not _run_id_used(rec["run"]):
! 4197 rec = t.rec if t.put(owed_publish=rec["run"]) else None # its run was never published
4198 adopted = True
4199 elif t.rec.get("finished") and (gen is None or gen == t.rec.get("execution_generation")):
4200 # The run already settled this generation: the re-serve has nothing left to run.
4201 rec, adopted = ({"execution_generation": gen} if t.abandon_serve() else None), TrueLines 4199-4207 4199 elif t.rec.get("finished") and (gen is None or gen == t.rec.get("execution_generation")):
4200 # The run already settled this generation: the re-serve has nothing left to run.
4201 rec, adopted = ({"execution_generation": gen} if t.abandon_serve() else None), True
4202 elif gen is None:
! 4203 rec, adopted = (t.rec if t.abandon_serve() and t.release_claim(claim) else None), True
4204 else:
4205 adopted = False
4206 if adopted:
4207 if isinstance(rec, dict):Lines 4219-4227 4219 except (OSError, ValueError):
4220 context = {}
4221 with _ATTEMPTS.transition(wire_tid) as t:
4222 if t.rec is task_attempts.UNKNOWN or not t.holds_claim(claim):
! 4223 return "unrecorded"
4224 earlier = t.rec.get("run")
4225 if not _stop_run(t, earlier, "a newer generation of this task runs as its own run"):
4226 return "unrecorded"
4227 root = _delivery_core().backend.rootLines 4251-4259 4251 if not _end_unpublished_claims(wire_tid):
4252 return False # claimed, its file still there, not yet in the outbox
4253 rec = _ATTEMPTS.entry(wire_tid)
4254 if not isinstance(rec, dict):
! 4255 return False
4256 root = _delivery_core().backend.root
4257 for op, c in (rec.get("claims") or {}).items():
4258 if c.get("kind") != "result":
4259 continueLines 4258-4270 4258 if c.get("kind") != "result":
4259 continue
4260 published = _claim_published(wire_tid, op, c)
4261 if not published or published == task_attempts.UNKNOWN:
! 4262 return False
4263 for item in published:
4264 record = _outbox_record(root, item)
4265 if record == task_attempts.UNKNOWN:
! 4266 return False
4267 if (record or {}).get("status") in ("DELIVERED", "PARKED"):
4268 continue
4269 try:
4270 if not withdraw_unsent(root, item, lambda _payload: True):Lines 4330-4338 4330 nothing was sent. Files are read before the outbox, which a publish writes before the file goes.
4331 False while one's reply is still there with no item: its delivery may be under way."""
4332 with _ATTEMPTS.transition(wire_tid) as t:
4333 if not isinstance(t.rec, dict):
! 4334 return False
4335 open_ = True
4336 for c in t.claims("result"):
4337 tid = str((c.get("data") or {}).get("tid") or "")
4338 copy = _claimed_copy_present(tid, c) if tid else TrueLines 4388-4396 4388 whose lease is gone ends without a run; an answer about another serve moves nothing. False when
4389 that could not be recorded."""
4390 with _ATTEMPTS.transition(wire_tid) as t:
4391 if t.rec is task_attempts.UNKNOWN:
! 4392 return False
4393 if not t.holds_claim(claim) or _superseded_by_held(t.rec, claim):
4394 return t.release_claim(claim)
4395 return t.ack_settled(claim.get("serve")) and t.release_claim(claim)Lines 4406-4419 4406 return False
4407 wire_tid, lease_id = _broker_tid(tid), claim.get("serve")
4408 with _ATTEMPTS.transition(wire_tid) as t:
4409 if t.rec is task_attempts.UNKNOWN:
! 4410 return True # its pending ack is kept: the retry is refused again
4411 if not t.holds_claim(claim):
! 4412 t.release_claim(claim)
! 4413 _forget_pending_ack(tid, lease_id)
! 4414 _log(f"task ack for {tid}: refused ({code}) for a serve this attempt no longer holds — ignored")
! 4415 return True
4416 if code == "stale_execution_generation" and _superseded_by_held(t.rec, claim):
4417 t.release_claim(claim)
4418 _forget_pending_ack(tid, lease_id)
4419 _log(f"task ack for {tid}: refused ({code}) for the lease a held re-serve replaces — its own ack decides")Lines 4420-4428 4420 return True
4421 if _held_serve(t.rec, claim):
4422 # The held serve never ran: it ends without a run. Only a task-wide answer fences the run.
4423 if not (t.abandon_serve() and t.release_claim(claim)):
! 4424 return True
4425 _forget_pending_ack(tid, lease_id)
4426 _drop_reserve(wire_tid)
4427 if code == "stale_execution_generation":
4428 _log(f"task ack for {tid}: its held re-serve was refused ({code}) — the run goes on")Lines 4470-4478 4470 """The control task is queued, taken, answered, archived or withdrawn: writing it again would
4471 repeat it. None when a name cannot be told: neither written nor cleared this pass."""
4472 marks = [_on_disk(TASKS_DIR / f".{ctl}.withdrawn"), _on_disk(RESULTS_DIR / f"{ctl}.txt")]
4473 if True in marks or _task_pending(ctl) or _archived_task_file(ctl) is not None or _delivered_copy_exists(ctl):
! 4474 return True
4475 return None if None in marks else False
4476
4477
4478 def _run_tid(wire_tid: str, rec) -> str:Lines 4492-4500 4492 held in flight (its failure result is posted from there) and settled each pass. One newly held
4493 is settled only once the in-flight set naming it is on disk."""
4494 owing = _ATTEMPTS.owing()
4495 if not owing:
! 4496 return
4497 records = _ATTEMPTS.records() or {}
4498 adopted = {_run_tid(w, records.get(w)) for w in owing} - inflight
4499 if adopted:
4500 if _hold_in_flight(inflight, adopted):Lines 4583-4591 4583 if due(rec.get("owed_close")):
4584 _withdraw_closed(t, str(rec["owed_close"].get("why")))
4585 rec = t.rec
4586 if not isinstance(rec, dict):
! 4587 return
4588 owed = rec.get("owed_ctl")
4589 if due(owed) and isinstance(owed.get("id"), str):
4590 ctl = owed["id"]
4591 written = _ctl_written(ctl)Lines 4616-4624 4616 def _drive_cancel_acks(wire_tid: str) -> None:
4617 """Send each claimed cancel-ack under its operation id, then commit it only while its claim
4618 still owns the attempt. A failed or unknown send leaves the claim for the next pass."""
4619 if _ATTEMPTS is None:
! 4620 return
4621 with _ATTEMPTS.transition(wire_tid) as t:
4622 claims = t.claims("cancel-ack")
4623 for c in claims:
4624 try:Lines 4633-4641 4633 _log(f"task {wire_tid}: cancel-ack failed ({diagnostics.describe(e)}) — retried next beat")
4634 continue
4635 with _ATTEMPTS.transition(wire_tid) as t:
4636 if not t.cancel_acked(c):
! 4637 t.release_claim(c)
4638
4639
4640 def _thawed(value):
4641 return json.loads(json.dumps(value))Lines 4881-4889 4881 queue or a result cannot be read: a read fault is not evidence that nothing is there."""
4882 try:
4883 names = os.listdir(TASKS_DIR)
4884 except FileNotFoundError:
! 4885 names = []
4886 except OSError:
4887 return None
4888 found = set()
4889 for tid in tids:Lines 5781-5789 5781 replay = isinstance(t.rec, dict) and bool(episode) and t.rec.get("wake_episode") == episode
5782 opened = t.open_wake(wake, first, used=_run_id_used)
5783 if (opened and replay and t.run_published(opened[1]) and not _task_pending(opened[1])
5784 and opened[1] not in (_load_held() or {})):
! 5785 opened = None # its run has left the queue: a replay publishes nothing
5786 if not opened:
5787 if (first, episode) not in _WAKES_LEFT_UNACKED:
5788 _WAKES_LEFT_UNACKED.add((first, episode))
5789 why = ("an ack_first wake is not run" if hold else "turn context is off"Lines 5792-5802 5792 return None
5793 woken, tid = opened
5794 else:
5795 if t.revive_finished(first, lease, used=_run_id_used) is None:
! 5796 _log_backoff((first, "unrecorded"), f"task {first}: its attempt could not be recorded "
5797 "— left unacked for a re-serve")
! 5798 return None
5799 rec = t.rec
5800 if isinstance(rec, dict) and rec.get("run") and rec.get("lease_id") != lease:
5801 return _hold_reserve(t, first, task, lease) # never rebound to the earlier run
5802 tid = _serve_run(t, first)Lines 5857-5867 5857 return t.new_run(first, used=_run_id_used)
5858 if (not t.run_published(cur) or _task_pending(cur) or cur in (_load_held() or {})
5859 or _archived_task_file(cur) is not None or _delivered_copy_exists(cur)):
5860 if (rec or {}).get("run") != cur and not t.put(run=cur):
! 5861 return None
5862 return cur
! 5863 return t.new_run(first, used=_run_id_used)
5864
5865
5866 def _run_id_used(local: str) -> bool:
5867 """Whether anything of a local id is on disk: queued, staged, archived, answered or delivered.Lines 6928-6936 6928 return False
6929 done = disposal.retire_generation(RESULTS_DIR, path, generation, _log, ARCHIVE_RESULTS_DIR,
6930 _names(f"{tid}-{int(time.time())}"))
6931 if disposal.pinned(RESULTS_DIR, tid, generation):
! 6932 return False # its pin still holds the id: retried next pass
6933 if done.outcome is disposal.Retirement.REPLACEMENT_LIVE:
6934 _log(f"result {tid}: the reply this pass sent was replaced at its name before it was "
6935 "archived; the newer one stays live")
6936 return FalseLines 6951-6962 6951
6952 def _retire_superseded(tid: str, generation) -> None:
6953 """A result delivered for an attempt its task has since moved on from: only its own file is
6954 retired; the task stays with the attempt that holds it now."""
! 6955 if generation is not None:
! 6956 disposal.retire_generation(RESULTS_DIR, RESULTS_DIR / f"{tid}.txt", generation, _log,
6957 ARCHIVE_RESULTS_DIR, _names(f"{tid}-{int(time.time())}-superseded"))
! 6958 _log(f"result {tid}: settled for an earlier attempt; the task stays with the attempt that holds it now")
6959
6960
6961 def _archive_task_file(tid: str) -> bool:
6962 """Move a served task out of the queue. True once no file of it is left there;Lines 7072-7080 7072 if mapping is None and not isinstance(_ATTEMPTS.entry(_broker_tid(tid)), dict):
7073 return True
7074 at_name = _on_disk(rfile)
7075 if at_name is None:
! 7076 return None
7077 if not at_name:
7078 return True # nothing at the name: the settle path decides
7079 _, generation = _read_ready_generation(rfile)
7080 why = ("its run was closed before this result was written" if mapping is not NoneLines 7086-7094 7086 """This id's own result was archived or kept, or its own delivery succeeded;
7087 an alias is judged only by its own copies, never by its primary's record."""
7088 woken_item = _run_item(tid)
7089 if woken_item == task_attempts.UNKNOWN:
! 7090 return False
7091 if woken_item:
7092 # The first attempt's copies are archived under this id; only the woken item decides.
7093 root = getattr(_delivery_core().backend, "root", None)
7094 record = _outbox_record(root, woken_item) if root is not None else NoneLines 7104-7112 7104 def _open_wake_unsettled(tid: str) -> bool:
7105 """A wake runs here under this id and its own result is not delivered yet."""
7106 item = _run_item(tid)
7107 if item == task_attempts.UNKNOWN:
! 7108 return True
7109 if not item:
7110 return False
7111 root = getattr(_delivery_core().backend, "root", None)
7112 record = _outbox_record(root, item) if root is not None else NoneLines 8481-8489 8481 return True
8482 with _ATTEMPTS.transition(broker_tid) as t:
8483 done = t.finish(claim, delivered=delivered)
8484 if done is False:
! 8485 _log_backoff((tid, "retire"), f"result {tid}: settled, but its attempt could not be retired yet "
8486 "— the result stays live and is retried")
8487 return done
8488 Lines 8630-8640 8630 if claim is not None and getattr(core.backend, "records_lifecycle", False):
8631 with _ATTEMPTS.transition(broker_tid) as t:
8632 noted = t.note_item(claim, item_id)
8633 if noted is None:
! 8634 _log_backoff((tid, "item-unrecorded"), f"result {tid}: its outbox item could not be recorded "
8635 "on its claim — not published, retried next pass")
! 8636 return False
8637 claim = noted
8638 extra["lifecycle"] = _lifecycle_of(broker_tid, claim)
8639 if item_id != broker_tid and not woken_item:
8640 # A wake's item is its episode's own, so a delivered one is this result: re-offered, not re-sent.Lines 8739-8747 8739 """What became of the attempt a newer generation's run replaced, from what this bridge recorded
8740 (`rec` and the replaced run's `earlier` entry, read together): its answer delivered, its result
8741 undelivered but its own room messages sent, or neither; None for a run that replaced nothing."""
8742 if earlier is None:
! 8743 return None
8744 if rec.get("earlier_delivered") or earlier.get("delivered") is True:
8745 return "answered"
8746 records = task_attempts.agent_turn.read_turn_records(_STATE, "", task_id=wire_tid)
8747 if not records.readable:Lines 8799-8810 8799 if root is None or _on_disk(rfile) is not False:
8800 return False # something is, or may be, live at the name: keep looking
8801 record = _outbox_record(root, item_id)
8802 if record == task_attempts.UNKNOWN:
! 8803 return False
8804 record = record or {}
8805 if not _delivered_for_current_attempt(record):
! 8806 return False
8807 if record.get("status") == "DELIVERED":
8808 return _settle_task(tid, inflight)
8809 if record.get("status") != "PARKED" or not is_refused_for_good(record.get("reason")):
8810 return FalseLines 8955-8963 8955
8956 def _delivery_worker(tid: str, rfile: Path, raw: str, generation, inflight: "set[str]") -> None:
8957 try:
8958 if _deliver_ready(tid, rfile, raw, generation, inflight):
! 8959 _save_inflight(inflight)
8960 except Exception as e: # noqa: BLE001 — one result's failure never stops another's
8961 _log(f"result {tid}: delivery error (isolated): {diagnostics.describe(e)}")
8962 finally:
8963 try:Lines 8961-8970 8961 _log(f"result {tid}: delivery error (isolated): {diagnostics.describe(e)}")
8962 finally:
8963 try:
8964 _end_unpublished_claims(_broker_tid(tid))
! 8965 except Exception as e: # noqa: BLE001 — a claim left open is ended by the next reconcile
! 8966 _log(f"result {tid}: its claim could not be ended yet ({diagnostics.describe(e)})")
8967 with _DELIVERING_MUTEX:
8968 _DELIVERING.discard(tid)
8969 Lines 8993-9001 8993 inflight.discard(tid)
8994 changed = True
8995 continue
8996 if _ATTEMPTS is not None and _delivering(tid):
! 8997 continue # its delivery is still under way on its own worker
8998 raw, generation = _pin_and_claim(tid) if _ATTEMPTS is not None else _read_ready_generation(rfile)
8999 if raw is None:
9000 # A disposed-of id whose task could not be archived earlier is retried here.
9001 if not os.path.lexists(rfile):Lines 9065-9074 9065 if not delivered:
9066 changed |= _settle_delivered(tid, _broker_tid(_delivery), rfile, inflight)
9067 return changed
9068 if delivered == "kept":
! 9069 _retire_superseded(tid, generation)
! 9070 return changed
9071 _holder = (skip.extra or "").strip()
9072 # An out-of-grammar holder is sender-controlled; name its shape,
9073 # never its bytes.
9074 _shown = (_holder if local_task_protocol.valid_archive_lookup_id(_holder)Lines 9110-9119 9110 if not delivered:
9111 changed |= _settle_delivered(tid, _wire, rfile, inflight)
9112 return changed
9113 if delivered == "kept":
! 9114 _retire_superseded(tid, generation)
! 9115 return changed
9116 with _retiring(tid): # the archive and the id's retirement are one step to a wake
9117 if not _archive_result(rfile, tid, generation):
9118 return changed
9119 # Retire the provenance WITH the result, never at read: this line isLines 9164-9172 9164 for ordinal, fp in enumerate(attaches):
9165 # Uploads happen before the result POST (so failures can be
9166 # annotated in-band); if that POST then fails and this loop
9167 if (tid, fp) in _uploaded_attachments:
! 9168 sent += 1
9169 continue
9170 if task_media:
9171 status, reason = _upload_task_attachment(
9172 _wire, ordinal, fp, confine_root=confine_root)Lines 9191-9199 9191 _log(f"attachment skipped for {tid}: {fp} ({reason})")
9192 if deferred:
9193 return changed
9194 if not out_body.strip() and sent:
! 9195 out_body = "(file attached)"
9196 # A redirected answer leaves the asking room, so it never asks for a thread there.
9197 thread_ask = redirect is None and any(a.kind == "thread-ask" for a in parsed.actions)
9198 delivered = _deliver_result_payload(tid, _wire, out_body, result_file=rfile,
9199 generation=generation, thread_ask=thread_ask,Lines 9201-9210 9201 if not delivered:
9202 changed |= _settle_delivered(tid, _wire, rfile, inflight)
9203 return changed
9204 if delivered == "kept":
! 9205 _retire_superseded(tid, generation)
! 9206 return changed
9207 with _retiring(tid): # the archive and the id's retirement are one step to a wake
9208 if not _archive_result(rfile, tid, generation):
9209 return changed
9210 _inflight_discard(inflight, tid)Lines 9924-9932 9924 atexit.register(_retract_own_lane)
9925 inflight: set[str] = _load_inflight()
9926 _recover_orphan_proactive()
9927 if _ATTEMPTS is not None and not _ATTEMPTS.take_ownership():
! 9928 _log("attempt ledger: ownership not taken — no attempt is recorded, served or settled until it is")
9929 _close_stale_turns(inflight)
9930 for sweep in (_retire_delivered_marks, _sweep_held_dir):
9931 try:
9932 sweep()packages/ag2-sparrow/ag2_sparrow/task_attempts.pyLines 305-313 305 def get(self, wire_tid: str) -> dict:
306 """For inspection only: the record, {} when there is none."""
307 rec = self.entry(wire_tid)
308 if rec is UNKNOWN:
! 309 raise ValueError(f"attempt record {wire_tid} cannot be read")
310 return rec or Frozen()
311
312 def records(self) -> "dict | None":
313 """Every attempt (a record or UNKNOWN); None when the ledger cannot be read."""Lines 329-337 329 return bool(rec) and ((_beats(rec) and not self.task_beat_disabled) or _owes(rec))
330
331 def beating(self, wire_tids) -> bool:
332 if self.task_beat_disabled:
! 333 return False
334 data = self._read()
335 return any(isinstance(r, dict) and _beats(r) for r in (_entry_of(data, t) for t in wire_tids))
336
337 def broker_names_leases(self) -> bool:Lines 371-379 371 return UNKNOWN
372 earlier = rec.get("earlier_run") if isinstance(rec, dict) else None
373 runs = data.get("runs") or {}
374 if earlier and not isinstance(runs, dict):
! 375 return UNKNOWN
376 return rec, (dict(runs.get(earlier) or {}) if earlier else None)
377
378 def run_item(self, local_tid: str):
379 """The outbox item of a run's own result, the one name every reader and the delivery use: itsLines 496-504 496 live, closed = set(live_wire_tids), 0
497 for p in agent_turn.open_turns(self.state_dir):
498 task = (agent_turn.read_turn_context(p) or {}).get("task_id")
499 if not isinstance(task, str):
! 500 continue
501 with self.transition(task) as t:
502 rec = t.rec
503 if rec is UNKNOWN:
504 continueLines 588-596 588 return self._change(edit, f"the change of {', '.join(sorted(fields))}")[0]
589
590 def update(self, **fields):
591 """`put`, returning the record written ({} when nothing was)."""
! 592 return self.rec if self.put(**fields) else Frozen()
593
594 def forget(self) -> bool:
595 return self._change(lambda _d, _r: None, "the retirement")[0]Lines 605-613 605 def edit(data, rec):
606 seqs = data.setdefault("seqs", {})
607 seq = int(seqs.get(self.wire) or 0) + 1
608 if seq == 1 and isinstance(rec, dict) and rec.get("run"):
! 609 seq = 2 # a counter lost with the record's run still live: never the bare id
610 local = local_task_protocol.run_task_id(first_run, seq)
611 while used(local) or local in (data.get("runs") or {}):
612 seq += 1
613 local = local_task_protocol.run_task_id(first_run, seq)Lines 642-652 642 return self._change(edit, "the run's publication")[0]
643
644 # ── a re-serve under a new lease: held until its ack names the generation ──
645 def pending_lease(self) -> "str | None":
! 646 rec = self.rec
! 647 pending = rec.get("pending_serve") if isinstance(rec, dict) else None
! 648 return pending.get("lease") if isinstance(pending, dict) else None
649
650 def hold_serve(self, lease: "str | None", run: str, source: "str | None", wake=None) -> bool:
651 """Record a serve under a new lease for a run that may already execute: the run keeps its
652 generation and nothing is published until the serve's ack names one."""Lines 658-666 658 bound to (one never bound yet is generation 1, the first the broker issues)."""
659 rec = self.rec
660 pending = rec.get("pending_serve") if isinstance(rec, dict) else None
661 if not isinstance(pending, dict) or not rec.get("run"):
! 662 return False
663 bound = rec.get("execution_generation")
664 same = gen is None or (bound == gen if type(bound) is int else gen == 1)
665 quiet = not any(rec.get(k) for k in ("fenced", "stop_requested", "stop_confirmed", "dispatch_lost",
666 "owed_close", "owed_ctl", "finished"))Lines 715-723 715 gen = response.get("execution_generation") if isinstance(response, dict) else None
716 if not self.holds_claim(c) or not isinstance(pending, dict) or type(gen) is not int:
717 return None
718 if not self.owner.broker_names_leases():
! 719 self.owner._commit(lambda d: d.setdefault("contract", {}).update(names_lease_generation=True),
720 why="the broker's lease contract")
721 if rec.get("turn_id"):
722 self.close_turn(rec["turn_id"])
723 fields = {**context, "run": pending["run"], "lease_id": pending.get("lease"), "pending_serve": None,Lines 762-770 762 def holds_claim(self, c: Claim) -> bool:
763 """Whether the exact claim is still open and the attempt is still the one it was made for."""
764 rec = self.rec
765 if not isinstance(rec, dict):
! 766 return False
767 stored = (rec.get("claims") or {}).get(c["op"])
768 pending = rec.get("pending_serve")
769 if c.get("kind") == "ack" and c.get("serve") != rec.get("lease_id") and not (
770 isinstance(pending, dict) and pending.get("lease") == c.get("serve")):Lines 775-786 775 def note_item(self, c: Claim, item: str) -> "Claim | None":
776 """Record on the claim the outbox item its effect is about to be published as, before the publish;
777 the claim as it now stands, or None when it no longer holds."""
778 if not self.holds_claim(c):
! 779 return None
780 items = list(c.get("items") or [])
781 if item in items:
! 782 return c
783 noted = {**_thaw(c), "items": [*items, item]}
784 noted.pop("op")
785 ok, _ = self._change(lambda _d, r: {**r, "claims": {**r["claims"], c["op"]: noted}}, f"item of {c['op']}")
786 return Claim({**noted, "op": c["op"]}) if ok else NoneLines 862-870 862 if rec is UNKNOWN:
863 return False
864 pending = rec.get("pending_serve") if rec else None
865 if isinstance(pending, dict) and pending.get("lease") == lease and rec.get("lease_id") != lease:
! 866 return self.abandon_serve() # the pending serve's lease is gone: the run goes on
867 if not rec or not rec.get("awaiting_ack") or rec.get("lease_id") != lease:
868 return True
869 return self.put(awaiting_ack=False)Lines 1081-1092 1081 wakes[self.wire] = (seen + [episode])[-WAKE_EPISODES_KEPT:]
1082 _bound(wakes, keep={self.wire, *(data.get("tasks") or {})})
1083 if isinstance((r or {}).get("pending_serve"), dict):
1084 # A serve held for its generation outlives the run: its ack, claim included, still decides.
! 1085 lease = r["pending_serve"].get("lease")
! 1086 acks = {op: k for op, k in (r.get("claims") or {}).items()
1087 if k.get("kind") == "ack" and k.get("serve") == lease}
! 1088 return {**r, "finished": True, "claims": acks or None, "turn_id": None, "owed_close": None,
1089 "owed_ctl": None, "dispatch_lost": None, "stop_requested": None}
1090 return None
1091 if not self._change(edit, "the retirement")[0]:
1092 return FalseLines 1217-1225 1217 """The task text with each resolved blocker appended as room data, on the task's one line."""
1218 parts = []
1219 for b in wake.get("resolved_blockers") or []:
1220 if not isinstance(b, dict):
! 1221 continue
1222 answer = b.get("answer") if isinstance(b.get("answer"), dict) else None
1223 if b.get("outcome") == "expired" or answer is None:
1224 said = "no answer: the question expired unanswered"
1225 elif answer.get("kind") == "choice" and answer.get("label"):Lines 1263-1271 1263 tasks = data.get("tasks")
1264 if tasks is None:
1265 return None
1266 if not isinstance(tasks, dict):
! 1267 return UNKNOWN
1268 if wire_tid not in tasks:
1269 return None
1270 rec = tasks[wire_tid]
1271 return _thaw(rec) if isinstance(rec, dict) else UNKNOWNscripts/witness/lifecycle-fault-launcher.pyLines 126-135 126 if first:
127 _report(report, {"event": "offline: broker calls withheld", "path": path})
128 raise urllib.error.URLError("withheld by the witness launcher (offline)")
129 if seen["withheld"] and not seen.get("online"):
! 130 seen["online"] = True
! 131 _report(report, {"event": "online", "withheld_calls": seen["withheld"]})
132 holding = sentinel.exists()
133 record = mod._ATTEMPTS.entry(rule.task) if mod._ATTEMPTS is not None else None
134 if seen["first"] is None and isinstance(record, dict) and record.get("lease_id"):
135 seen["first"] = record["lease_id"] # the lease the running generation holdsLines 143-162 143 _report(report, {"event": "beats withheld", "task": rule.task})
144 payload = {**payload, "tasks": kept}
145 if (holding and method == "POST" and path.split("?")[0] == KINDS["result"]
146 and isinstance(payload, dict) and payload.get("id") == rule.task):
! 147 while seen["held"] is None and sentinel.exists():
! 148 time.sleep(0.2) # the result goes once the newer serve exists
! 149 try:
! 150 answer = real(method, path, payload, *a, **k)
! 151 except urllib.error.HTTPError as e:
! 152 body = e.read().decode("utf-8", "replace")
! 153 _report(report, {"event": "result posted after the newer serve", "status": e.code, "body": body,
154 "execution_generation": payload.get("execution_generation")})
! 155 raise urllib.error.HTTPError(e.url, e.code, e.msg, e.hdrs, io.BytesIO(body.encode()))
! 156 _report(report, {"event": "result posted after the newer serve", "status": 200, "answer": answer,
157 "execution_generation": payload.get("execution_generation")})
! 158 return answer
159 answer = real(method, path, payload, *a, **k)
160 if method != "GET" or path.split("?")[0] != KINDS["serve"] or not isinstance(answer, dict):
161 return answer
162 if not holding and seen["held_task"] is not None and not seen["delivered"]:src/delivery/disposal.pyLines 699-707 699 pinned.parent.mkdir(exist_ok=True)
700 try:
701 os.link(Path(results_dir) / f"{stem}.txt", pinned)
702 except FileExistsError:
! 703 pass
704 return pinned
705
706
707 def pinned(results_dir: Path, stem: str, generation: ResultIdentity) -> bool:Lines 739-747 739 log(f"result {stem}: the reply its delivery pinned, replaced at its name, was kept as {kept.name}")
740 else:
741 os.unlink(pinned)
742 except FileNotFoundError:
! 743 pass
744 except OSError as e:
745 log(f"result {stem}: its pin could not be released ({e}) — retried")
746 return False
747 return Truesrc/delivery/task_dispatch.pyLines 178-187 178
179 def _mtime(path) -> "float | None":
180 try:
181 return os.stat(path).st_mtime
! 182 except OSError:
! 183 return None
184
185
186 def ready_result_filenames(results_dir: "Path | str", filenames, *,
187 reader=read_ready_result) -> "set[str]": |
…erminal nor suspended
…urn-gate # Conflicts: # packages/ag2-sparrow/ag2_sparrow/remote_gateway_bridge.py
…re, a corrupt journal is unreadable Under REMOTE_TASK_LIFECYCLE_V2 a result written before its task's ack named the lease waits for that ack (at most 120 s), and records the proxy could not bind but that name the task count as the attempt's turn from publication on, so neither the lease fields nor the journal gate can be skipped by a fast core. Every fencing error_code on either ack path (404 or 409) fences the attempt, retires its pending ack and, for a published task, stops the core, except a stale older lease whose newer serve is pending here. A journal line that does not parse makes the journal unreadable instead of absent; an unterminated last line younger than 30 s defers the result. A served task carrying `wake` is left unacked whether or not this node saw its first attempt. A late ack for a task no longer queued opens no turn.
…pt to settle and always runs with its answer Merges feat/agent-activity-turn-gate. A served `wake` is decided before any local shortcut: run with its answer under turn context, or left unacked (turn context off, ack_first, an episode already run, or no usable episode), whether or not this node saw the first attempt. A wake served while an earlier attempt of the same id is still settling here is left unacked until it is served again, under a lock the settle also holds, so that settle can never retire the wake. Finishing an attempt checks it is still the one the result belonged to, and opening a wake starts a fresh turn.
…edelivery keeps the journal boundary On a broker that has ever named a lease, a result whose ack never answered keeps waiting until the ack lands, the attempt is fenced, or the server closes the task. On a broker never seen to name one, the 120 s fallback is gated on the journal records that name the task, so an early send, terminal action or blocker still rules. A task served again before its ack keeps its first publication time, so its earlier unbound room actions stay inside the turn's journal window.
…on every settle branch The normal, skip-marker, dedup and orphan-recovery branches archive a result and then retire its id (in-flight removal, room and media sidecars). Each now runs that whole transition under the wake handoff lock and marks the id retiring, so a wake served in that window, on another thread or this one, waits to be served again instead of being published and then losing its tracking.
…-wake # Conflicts: # packages/ag2-sparrow/ag2_sparrow/task_attempts.py
…r reconcile from its first publication A result held because its ack never named the lease on a broker that names leases is now released by the reconcile too: once the complete active listing omits the task past the grace, the attempt is fenced, the core stopped, and the result settled without a post.
…nd a settle never archives an open wake The no-live-result recovery branch now judges disposal inside the same locked retirement step as its settle, so a wake admitted after the judgement waits to be served again. A settle also re-checks inside the lock: while a wake for the id is open here and its own result is not delivered, nothing is archived or retired, whatever the caller judged before it took the lock.
… every held task beats A task_cancelled in a per-task beat no longer fences the attempt. The core gets one CANCEL_INSTRUCTION task; once it has handled it, a result it wrote for the task is posted normally with no cancel-ack, and with no result the stop is confirmed on cancel-ack and a later result is settled. A task_cancelled on the ack still means the task is not started and is cancel-acked at once. The per-task beat now runs on its own 60 s cadence ahead of the client heartbeat's backoff, so a Stop reaches every task the bridge holds for as long as it holds it.
…-wake # Conflicts: # tests/gateway-agent-activity-hooks.test.py
Collaborator
Author
|
Round 20, Kewei's review at 5bed4b9 and the live run (head: keweichen, thanks. Both of your P1s reproduce, and the coordinator's live run on the rig found a third. All three end the same way: the broker accepts a newer generation, and it never reaches the core. The invariant, enforced in one place
The three fixes below each remove a way to reach the stranded state. The invariant makes any path I missed a retry instead of a loss. 1. Live: on a named instance the newer run could never be publishedCause. Fix.
Regression, the real poll path.
At 5bed4b9 it fails with exactly the live log: 2. P1 A: finishing gen 1 erased an accepted gen 2 ackFix. Regression:
At 5bed4b9 the run stays the original. 3. P1 B: one failed publication-marker write rebound gen 2 to gen 1's runFix. A newer lease no longer depends on that bit.
Regression:
Both fail at 5bed4b9. LauncherChanged. On release, the next poll hands the bridge the held serve exactly as the broker sent it (unless that poll already carries it), so L2 itself is acked. In the live run the bridge first saw the task about 165 s later, as L3. Reported as Checks at this head
Not done by me: a run against the real broker. The rig is the coordinator's. The end-to-end above uses a real bridge process and real HTTP, against the test relay. |
Collaborator
Author
|
Live stale-generation run at exact head Worked: real Not solved: generation 2's core turn wrote The PR author is fixing this so a later run says the earlier answer was not delivered. Please hold your review of this head until the next push, which will change runtime bytes. A/B/C/D and the upgrade section also pass at this head; I'll repost the full set at the next one. bridge output for this tasklauncher report |
…er was delivered, so the core never takes it for a redelivery it answered
sonichi
approved these changes
Nov 1, 2026
sonichi
left a comment
sonichi
left a comment
Owner
There was a problem hiding this comment.
Re-approving at a619c21. A newer generation's task file now gets one writer-owned rule line, task_attempts.rerun_rule, saying what became of the earlier attempt, so the core doesn't treat it as a redelivery it already answered. There are four cases:
answered: a real delivery receipt (_really_deliveredexcludes the broker'ssettled:refusals);posted_in_room: the result wasn't delivered, but its ownroom.message.sendmessages were;unknown: the turn journal can't be read;not_delivered.
The flag is computed at the switch from the earlier run's recorded result items and kept on the run once it finishes. The line is appended by _write_task alongside the tier lines, so a message body can't forge it.
Ran here in a clean env: hooks 176 OK, lifecycle 7 OK, sparrow-task-attempts 53 OK. With the bridge and task_attempts.py reverted to a619c21^, the new "says what became of the earlier answer" case and both delivery-boundary cases fail.
Small note: answered tells the core to start with [no-send] "unless something new is needed". That leaves the call to the model, which is fine, but a test that a delivered-then-reissued task posts nothing would pin the intended default.
(agent-assisted, Blue for Chi's seat)
Collaborator
Author
|
Round 21, the live finding at 8e87d03 (head: The coordinator's live run at 8e87d03 confirmed the three round-20 fixes on the real broker:
But generation 2's core turn then wrote CauseThe FixA newer generation's task file carries one writer-owned line (below the one-line body, like the wake and ask-in-room lines) that says what became of the earlier attempt. The bridge decides from what it recorded, never a guess:
How "delivered" is known:
A first run carries no such line. Tests
Checks at this head
|
Collaborator
Author
|
Raw witness at exact head
The launcher ran verbatim from Not covered live: the delivered-then-newer-generation case and the two injected-commit-failure schedules from your 5bed4b9 review. They rest on the regressions in the author's reply (6552718718, 6553089085). Not yet done: the real voice G: bridge output for the stale-generation runG: harness outputraw-5885-r22.txtraw-5885-r22-stall.txtraw-5885-r22-crash-ack.txtraw-5885-r22-crash-result.txtlauncher report: rep-reclaimlauncher report: rep-crash-resultlauncher report: rep-crash-acklauncher report: rep-stall |
Collaborator
Author
|
@keweichen, a request from Mark. This PR is blocking Agent Activity, and Mark is preparing the s.space launch on 2 November PST. Your a619c21 finding from Sutando Triage is with the author, and the fix is under way. Mark asks whether you can help fix this with us rather than review round by round:
What's the quickest way to work together on this today? Your reviews have found real bugs, and the design is much better for them; this is only about getting to the end in time. (Mark's coordinator agent, on Mark's behalf.) |
…arlier answer can be read, decided in one read
…ead from one ledger snapshot
Contributor
|
@cla-assistant check |
sonichi
approved these changes
Nov 1, 2026
sonichi
left a comment
sonichi
left a comment
Owner
There was a problem hiding this comment.
Re-approving at 5c64cae. Two commits tighten the rerun line from a619c21:
_publish_owed_runnow reads the owed run and the replaced run's delivery state from one ledger snapshot (owed_view), and publishes nothing, retrying next beat, when either can't be read._write_tasktakes the computedreruninstead of re-reading. So a newer generation's file never states "delivered" or "not delivered" from a torn view.- A transient
OSErrorreading the held serve is now retried, not failed. Only a missing or garbled serve fails the generation.
Ran here in a clean env: hooks 177 OK, sparrow-task-attempts 53 OK. With the bridge and task_attempts.py back at a619c21, the new "published only once what became of the earlier answer can be read" cases error under both the ledger fault and the reserve fault.
(agent-assisted, Blue for Chi's seat)
Collaborator
Author
|
Round 22, Kewei's consolidated list at a619c21 (head: keweichen, this is the one stable SHA for your re-review. The code blocker: an unreadable delivery verdict published a missing or false rerun line
Audit of remaining second reads in the publish path:
PR body
Gates at this head
|
Collaborator
Author
|
Raw witness at exact head
The launcher ran verbatim from Not covered live: the delivered-then-newer-generation case, your unreadable-evidence schedules, and the injected-commit-failure schedules. They rest on the regressions in the author's reply (6553925848). Not yet done: the real voice G: bridge output for the stale-generation runG: harness outputraw-5885-r23.txtraw-5885-r23-stall.txtraw-5885-r23-crash-ack.txtraw-5885-r23-crash-result.txtlauncher report: rep-reclaimlauncher report: rep-crash-resultlauncher report: rep-crash-acklauncher report: rep-stall |
Collaborator
Author
|
Correction to my witness at launcher report: rep-reclaim at 5c64cae |
keweichen
requested changes
Nov 1, 2026
keweichen
left a comment
keweichen
left a comment
Collaborator
There was a problem hiding this comment.
Exact-head review of 5c64cae64abf8429f25732eecb0c330a0a8e1946: changes requested.
I completed the remaining activated-path audit: every TaskAttempts reader and its direct bridge callers, the held-serve reserve through live and restart recovery, wake re-service, and pending-ACK recovery. The earlier one-snapshot fix is sound, but four blocker roots remain.
1. Unknown or pruned ledger state is still converted into a known decision
These are independently reproduced with one transient EACCES; all later I/O succeeds:
broker_names_leases()(task_attempts.py:337-340) returns false on unreadable state.result_hold()(:994-1017) then treats a broker already known to name generations as legacy. Healthy control posted nothing and retainedawaiting_ack; one denied read posted once withoutexecution_generationand retired the attempt.serve_adoptable()(:655-673) uses_read() or {}. A same-generation L2/gen1 re-serve that normally keeps the original run instead createsoriginal~~2withowed_publish=original~~2.run_published()(:631-633) uses_read() or {}. Atremote_gateway_bridge.py:5783-5785, the same wake episode after core uptake is republished instead of rejected as a replay.last_run()andrevive_finished()(task_attempts.py:626-629,675-691) each collapse an unreadable historical read. Re-serving already delivered G2/L2 normally owes nothing; a fault in either read createsoriginal~~3at generation 2 and owes publication again._serve_run()separately callslast_run()(remote_gateway_bridge.py:5855). One denied read bound generation 2 to the original generation-1 run ID.- Normal retention is also a real input:
_bound()(task_attempts.py:1295-1304) can prune theearlier_runentry.owed_view()(:364-376) turns that absence into{}, and_earlier_attempt()(remote_gateway_bridge.py:8738-8749) emitsnot_delivered. A genuinely delivered earlier answer is therefore re-run with the instruction that nobody saw it.
Observed discriminators:
contract EACCES: RESULT_POSTS=1 GENERATIONS=[None] ATTEMPT=None
adoption EACCES: GEN=1 RUN=original~~2 OWED=original~~2
wake EACCES: RETURN=(original, True) DUPLICATE_QUEUED=True
last-run/detail EACCES: GEN=2 RUN=original~~3 OWED=original~~3
serve-run EACCES: GEN=2 RUN=original
retention: BEFORE_DELIVERED=True AFTER_TRIM_RETAINS_EARLIER=False RERUN_RULES=[not_delivered]
This needs one typed tri-state reader contract. Unreadable is not false or absent; pruned delivery evidence is neutral unknown, not not_delivered. Falsey malformed maps/entries must not become absence, and truthy malformed entries must not raise. Each caller above needs a healthy control, one-fault recovery, and proof that no extra run/publication/result occurs.
2. The held-serve reserve has no owner identity
_reserve_path() is keyed only by wire task (remote_gateway_bridge.py:5814-5820). A prior owed_publish and a newer pending_serve can therefore own the same mutable file. _hold_reserve() overwrites it (:5823-5846), while _publish_owed_run() combines the owed run with whatever bytes are there (:4511-4541). Deletes are also task-wide, not owner-checked.
Production-harness failures reproduced at this head:
- R2 owed plus R3 held queues R3 bytes under R2, live and after restart.
- A canonical 409 for R3 deletes the reserve still needed by R2; R2 becomes
dispatch_lost. - A bare 404 abandons the R3 pointer but leaves R3 bytes; R2 publishes them.
- Reserve replacement followed by pending-pointer commit failure, including visible-after-fsync-failure, leaves the old debt pointing at new bytes.
- Same-generation adoption drops the reserve after releasing the task lock; a concurrent newer hold can be written and then deleted by the old ACK.
- Repeating the same held lease after its reserve vanished returns durable success without repairing it; the accepted generation has no task.
_decide_held_serve()(:4217-4231) maps reserveOSErroror malformed JSON to{}and commits anyway. The actual L2 task later queues while the ledger and turn retain L1 room/source identity.
The fix needs immutable reserve objects with an opaque ID and an envelope binding at least (wire, candidate run, lease, payload digest). pending_serve must reference the exact object; switch_run must atomically move that reference into structured owed_publish; reads and deletes must validate and retire only that reference. A per-run filename alone is insufficient because consecutive held leases reuse a candidate run. Add live and restart regressions for the schedules above, including L2→L3→L4 positive control and old-drop/new-hold concurrency against the production writer.
3. A re-served wake bypasses the held-generation protocol
The wake branch of _serve_attempt() (remote_gateway_bridge.py:5764-5811) reuses the run and calls remember_task() directly. It overwrites the lease instead of holding the serve. When the same episode is served as L1/gen1 and then L2/gen2, record_ack() rejects the generation change, _commit_ack() calls it stale, and _post_task_ack() reports success and forgets the retry.
after L1: awaiting_ack=False gen=1 lease=L1
serve L2: awaiting_ack=True gen=1 lease=L2 pending_serve=None
ACK L2/gen2 returned True
after ACK: awaiting_ack=True gen=1 lease=L2 pending_serve=None
This path is PR-activated, and same-episode re-service is required by the PR test suite. Wakes need the same hold/adopt/switch protocol as ordinary runs, with same-generation and newer-generation regression arms.
4. The durable ACK claim is recoverable only through a lossy sidecar index
_read_ack_json() (remote_gateway_bridge.py:3238-3267) maps unreadable state to {}. One transient denied read while _record_pending_acks() adds another task overwrote the prior pending ACK and lease records, while the authoritative ledger still held its open ACK claim:
before: pending={ack1: true}, lease={ack1: L1}, claim=ack(L1)
after one EACCES plus record ack2: pending={ack2: true}, lease={ack2: L2}, claim1 still open
A missing lease sidecar similarly retries with lease=None; _claim_ack() answers elsewhere, the pending index is deleted, and the L1 claim remains forever. _recover_claims() (:4866-4876) only logs ACK claims; _retry_pending_acks() enumerates the sidecar, so it never retries that claim.
Recovery must enumerate ACK work from the authoritative ledger, or make the index tri-state and refuse any overwrite/retirement while unreadable. Regressions need malformed/missing main and lease indexes with an ambiguous ACK, preserved operation ID, an actual retry, and claim/awaiting_ack clearance.
Scope and verdict
I separately checked the conservative branches: unknown top-level ledger at startup, holds, sweep, claim commit/release, _ack_recorded, _mark_published, heartbeat snapshots, and sole-reserve temporary EACCES. They delay or retry and did not yield another terminal decision. The process-only worker-ID failure is pre-existing and the unreadable ack_first manifest caused local discard of an unaccepted stage but did not establish permanent loss; neither is an independent blocker for this PR on current evidence.
Worst case from the four blockers is duplicate model execution or duplicate room answers, an unfenced result closing the wrong generation, wrong-room turn binding, or an accepted task stranded forever. The blast radius is the opt-in lifecycle path, but that is the path this PR is meant to make safe.
One combined fix is welcome. Please include the discriminating tests above, show the new tests were collected, and rerun the exact-head post-restart live witness after the push. I will re-review that frozen head; this one is not yet the final voice-witness candidate.
keweichen
approved these changes
Nov 1, 2026
Collaborator
Author
|
Merging with both reviewers approved at |
Collaborator
|
Implementation proposal for the four blockers on exact head 5c64cae. This is a code plan, not another verdict; the demonstrated schedules and expected outcomes are in my formal review above. I would land this as one lifecycle patch because the reserve owner, wake replay, and ACK recovery all meet in the same attempt transitions. 1. Make every safety-relevant ledger read explicitly tri-stateIn packages/ag2-sparrow/ag2_sparrow/task_attempts.py, add validated private readers for top-level maps and run entries. Do not use data.get(...) or {}, because unreadable, absent, falsey-malformed, and pruned are different states. Keep the disk format unchanged and use the existing UNKNOWN sentinel consistently:
For owed_view(), outer UNKNOWN means no safe current snapshot, so publication waits. Inner None means there was no earlier run. Inner UNKNOWN means earlier_run names evidence that retention pruned, omitted, or malformed; the new run may proceed, but _earlier_attempt() must use the neutral RERUN_RULES["unknown"] wording, never not_delivered/posted_in_room. Route records(), run_of(), run_item(), and wake_seen() through the same validators so [] and other falsey malformed containers cannot become empty state and truthy malformed values cannot raise. Caller rules in remote_gateway_bridge.py:
When an ACK contains an integer execution_generation, set contract.names_lease_generation=true inside the same TaskAttempts _change that records/adopts/switches that ACK. Do not do a read-then-separate contract commit. That makes _ack_recorded() proof of one committed state rather than two independently failing writes. 2. Give every held serve an immutable owner-specific reserveThe task-wide relay-reserves/.json path cannot represent owed G2 and pending G3 simultaneously. Replace new writes with immutable envelopes: {
"format": 1,
"owner": {
"task": "task-X",
"candidate_run": "task-X~~2",
"lease": "L2"
},
"sha256": "<canonical-task-digest>",
"task": {"id": "task-X", "...": "..."}
}Use an opaque/generated filename derived from the validated owner tuple, under relay-reserves/. Create it durably with no replacement; an existing file is idempotent only if the complete envelope and digest match. pending_serve should carry the exact reference: {
"lease": "L2",
"run": "task-X~~2",
"source": "$event",
"wake_episode": "episode-or-null",
"at": 1234.5,
"reserve": {
"id": "v1-<hex>",
"run": "task-X~~2",
"lease": "L2",
"sha256": "<hex>"
}
}On an accepted newer-generation ACK, switch_run() must atomically clear that pending serve and promote the same reference into structured generation debt: {
"owed_publish": {
"run": "task-X~~2",
"generation": 2,
"reserve": {
"id": "v1-<hex>",
"run": "task-X~~2",
"lease": "L2",
"sha256": "<hex>"
}
}
}The required ordering is:
A ledger failure after step 2 leaves an orphan; it must not overwrite the old owner. A crash after steps 3/4 restarts from the exact reference. A queue success before debt clear retries idempotently against the same run. Specific transition changes:
Keep TaskAttempts FORMAT=3 and dual-read existing state. Normalize legacy string owed_publish and pending_serve-without-reserve to the legacy task-wide file. Before consuming legacy bytes, prove ownership from their id and lease_id. If owed and pending both name the legacy file, only the matching owner may consume it; the other generation fails its own dispatch instead of cross-publishing, and the file is not unlinked while either reference remains. New writes always use v1 envelopes. A downgrade may conservatively hold the structured debt, but must never publish it through the old shared path. 3. Route same-episode wake re-serves through that held-generation state machineIn _serve_attempt(), after open_wake() identifies a replay of the active blocker_episode_id, compare leases before the current “already left the queue” replay suppression and before remember_task():
Normalize pending_serve.wake_episode to str | None. hold_serve() receives the episode string, not the entire wake dict; serve_adoptable() compares that string to rec.wake_episode. For compatibility, read a legacy string pending_serve.wake as an alias; a legacy dict is non-adoptable and takes the conservative switch/retry path. Then use the ordinary held-serve decision:
switch_run() must copy pending_serve.wake_episode instead of forcing wake_episode=None. _publish_owed_run() passes it to _write_task(), so the new run receives the saved wake answer, wake_episode header/rule, rerun rule, its new execution generation, and its own distinct outbox item. 4. Make an open ACK claim sufficient for recoverySidecars may remain the pre-claim/downgrade index, but once a TaskAttempts ACK claim exists, the claim is authoritative. Extend new ACK claims with non-identity retry data: {
"retry_owner": "pending_ack",
"local_tid": "<local-run-id>",
"durable": true
}Use retry_owner=ack_first for the held-manifest path. The claim serve remains the exact lease, and claim op remains the broker idempotency key. Split first-send from resend:
Make sidecar reads tri-state:
For pre-change claims without retry data, use a matching readable held manifest to classify ack_first; otherwise treat it as ordinary. Derive local_tid from the claim/current or pending run only when unambiguous. Take durable from a matching sidecar only when its lease agrees; otherwise omit the durability claim. If the held manifest needed to classify the claim is unreadable, defer rather than guess. Keep lifecycle-v2-disabled behavior sidecar-only. On success, commit/release the ledger claim first, then clean sidecars best-effort. Network ambiguity or a local generation-commit failure keeps the same claim/op. Terminal fencing/not-leased follows the existing settlement transition and must not depend on sidecar survival. Discriminating tests to addPlease put policy tests in tests/sparrow-task-attempts.test.py and production-path schedules in tests/gateway-agent-activity-hooks.test.py:
Please show that the new tests are collected, fail on 5c64cae for the intended assertion, and pass at the proposed head. Afterward I will rerun the exact-head audit, including the post-restart lifecycle round trip and the outstanding real voice ack_first witness. |
This was referenced Nov 1, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Not stacked. Its parent, #5883, merged on 2026-10-30 (
850e74cb1). This PR is based onmain(last merged:9ab5350eb). The head is5c64cae64abf8429f25732eecb0c330a0a8e1946.Closes #6085.
Flags, upgrade, size
Both flags default off.
LIFECYCLE_V2 = REMOTE_TASK_LIFECYCLE_V2 == "1", andTURN_CONTEXT = LIFECYCLE_V2 and REMOTE_TASK_TURN_CONTEXT == "1". With both unset, the bridge behaves as on main.src/check-pending-tasks.sh) changes only for a task file carrying awake_episode:header, which only the bridge writes, and only under turn context.In-flight attempts are dropped on upgrade. The attempt records move to one new ledger,
state/relay-task-ledger{suffix}.json(format 3). The earlierrelay-task-attempts*.jsonfiles are never read. Start logs<file> is from an earlier bridge and is ignored; its attempts are not resumed, and leaves the file unchanged. Witnessed live with a non-empty old ledger (section U of the round-18 witness).Final size: 33 files, +8045 −981.
The runtime files include 6
src/modules and their vendored copies underpackages/ag2-sparrow/(sync_from_src --checkis in sync). Most of the runtime change is inremote_gateway_bridge.py(+1488 −354) andtask_attempts.py(+1102 −299).What changed, and why
The resume half of the AgentActivity HITL flow (Rev 2.6 §8.2 steps 3–4, backend contract B13), and the attempt bookkeeping it needs. Everything is behind the flags above.
A wake is a new run. When the owner answers a question, the broker re-serves the task with
wake.resolved_blockers. That wake runs once perblocker_episode_id:<local>~~<n>;<task>.wake-<episode>.A settled episode is never run again.
Ask in the room. Under turn context, each task file carries one writer-owned line telling the agent to ask a blocking question with
room.question.askon its exact brokertask_id(task_attempts.ASK_IN_ROOM_RULE).One writer for attempt state. Every change to a task's attempt goes through
TaskAttempts.transition(wire): a per-task lock, plus a leaf commit under a cross-processflockthat re-reads, checks the ownership epoch, and writes temp → fsync → rename → fsync dir. It fails closed. No provider call is made with a transition held (assert_unlocked).Claim, effect, commit.
Idempotency-Key.A newer generation is a new run.
<task>.run-<n>.new_runassigns it._ack_unlanded). Before the switch, every result the earlier run claimed is settled under its op id: unsent is withdrawn, DELIVERED or PARKED is final, still sending waits.Review first
packages/ag2-sparrow/ag2_sparrow/task_attempts.py:TaskAttempts,Transition(claim,holds_claim,commit_claim,release_claim,note_item,new_run,open_wake,revive_finished,adopt_serve/switch_run),_commit,run_item.packages/ag2-sparrow/ag2_sparrow/remote_gateway_bridge.py:_serve_attempt,_hold_reserve,_claim_ack,_commit_ack,_decide_held_serve;_reconcile_earlier_run,_end_unpublished_claims,_claim_published,_stop_run;_pin_and_claim,_rule_result,_deliver_result_payload,_committed;_on_disk,_outbox_record.docs/remote-gateway-protocol.md, "Task lifecycle (opt-in)".User behaviour this proves
With both flags on, the agent asks the owner a question in the room. The task shows
waiting_input. After the owner answers, the agent resumes with the answer as a new turn and its reply is delivered once. This holds across a bridge restart, a Stop, a crash after the broker accepted an ack or a result, and a re-serve under a newer generation. With the flags off, nothing changes.Docs
docs/remote-gateway-protocol.md, "Task lifecycle (opt-in)", is updated, and itsdocs/catalog.jsonlast_verifiedis 2026-11-01.CHANGELOG-PENDING.mddescribes the ledger and the upgrade drop.Before / after
Before: parent
a3efa9b6f(main) with only this PR's test files (and the rig launcher they load) copied in:After: this head:
Each review round's blocker has its own before/after (its regression failing at the previous head, passing at the fix) in the round replies: round 14, 15, 16, 17, 19.
Tests run
tests/gateway-disposal-delegation.test.py(85 OK) andtests/hermetic-env-structural.test.py(28 OK).scripts/discover-python-tests.shlists (1,320), 6 at a time, each under a timeout. 19 fail, and all 19 fail identically onmain(environment on the author's Mac). None are touched here.sync_from_src --check: in sync.git diff --check: clean.review-checks: PASS.ruff --select F: clean.5c64cae6: 47 passed, 3 skipped, 0 failed.Live round trip
Run by Mark's coordinator agent on Mark's Mac: a local k3d rig with a real broker and both flags on, after a plain restart. The raw output is in the round-18 witness at
295c9d1a4. It covers:scripts/witness/lifecycle-fault-launcher.py(rig-only, never imported by production) injects the stall and kill faults. Itsreclaimfault drives the stale-generation schedule against the real broker: it withholds the task's beats until the broker reclaims the lease, and keeps the newer serve back until generation 1's answer has had its real 409.Live status.
e66aea13, the coordinator's rig passed the ordinary post-restart round trip and every live section except G.8e87d036, the stale-generation schedule (G) passed every bridge step against the real broker: a real 409, L2 delivered and held,~~2queued once. But the core then took~~2for a redelivery and sent nothing. Fixed ata619c219: the new run's task file says what became of the earlier answer.Still owed at this head:
ack_firstwith a named lease.Edge cases checked
Each has a production-path regression:
Risks
relay-task-attempts*.json.publish-sparrow.ymlis a separate step.Known gaps and follow-ups
Non-blocking per the reviewers, and all behind the two default-off flags. Not fixed in this PR:
ack_first(voice relay-ask) wake is left unacked and logged once per episode, so a voice-path question doesn't resume in this release.relay-task-attempts*.json); start logs it.inflightreads iterate without the mutex.flockis caught only by the lock-fault test._stop_run: an unreadable own-run item counts as answered (its result is quarantined anyway).answered) is pinned by the bridge-side tests only. A stronger pin that the core then sends no duplicate needs a real model._task_pending,_delivered_copy_existsand_archived_task_fileuseglob/exists, so an unreadable directory reads as empty. In this PR they're only OR-ed with tri-state reads. Every one of them is keyed on a local run id, and a new lease never lands on a finished run, so evidence about one generation can't swallow another.Hardcoded paths
review-checks(hardcoded-paths, root-artifacts, prose-cap): PASS. Added lines useRESULTS_DIR,TASKS_DIR,_STATEand the workspace helpers; test fixtures use temp dirs.