fix(bot-mode): carry late live DM replies to sender
(cherry picked from commit c274db6ea52504fa661339dd5d1db141db1cfb65)
This commit is contained in:
z0zero
committed by
Brooklyn Nicholson
Brooklyn Nicholson
parent
cb375015e3
commit
a94b975831
@@ -54,6 +54,34 @@ def test_delivery_is_idempotent_fenced_and_permanent(tmp_path, terminal_status):
|
||||
for path in (tmp_path / "runtime" / mailbox.DELIVERY_DIR_NAME).iterdir():
|
||||
assert path.stat().st_mode & 0o077 == 0
|
||||
|
||||
|
||||
def test_cancel_queued_delivery_fences_claim_and_preserves_terminal_result(tmp_path):
|
||||
from tools import bot_live_delivery as mailbox
|
||||
|
||||
owner = dict(profile_home=str(tmp_path.resolve()), session_id="chat",
|
||||
lease_id="lease", live_session_id="live")
|
||||
queued = mailbox.deliver_to_live_owner(tmp_path, owner, "first", delivery_id="3" * 32)
|
||||
cancelled = mailbox.cancel_queued_delivery(
|
||||
tmp_path, queued["delivery_id"], error="owner closed", reason="runtime_offline")
|
||||
assert cancelled is not None
|
||||
assert cancelled["status"] == "cancelled"
|
||||
assert cancelled["error"] == "owner closed"
|
||||
assert mailbox.claim_pending_delivery(tmp_path, owner) is None
|
||||
assert mailbox.deliver_to_live_owner(tmp_path, owner, "first", delivery_id="3" * 32) == cancelled
|
||||
assert mailbox.cancel_queued_delivery(
|
||||
tmp_path, queued["delivery_id"], error="owner closed", reason="runtime_offline") == cancelled
|
||||
|
||||
queued = mailbox.deliver_to_live_owner(tmp_path, owner, "second", delivery_id="4" * 32)
|
||||
claimed = mailbox.claim_pending_delivery(tmp_path, owner)
|
||||
assert claimed is not None
|
||||
assert claimed["delivery_id"] == queued["delivery_id"]
|
||||
assert mailbox.cancel_queued_delivery(
|
||||
tmp_path, queued["delivery_id"], error="owner closed", reason="runtime_offline") == claimed
|
||||
settled = mailbox.complete_delivery(tmp_path, queued["delivery_id"], status="settled", reply="PONG")
|
||||
assert mailbox.cancel_queued_delivery(
|
||||
tmp_path, queued["delivery_id"], error="owner closed", reason="runtime_offline") == settled
|
||||
|
||||
|
||||
@pytest.mark.parametrize("intent_state", ["new", "existing", "raced"])
|
||||
def test_live_dm_bom_readers_preserve_pinned_intent(tmp_path, monkeypatch, intent_state):
|
||||
from pathlib import Path
|
||||
|
||||
@@ -617,22 +617,32 @@ def test_live_dm_runner_retry_never_reexecutes_failed_claim(tmp_path, monkeypatc
|
||||
monkeypatch.setenv("HERMES_HOME", str(home))
|
||||
owner = dict(profile_home=str(target), session_id="bot", lease_id="lease", live_session_id="live")
|
||||
monkeypatch.setattr(live, "find_canonical_live_owner", lambda h: owner)
|
||||
monkeypatch.setattr(bot_mode_dm, "_LIVE_WAIT_SECONDS", 0)
|
||||
monkeypatch.setattr(bot_mode_dm, "_LIVE_WAIT_SECONDS", 0.01)
|
||||
monkeypatch.setattr(live, "_POLL_SECONDS", 0.01)
|
||||
monkeypatch.setattr(subprocess, "run", lambda *a, **k: pytest.fail("must not launch a model turn"))
|
||||
dm_file = tmp_path / "message.txt"
|
||||
dm_file.write_text("hello", encoding="utf-8")
|
||||
argv = ["hermes", "-p", "researcher"]
|
||||
assert bot_mode_dm._run_delivery(argv, str(dm_file), stdin_file=False) == 0
|
||||
queued = json.loads(capsys.readouterr().out)
|
||||
assert queued["status"] == "queued"
|
||||
claimed = live.claim_pending_delivery(target, owner)
|
||||
assert claimed is not None
|
||||
live.complete_delivery(target, claimed["delivery_id"], status="failed", error="HTTP 429 rate limit")
|
||||
|
||||
def fail_later():
|
||||
time.sleep(0.05)
|
||||
claimed = live.claim_pending_delivery(target, owner)
|
||||
assert claimed is not None
|
||||
live.complete_delivery(target, claimed["delivery_id"], status="failed", error="HTTP 429 rate limit")
|
||||
|
||||
failing = threading.Thread(target=fail_later)
|
||||
failing.start()
|
||||
try:
|
||||
assert bot_mode_dm._run_delivery(argv, str(dm_file), stdin_file=False) == 1
|
||||
finally:
|
||||
failing.join(timeout=2)
|
||||
first = json.loads(capsys.readouterr().out)
|
||||
assert first["status"] == "failed"
|
||||
monkeypatch.setattr(live, "find_canonical_live_owner", lambda h: None)
|
||||
assert bot_mode_dm._run_delivery(argv, str(dm_file), stdin_file=False) == 1
|
||||
failed = json.loads(capsys.readouterr().out)
|
||||
assert failed["status"] == "failed"
|
||||
assert failed["delivery_id"] == queued["delivery_id"]
|
||||
assert failed["delivery_id"] == first["delivery_id"]
|
||||
assert dm_file.read_text(encoding="utf-8") == "hello"
|
||||
|
||||
|
||||
@@ -642,8 +652,9 @@ def test_live_dm_wait_reports_reply_after_initial_wait_budget(tmp_path, monkeypa
|
||||
|
||||
owner = dict(profile_home=str(tmp_path), session_id="bot", lease_id="lease", live_session_id="live")
|
||||
record = live.deliver_to_live_owner(tmp_path, owner, "ping", delivery_id="a" * 32)
|
||||
monkeypatch.setattr(bot_mode_dm, "_LIVE_WAIT_SECONDS", 0)
|
||||
monkeypatch.setattr(bot_mode_dm, "_LIVE_WAIT_SECONDS", 0.01)
|
||||
monkeypatch.setattr(live, "_POLL_SECONDS", 0.01)
|
||||
monkeypatch.setattr(live, "find_canonical_live_owner", lambda h: owner)
|
||||
|
||||
def settle_later():
|
||||
time.sleep(0.05)
|
||||
@@ -664,6 +675,135 @@ def test_live_dm_wait_reports_reply_after_initial_wait_budget(tmp_path, monkeypa
|
||||
assert notice["reply"] == "PONG"
|
||||
|
||||
|
||||
def test_live_dm_wait_releases_queued_runner_when_pinned_owner_is_gone(tmp_path, monkeypatch, capsys):
|
||||
from tools import bot_live_delivery as live
|
||||
|
||||
owner = dict(profile_home=str(tmp_path), session_id="bot", lease_id="lease", live_session_id="live")
|
||||
record = live.deliver_to_live_owner(tmp_path, owner, "ping", delivery_id="c" * 32)
|
||||
monkeypatch.setattr(bot_mode_dm, "_LIVE_WAIT_SECONDS", 0.01)
|
||||
monkeypatch.setattr(live, "_POLL_SECONDS", 0.01)
|
||||
monkeypatch.setattr(live, "find_canonical_live_owner", lambda h: None)
|
||||
|
||||
assert bot_mode_dm._wait_live_dm(str(tmp_path), record["delivery_id"]) == 1
|
||||
notice = json.loads(capsys.readouterr().out)
|
||||
assert notice["status"] == "cancelled"
|
||||
final = live.read_delivery_result(tmp_path, record["delivery_id"])
|
||||
assert final is not None and final["status"] == "cancelled"
|
||||
|
||||
|
||||
def test_live_dm_wait_rechecks_receipt_when_owner_disappears_at_claim(tmp_path, monkeypatch, capsys):
|
||||
from tools import bot_live_delivery as live
|
||||
|
||||
owner = dict(profile_home=str(tmp_path), session_id="bot", lease_id="lease", live_session_id="live")
|
||||
record = live.deliver_to_live_owner(tmp_path, owner, "ping", delivery_id="f" * 32)
|
||||
monkeypatch.setattr(bot_mode_dm, "_LIVE_WAIT_SECONDS", 0.01)
|
||||
monkeypatch.setattr(live, "_POLL_SECONDS", 0.01)
|
||||
|
||||
def settle_during_owner_check(_home):
|
||||
claimed = live.claim_pending_delivery(tmp_path, owner)
|
||||
assert claimed is not None
|
||||
live.complete_delivery(tmp_path, claimed["delivery_id"], status="settled", reply="PONG")
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(live, "find_canonical_live_owner", settle_during_owner_check)
|
||||
assert bot_mode_dm._wait_live_dm(str(tmp_path), record["delivery_id"]) == 0
|
||||
notice = json.loads(capsys.readouterr().out)
|
||||
assert notice["status"] == "settled"
|
||||
assert notice["reply"] == "PONG"
|
||||
|
||||
|
||||
def test_live_dm_wait_does_not_cancel_a_claim_after_the_last_read(tmp_path, monkeypatch, capsys):
|
||||
from tools import bot_live_delivery as live
|
||||
|
||||
owner = dict(profile_home=str(tmp_path), session_id="bot", lease_id="lease", live_session_id="live")
|
||||
record = live.deliver_to_live_owner(tmp_path, owner, "ping", delivery_id="2" * 32)
|
||||
monkeypatch.setattr(bot_mode_dm, "_LIVE_WAIT_SECONDS", 0.01)
|
||||
monkeypatch.setattr(live, "_POLL_SECONDS", 0.01)
|
||||
owner_checked = False
|
||||
|
||||
def owner_gone(_home):
|
||||
nonlocal owner_checked
|
||||
owner_checked = True
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(live, "find_canonical_live_owner", owner_gone)
|
||||
original_read = live.read_delivery_result
|
||||
claimed_during_read = False
|
||||
|
||||
def read_then_claim(home, delivery_id):
|
||||
nonlocal claimed_during_read
|
||||
snapshot = original_read(home, delivery_id)
|
||||
assert snapshot is not None
|
||||
if owner_checked and snapshot["status"] == "queued" and not claimed_during_read:
|
||||
claimed_during_read = True
|
||||
claimed = live.claim_pending_delivery(tmp_path, owner)
|
||||
assert claimed is not None
|
||||
live.complete_delivery(tmp_path, delivery_id, status="settled", reply="PONG")
|
||||
return snapshot
|
||||
|
||||
monkeypatch.setattr(live, "read_delivery_result", read_then_claim)
|
||||
assert bot_mode_dm._wait_live_dm(str(tmp_path), record["delivery_id"]) == 0
|
||||
notice = json.loads(capsys.readouterr().out)
|
||||
assert claimed_during_read
|
||||
assert notice["status"] == "settled"
|
||||
assert notice["reply"] == "PONG"
|
||||
|
||||
|
||||
def test_live_dm_wait_keeps_claimed_turn_until_it_settles(tmp_path, monkeypatch, capsys):
|
||||
from tools import bot_live_delivery as live
|
||||
|
||||
owner = dict(profile_home=str(tmp_path), session_id="bot", lease_id="lease", live_session_id="live")
|
||||
record = live.deliver_to_live_owner(tmp_path, owner, "ping", delivery_id="d" * 32)
|
||||
assert live.claim_pending_delivery(tmp_path, owner) is not None
|
||||
monkeypatch.setattr(bot_mode_dm, "_LIVE_WAIT_SECONDS", 0.03)
|
||||
monkeypatch.setattr(live, "_POLL_SECONDS", 0.01)
|
||||
monkeypatch.setattr(live, "find_canonical_live_owner", lambda h: None)
|
||||
|
||||
def settle_later():
|
||||
time.sleep(0.05)
|
||||
live.complete_delivery(tmp_path, record["delivery_id"], status="settled", reply="PONG")
|
||||
|
||||
settling = threading.Thread(target=settle_later)
|
||||
settling.start()
|
||||
try:
|
||||
exit_code = bot_mode_dm._wait_live_dm(str(tmp_path), record["delivery_id"])
|
||||
finally:
|
||||
settling.join(timeout=2)
|
||||
|
||||
assert exit_code == 0
|
||||
notice = json.loads(capsys.readouterr().out)
|
||||
assert notice["status"] == "settled"
|
||||
assert notice["reply"] == "PONG"
|
||||
|
||||
|
||||
def test_live_dm_wait_releases_claimed_runner_after_owner_is_gone(tmp_path, monkeypatch, capsys):
|
||||
from tools import bot_live_delivery as live
|
||||
|
||||
owner = dict(profile_home=str(tmp_path), session_id="bot", lease_id="lease", live_session_id="live")
|
||||
record = live.deliver_to_live_owner(tmp_path, owner, "ping", delivery_id="1" * 32)
|
||||
assert live.claim_pending_delivery(tmp_path, owner) is not None
|
||||
monkeypatch.setattr(bot_mode_dm, "_LIVE_WAIT_SECONDS", 0.01)
|
||||
monkeypatch.setattr(live, "_POLL_SECONDS", 0.01)
|
||||
monkeypatch.setattr(live, "find_canonical_live_owner", lambda h: None)
|
||||
|
||||
outcomes = []
|
||||
waiting = threading.Thread(target=lambda: outcomes.append(bot_mode_dm._wait_live_dm(
|
||||
str(tmp_path), record["delivery_id"])), daemon=True)
|
||||
waiting.start()
|
||||
waiting.join(timeout=1)
|
||||
stopped_with_unknown_outcome = not waiting.is_alive()
|
||||
if not stopped_with_unknown_outcome:
|
||||
live.complete_delivery(tmp_path, record["delivery_id"], status="failed", error="test cleanup")
|
||||
waiting.join(timeout=2)
|
||||
|
||||
assert stopped_with_unknown_outcome, "a vanished owner must not retain the runner indefinitely"
|
||||
assert outcomes == [0]
|
||||
notice = json.loads(capsys.readouterr().out)
|
||||
assert notice["status"] == "claimed"
|
||||
final = live.read_delivery_result(tmp_path, record["delivery_id"])
|
||||
assert final is not None and final["status"] == "claimed"
|
||||
|
||||
|
||||
# ── plaintext tempfile lifecycle ─────────────────────────────────────────────
|
||||
|
||||
|
||||
@@ -1083,15 +1223,26 @@ def test_settled_live_wait_unlinks_the_intent_but_a_pending_one_keeps_it(tmp_pat
|
||||
dm_file.write_text("secret plaintext", encoding="utf-8")
|
||||
intent = tmp_path / "dm-x.txt.live.json"
|
||||
intent.write_text("{}", encoding="utf-8")
|
||||
monkeypatch.setattr(bot_mode_dm, "_LIVE_WAIT_SECONDS", 0)
|
||||
owner = dict(profile_home=str(tmp_path), session_id="bot", lease_id="lease", live_session_id="live")
|
||||
record = live.deliver_to_live_owner(tmp_path, owner, "ping", delivery_id="b" * 32)
|
||||
monkeypatch.setattr(bot_mode_dm, "_LIVE_WAIT_SECONDS", 0.01)
|
||||
monkeypatch.setattr(live, "_POLL_SECONDS", 0.01)
|
||||
monkeypatch.setattr(live, "find_canonical_live_owner", lambda h: owner)
|
||||
|
||||
monkeypatch.setattr(live, "read_delivery_result", lambda home, did: {"status": "queued"})
|
||||
assert bot_mode_dm._wait_live_dm(str(tmp_path), "d1", dm_file=dm_file) == 0
|
||||
assert intent.exists(), "a pending delivery may still be retried from the same intent"
|
||||
assert dm_file.exists()
|
||||
outcomes = []
|
||||
waiting = threading.Thread(target=lambda: outcomes.append(bot_mode_dm._wait_live_dm(
|
||||
str(tmp_path), record["delivery_id"], dm_file=dm_file)), daemon=True)
|
||||
waiting.start()
|
||||
time.sleep(0.05)
|
||||
pending_kept = waiting.is_alive() and intent.exists() and dm_file.exists()
|
||||
claimed = live.claim_pending_delivery(tmp_path, owner)
|
||||
assert claimed is not None
|
||||
live.complete_delivery(tmp_path, claimed["delivery_id"], status="settled", reply="ok")
|
||||
waiting.join(timeout=2)
|
||||
|
||||
monkeypatch.setattr(live, "read_delivery_result", lambda home, did: {"status": "settled", "reply": "ok"})
|
||||
assert bot_mode_dm._wait_live_dm(str(tmp_path), "d1", dm_file=dm_file) == 0
|
||||
assert pending_kept, "a pending delivery keeps its intent while the runner waits"
|
||||
assert not waiting.is_alive()
|
||||
assert outcomes == [0]
|
||||
assert not intent.exists()
|
||||
assert not dm_file.exists(), "the dm .txt holds the same plaintext as the settled intent"
|
||||
|
||||
|
||||
@@ -292,6 +292,25 @@ def complete_delivery(
|
||||
return record
|
||||
|
||||
|
||||
def cancel_queued_delivery(
|
||||
profile_home: Path | str, delivery_id: str, *, error: str, reason: str,
|
||||
) -> dict[str, Any] | None:
|
||||
"""Cancel an ownerless queued receipt without racing a consumer's claim.
|
||||
|
||||
A claim or terminal result that won the mailbox lock is returned unchanged.
|
||||
"""
|
||||
key = _delivery_id(delivery_id)
|
||||
with _locked(profile_home) as root:
|
||||
path = root / f"{key}.json"
|
||||
record = _read(path)
|
||||
if record is None or record["status"] != "queued":
|
||||
return record
|
||||
record.update(status="cancelled", reply="", error=error, reason=reason,
|
||||
completed_at=time.time_ns())
|
||||
_write(path, record)
|
||||
return record
|
||||
|
||||
|
||||
def read_delivery_result(profile_home: Path | str, delivery_id: str) -> dict[str, Any] | None:
|
||||
"""Read admission/claim/terminal state without waiting or deleting its receipt."""
|
||||
return _read(_root(profile_home) / f"{_delivery_id(delivery_id)}.json")
|
||||
|
||||
@@ -518,9 +518,46 @@ def _admit_live_dm(profile_home: Path | None, dm_file: str, author: Optional[dic
|
||||
|
||||
|
||||
def _wait_live_dm(home: str, delivery_id: str, *, dm_file: "str | os.PathLike | None" = None) -> int:
|
||||
from tools.bot_live_delivery import await_delivery
|
||||
from tools.bot_failure_reasons import RUNTIME_OFFLINE
|
||||
from tools.bot_live_delivery import (await_delivery, cancel_queued_delivery,
|
||||
find_canonical_live_owner, read_delivery_result)
|
||||
|
||||
record = await_delivery(home, delivery_id, _LIVE_WAIT_SECONDS)
|
||||
owner_missing_claimed = False
|
||||
while True:
|
||||
record = await_delivery(home, delivery_id, _LIVE_WAIT_SECONDS)
|
||||
if record is None or record["status"] not in ("queued", "claimed"):
|
||||
break
|
||||
try:
|
||||
owner = find_canonical_live_owner(home)
|
||||
except Exception:
|
||||
logger.debug("Could not check live DM owner; keep waiting for receipt", exc_info=True)
|
||||
continue
|
||||
pinned = record["owner"]
|
||||
if owner is not None and all(owner[key] == pinned[key] for key in
|
||||
("profile_home", "lease_id", "live_session_id")):
|
||||
owner_missing_claimed = False
|
||||
continue
|
||||
# A claim or settlement can race the owner lookup. Decide from a fresh
|
||||
# receipt so the runner cannot exit on a stale queued snapshot.
|
||||
record = read_delivery_result(home, delivery_id)
|
||||
if record is None or record["status"] not in ("queued", "claimed"):
|
||||
break
|
||||
if record["status"] == "queued":
|
||||
# The consumer may still have a captured owner and race this read.
|
||||
# Under the same mailbox lock as claim, either cancellation wins or
|
||||
# the claim wins and we must keep waiting for its outcome.
|
||||
record = cancel_queued_delivery(
|
||||
home, delivery_id, error="Live Bot Chat owner is no longer available",
|
||||
reason=RUNTIME_OFFLINE)
|
||||
if record is None or record["status"] != "claimed":
|
||||
break
|
||||
owner_missing_claimed = False
|
||||
continue
|
||||
if owner_missing_claimed:
|
||||
# A vanished consumer may leave a permanent claim. Allow one more
|
||||
# wait window for an in-flight turn to settle, then release the runner.
|
||||
break
|
||||
owner_missing_claimed = True
|
||||
status = record["status"] if record else "ambiguous"
|
||||
payload = {key: record[key] for key in ("reply", "error", "reason") if record and record.get(key)}
|
||||
payload.update(status=status, delivery_id=delivery_id)
|
||||
|
||||
Reference in New Issue
Block a user