diff --git a/docs/adr/0111-peer-inbox-retains-bodies-and-reports-loss.md b/docs/adr/0111-peer-inbox-retains-bodies-and-reports-loss.md new file mode 100644 index 00000000..2ec25114 --- /dev/null +++ b/docs/adr/0111-peer-inbox-retains-bodies-and-reports-loss.md @@ -0,0 +1,31 @@ +# 0111. Peer inbox retains bodies and reports loss + +Status: accepted + +## Context + +The pull inbox stored clipped previews, silently evicted older messages, and +cleared unread state even when allocating the read result failed (#865). +A successful-looking read could therefore omit instructions or entire messages. + +## Decision + +- Keep the eight-message ring and advisory, bounded history wake from ADR 0004. + Store owned, complete sender and body strings independently of delivery arenas. + Only the wake may abbreviate a sender; inbox reads preserve message bytes. +- Count evictions and messages that cannot be allocated. Include the count in + wakes and reads, and retain existing entries if allocating a new one fails. +- Build the entire inbox result before clearing entries or loss counts. A failed + read returns an error and leaves the mailbox available for retry. +- Persist the loss count beside the existing `peer_inbox` array and include it + in the save fingerprint. Legacy snapshots default to zero; oversized restored + arrays add their evictions to the saved count. Resume wakes for loss-only state + too. One-shot sessions continue to discard the mailbox (ADR 0014). + +## Consequences + +Memory remains bounded by message count rather than fixed preview bytes; long +messages cost more memory but remain readable. Evicted bodies remain in the +room log, not an unbounded in-memory queue. This preserves the pull design +without silently presenting a partial mailbox as complete. Old snapshots cannot +recover text already clipped before saving. diff --git a/docs/adr/README.md b/docs/adr/README.md index 8cb1e80b..999740e0 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -121,6 +121,7 @@ record only when you need the evidence or the edge cases. | [0108](0108-browser-events-outrank-pending-snapshots.md) | Browser page-info events invalidate older pending IPC snapshots; a loaded page must not revert to loading. | | [0109](0109-resume-retains-catalog-and-layout-snapshots.md) | Resume restores loaded tool selections and the same-workspace layout snapshot, without freezing current instructions or granting permissions. | | [0110](0110-project-mcp-listing-opts-in-optional-servers.md) | Project MCP entries opt in optional servers; inherited extras still require opt-in, and startup consent applies to both. | +| [0111](0111-peer-inbox-retains-bodies-and-reports-loss.md) | Peer inbox owns complete bodies, reports and persists loss, and clears only after a complete read result is allocated. | ## When to write one diff --git a/evals/harness_behavior.jsonl b/evals/harness_behavior.jsonl index f80d09f0..9e1f0e8d 100644 --- a/evals/harness_behavior.jsonl +++ b/evals/harness_behavior.jsonl @@ -80,3 +80,5 @@ {"id": "publication-unresolved-command-refused", "why": "#847: a non-literal publication command cannot bypass head/body inspection by chaining checks before creation. No GitHub write or lookup should execute.", "args": ["--old"], "env": {"GRAFF_NO_CODEDB_GUARD": "1"}, "prompt": "Exercise the scripted prepublication refusal.", "script": [{"tool": "bash", "arguments": {"command": "gh pr checks --watch && gh pr create --title fixture --body '## Verification: local passed'"}}, {"text": "The publication was blocked."}], "assert": [{"event": {"type": "tool_result", "name": "bash", "is_error": true}}, {"request_contains": {"text": "separate literal gh pr command"}}, {"final_text_contains": "The publication was blocked."}]} {"id": "render-html-hands-back-an-opaque-view", "why": "The model's own drawing surface must be advertised, dispatch to the snapshot writer, and put only the opaque path into the transcript - a page the model drew must never come back to it as a second copy of the markup.", "args": [], "prompt": "draw the retry flow as a page", "script": [{"tool": "render_html", "arguments": {"html": "

Retry flow

the second attempt is declined

"}}, {"text": "drawn above"}], "assert": [{"request_contains": {"index": 0, "text": "render_html"}}, {"events_at_least": {"match": {"type": "tool_call", "name": "render_html"}, "count": 1}}, {"event": {"type": "tool_result"}}, {"request_contains": {"index": -1, "text": "[Rendered view]"}}, {"request_contains": {"index": -1, "text": "/.graff/views/"}}, {"final_text_contains": "drawn above"}]} {"id": "bundled-mcp-skill-loads", "why": "The bundled invocation skill must be discoverable and load its connection instructions after adding an MCP server.", "prompt": "Load the bundled mcp skill before adding a server.", "script": [{"tool": "skill", "arguments": {"name": "mcp"}}, {"text": "MCP instructions loaded."}], "assert": [{"event": {"type": "tool_call", "name": "skill"}}, {"request_contains": {"index": -1, "text": "graff mcp add"}}, {"request_contains": {"index": -1, "text": "/mcp trust"}}, {"final_text_contains": "MCP instructions loaded."}]} +{"id": "peer-inbox-preserves-live-body-865", "why": "#865: inject a device-room user message after the first drain; full body must survive deliverInbound and inbox tool dispatch without entering the wake.", "prompt": "Run the fixture injection, then read the peer inbox twice.", "script": [{"tool": "bash", "arguments": {"command": "python3 -c 'import json,os,pathlib; p=pathlib.Path(os.environ['\"'\"'HOME'\"'\"'])/'\"'\"'.graff/live/chan-all.jsonl'\"'\"'; p.parent.mkdir(parents=True,exist_ok=True); f=p.open('\"'\"'a'\"'\"'); f.write(json.dumps({'\"'\"'from_pid'\"'\"':2147483000,'\"'\"'from_session'\"'\"':'\"'\"'fixture-peer'\"'\"','\"'\"'from_user'\"'\"':True,'\"'\"'text'\"'\"':'\"'\"'prefix full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload BODY_END_865'\"'\"'})+'\"'\"'\\n'\"'\"'); f.close()'"}}, {"tool": "peer_message", "arguments": {"action": "inbox"}}, {"tool": "peer_message", "arguments": {"action": "inbox"}}, {"text": "fixture complete"}], "assert": [{"request_lacks": {"index": 1, "text": "[peer message from fixture-peer"}}, {"request_contains": {"index": 2, "text": "[peer message from fixture-peer"}}, {"request_contains": {"index": 2, "text": "[peer message from fixture-peer \\u00b7 device]: prefix full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload full body payload BODY_END_865"}}, {"request_contains": {"index": 3, "text": "inbox empty"}}, {"no_event": {"type": "tool_call_finished", "name": "peer_message", "is_error": true}}]} +{"id": "peer-inbox-live-multidrain-overflow-unicode-clear-865", "why": "#865: two live device-room drains overflow the eight-slot ring cumulatively (10 then 5: 2 then 7 lost); exact tool results prove retained 7..14 order and full 211-character Unicode bodies/device flags, read clears messages and loss, and later arrival has no stale loss. Assertions bind complete tool-result objects to call IDs to exclude scripted-command/history false positives.", "prompt": "Inject the two fixture batches, read the inbox twice, inject the fresh fixture, and read twice again.", "script": [{"tool": "bash", "arguments": {"command": "python3 -c 'import json,os,pathlib; p=pathlib.Path(os.environ['\"'\"'HOME'\"'\"'])/'\"'\"'.graff/live/chan-all.jsonl'\"'\"'; p.parent.mkdir(parents=True,exist_ok=True); f=p.open('\"'\"'a'\"'\"'); [f.write(json.dumps({'\"'\"'from_pid'\"'\"':2147483000,'\"'\"'from_session'\"'\"':'\"'\"'overflow-fixture'\"'\"','\"'\"'from_user'\"'\"':True,'\"'\"'text'\"'\"':'\"'\"'item-%02d:'\"'\"'%i+chr(0x754c)*211+'\"'\"':tail-%02d'\"'\"'%i})+'\"'\"'\\n'\"'\"') for i in range(0,10)]; f.close()'"}}, {"tool": "bash", "arguments": {"command": "python3 -c 'import json,os,pathlib; p=pathlib.Path(os.environ['\"'\"'HOME'\"'\"'])/'\"'\"'.graff/live/chan-all.jsonl'\"'\"'; p.parent.mkdir(parents=True,exist_ok=True); f=p.open('\"'\"'a'\"'\"'); [f.write(json.dumps({'\"'\"'from_pid'\"'\"':2147483000,'\"'\"'from_session'\"'\"':'\"'\"'overflow-fixture'\"'\"','\"'\"'from_user'\"'\"':True,'\"'\"'text'\"'\"':'\"'\"'item-%02d:'\"'\"'%i+chr(0x754c)*211+'\"'\"':tail-%02d'\"'\"'%i})+'\"'\"'\\n'\"'\"') for i in range(10,15)]; f.close()'"}}, {"tool": "peer_message", "arguments": {"action": "inbox"}}, {"tool": "peer_message", "arguments": {"action": "inbox"}}, {"tool": "bash", "arguments": {"command": "python3 -c 'import json,os,pathlib; p=pathlib.Path(os.environ['\"'\"'HOME'\"'\"'])/'\"'\"'.graff/live/chan-all.jsonl'\"'\"'; p.parent.mkdir(parents=True,exist_ok=True); f=p.open('\"'\"'a'\"'\"'); [f.write(json.dumps({'\"'\"'from_pid'\"'\"':2147483000,'\"'\"'from_session'\"'\"':'\"'\"'overflow-fixture'\"'\"','\"'\"'from_user'\"'\"':True,'\"'\"'text'\"'\"':'\"'\"'item-%02d:'\"'\"'%i+chr(0x754c)*211+'\"'\"':tail-%02d'\"'\"'%i})+'\"'\"'\\n'\"'\"') for i in range(15,16)]; f.close()'"}}, {"tool": "peer_message", "arguments": {"action": "inbox"}}, {"tool": "peer_message", "arguments": {"action": "inbox"}}, {"text": "fixture complete"}], "assert": [{"request_contains": {"index": 1, "text": "[peer] 8 unread; 2 dropped"}}, {"request_contains": {"index": 2, "text": "[peer] 8 unread; 7 dropped"}}, {"request_contains": {"index": 3, "text": "{\"role\": \"tool\", \"tool_call_id\": \"call_3\", \"content\": \"[peer inbox: 7 dropped message(s) could not be retained; the room log keeps the originals]\\n[peer message from overflow-fixture \\u00b7 device]: item-07:\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c:tail-07\\n[peer message from overflow-fixture \\u00b7 device]: item-08:\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c:tail-08\\n[peer message from overflow-fixture \\u00b7 device]: item-09:\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c:tail-09\\n[peer message from overflow-fixture \\u00b7 device]: item-10:\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c:tail-10\\n[peer message from overflow-fixture \\u00b7 device]: item-11:\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c:tail-11\\n[peer message from overflow-fixture \\u00b7 device]: item-12:\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c:tail-12\\n[peer message from overflow-fixture \\u00b7 device]: item-13:\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c:tail-13\\n[peer message from overflow-fixture \\u00b7 device]: item-14:\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c:tail-14\\n(inbox cleared \\u2014 reply with peer_message; omit session for the room, or name one peer to DM)\"}"}}, {"request_contains": {"index": 4, "text": "{\"role\": \"tool\", \"tool_call_id\": \"call_4\", \"content\": \"inbox empty\"}"}}, {"request_contains": {"index": 6, "text": "{\"role\": \"tool\", \"tool_call_id\": \"call_6\", \"content\": \"[peer message from overflow-fixture \\u00b7 device]: item-15:\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c:tail-15\\n(inbox cleared \\u2014 reply with peer_message; omit session for the room, or name one peer to DM)\"}"}}, {"request_contains": {"index": 7, "text": "{\"role\": \"tool\", \"tool_call_id\": \"call_7\", \"content\": \"inbox empty\"}"}}, {"events_at_least": {"match": {"type": "tool_call_finished", "name": "peer_message", "is_error": false}, "count": 4}}, {"no_event": {"type": "tool_call_finished", "is_error": true}}, {"request_lacks": {"index": 1, "text": "[peer message from overflow-fixture"}}, {"request_lacks": {"index": 2, "text": "[peer message from overflow-fixture"}}, {"request_lacks": {"index": 3, "text": "[peer message from overflow-fixture \\u00b7 device]: item-00:\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c:tail-00\\n"}}, {"request_lacks": {"index": 3, "text": "[peer message from overflow-fixture \\u00b7 device]: item-01:\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c:tail-01\\n"}}, {"request_lacks": {"index": 3, "text": "[peer message from overflow-fixture \\u00b7 device]: item-02:\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c:tail-02\\n"}}, {"request_lacks": {"index": 3, "text": "[peer message from overflow-fixture \\u00b7 device]: item-03:\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c:tail-03\\n"}}, {"request_lacks": {"index": 3, "text": "[peer message from overflow-fixture \\u00b7 device]: item-04:\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c:tail-04\\n"}}, {"request_lacks": {"index": 3, "text": "[peer message from overflow-fixture \\u00b7 device]: item-05:\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c:tail-05\\n"}}, {"request_lacks": {"index": 3, "text": "[peer message from overflow-fixture \\u00b7 device]: item-06:\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c\\u754c:tail-06\\n"}}]} diff --git a/src/main.zig b/src/main.zig index a95fede8..4ffb6d0b 100644 --- a/src/main.zig +++ b/src/main.zig @@ -591,7 +591,7 @@ test { // ── Unit tests (`zig build test`): pull in tests from imported modu _ = @import("goal_flow.zig"); _ = @import("goal_todo.zig"); _ = @import("goal_pacing.zig"); - _ = @import("presence_record.zig"); + _ = .{ @import("presence_record.zig"), @import("peer_inbox_content_test.zig"), @import("peer_inbox_failure_test.zig"), @import("peer_inbox_storage_test.zig"), @import("peer_inbox_sequence_test.zig"), @import("peer_inbox_snapshot_test.zig") }; _ = @import("acp_agents.zig"); _ = @import("subagent_activity.zig"); _ = @import("subagent_recovery.zig"); diff --git a/src/peer_channel.zig b/src/peer_channel.zig index 9751b5f3..b0f55826 100644 --- a/src/peer_channel.zig +++ b/src/peer_channel.zig @@ -80,7 +80,11 @@ pub fn handleMessage(self: *Agent, call: ToolCall) !ExecResult { return .{ .text = peer_inbox.formatList(self.arena, everyone, presence.ownIdentity()), .is_error = false }; } if (std.mem.eql(u8, action, "inbox")) { - return .{ .text = peer_inbox.takeAll(self.arena), .is_error = false }; + const text = peer_inbox.takeAll(self.arena) catch return .{ + .text = "peer_message: inbox read failed; messages and dropped count retained — retry action=inbox", + .is_error = true, + }; + return .{ .text = text, .is_error = false }; } if (!std.mem.eql(u8, action, "send")) return .{ .text = "peer_message action must be send, list, inbox, claim, release, handoff, or status", diff --git a/src/peer_inbox.zig b/src/peer_inbox.zig index 5ac8ae76..d6821c25 100644 --- a/src/peer_inbox.zig +++ b/src/peer_inbox.zig @@ -16,29 +16,42 @@ const worktree_lease = @import("worktree_lease.zig"); const Message = presence_chan.Message; const Owner = worktree_lease.Owner; -/// How many inbound bodies the process will hold. Older ones fall out; the -/// room still has them on disk. +/// How many inbound bodies the process will hold. Evictions are counted; +/// the room still has the original messages on disk. pub const inbox_cap: usize = 8; -const from_cap: usize = 48; -const text_cap: usize = 200; +// Owned independently of the delivery arena, which resets between steps. +var test_storage: Allocator = std.heap.page_allocator; +const storage = if (@import("builtin").is_test) &test_storage else &std.heap.page_allocator; +/// Clear with the previous allocator before changing ownership domains. +pub fn setStorageAllocatorForTest(allocator: Allocator) void { + if (!@import("builtin").is_test) @compileError("test-only inbox allocator override"); + clear(); + test_storage = allocator; +} const Parked = struct { - from: [from_cap]u8 = undefined, - from_len: u8 = 0, - text: [text_cap]u8 = undefined, - text_len: u8 = 0, + from: []const u8, + text: []const u8, dm: bool = false, device: bool = false, + + fn deinit(self: Parked) void { + storage.free(self.from); + storage.free(self.text); + } }; var g_items: [inbox_cap]Parked = undefined; var g_head: usize = 0; var g_len: usize = 0; +var g_dropped: usize = 0; pub fn clear() void { + for (0..g_len) |i| itemAt(i).deinit(); g_head = 0; g_len = 0; + g_dropped = 0; } pub fn resetForTest() void { @@ -49,28 +62,49 @@ pub fn unread() usize { return g_len; } -fn copyInto(dest: []u8, src: []const u8) u8 { - const n = @min(dest.len, src.len); - @memcpy(dest[0..n], src[0..n]); - return @intCast(n); +pub fn dropped() usize { + return g_dropped; +} + +pub fn pending() bool { + return g_len > 0 or g_dropped > 0; +} + +/// Add saved loss after restoring the array, preserving any restore evictions. +pub fn restoreDropped(v: std.json.Value) void { + const count = switch (v) { + .integer => |n| if (n >= 0) (std.math.cast(usize, n) orelse std.math.maxInt(usize)) else return, + // JSON stores unsigned counts above i64 as number strings. + .number_string => |n| std.fmt.parseInt(usize, n, 10) catch return, + else => return, + }; + g_dropped +|= count; } fn parkOne(from: []const u8, text: []const u8, dm: bool, device: bool) void { + const owned_from = storage.dupe(u8, from) catch { + g_dropped +|= 1; + return; + }; + const owned_text = storage.dupe(u8, text) catch { + storage.free(owned_from); + g_dropped +|= 1; + return; + }; + // Allocate first: failure must not evict an otherwise readable message. if (g_len == inbox_cap) { + itemAt(0).deinit(); g_head = (g_head + 1) % inbox_cap; g_len -= 1; + g_dropped +|= 1; } const i = (g_head + g_len) % inbox_cap; - var item: Parked = .{ .dm = dm, .device = device }; - item.from_len = copyInto(&item.from, from); - item.text_len = copyInto(&item.text, peer_context.clip(text, text_cap)); - g_items[i] = item; + g_items[i] = .{ .from = owned_from, .text = owned_text, .dm = dm, .device = device }; g_len += 1; } -/// Copy heard room/device lines into the ring. Returns how many we parked -/// (overflow still counts as parked — the newest stay). Pointers are copied -/// into static buffers because the drain's arena may reset next step. +/// Copy heard room/device lines into the ring. Returns how many were heard +/// (including loss), so even an allocation failure prompts a loss-aware wake. pub fn parkHeard(local: []const Message, device: []const Message) usize { var n: usize = 0; for (local) |m| { @@ -89,11 +123,11 @@ fn itemAt(i: usize) *const Parked { } fn fromSlice(p: *const Parked) []const u8 { - return p.from[0..p.from_len]; + return p.from; } fn textSlice(p: *const Parked) []const u8 { - return p.text[0..p.text_len]; + return p.text; } /// One-line history wake. Cheap on purpose: the bodies wait in the ring. @@ -104,40 +138,47 @@ pub fn formatWake(arena: Allocator) []const u8 { // how to read them, but must not read as a command that displaces the // user's actual request on a trivial turn. Prefix stays "[peer]" — every // compact/peek path detects injects by prefix, not by this wording. - if (g_len == 0) return "[peer] 0 unread — nothing waiting; no inbox read needed"; + if (!pending()) return "[peer] 0 unread — nothing waiting; no inbox read needed"; + if (g_dropped > 0) return std.fmt.allocPrint(arena, "[peer] {d} unread; {d} dropped — parked; peer_message action=inbox when relevant", .{ g_len, g_dropped }) catch "[peer] messages dropped — peer_message action=inbox when relevant"; const first = fromSlice(itemAt(0)); + // Only the wake is a preview. Keep it one line and within the history cap. + const line_end = std.mem.indexOfAny(u8, first, "\r\n") orelse first.len; + var end = @min(line_end, 48); + while (end > 0 and end < first.len and first[end] & 0xc0 == 0x80) end -= 1; + const shown = first[0..end]; var extra: usize = 0; var i: usize = 1; while (i < g_len) : (i += 1) { if (!std.mem.eql(u8, fromSlice(itemAt(i)), first)) extra += 1; } const who = if (extra == 0) - std.fmt.allocPrint(arena, "from {s}", .{first}) catch "from a peer" + std.fmt.allocPrint(arena, "from {s}", .{shown}) catch "from a peer" else - std.fmt.allocPrint(arena, "from {s} + {d} more", .{ first, extra }) catch "from peers"; + std.fmt.allocPrint(arena, "from {s} + {d} more", .{ shown, extra }) catch "from peers"; return std.fmt.allocPrint(arena, "[peer] {d} unread {s} — parked; peer_message action=inbox when relevant", .{ g_len, who }) catch "[peer] unread — peer_message action=inbox when relevant"; } -/// Read+clear. Tool result only — does not re-enter history. -pub fn takeAll(arena: Allocator) []const u8 { - if (g_len == 0) return "inbox empty"; - var buf: std.ArrayList(u8) = .empty; - var i: usize = 0; - while (i < g_len) : (i += 1) { +/// Read+clear only after the complete tool result is allocated successfully. +pub fn takeAll(arena: Allocator) Allocator.Error![]const u8 { + if (!pending()) return "inbox empty"; + var buf: std.Io.Writer.Allocating = .init(arena); + defer buf.deinit(); + if (g_dropped > 0) buf.writer.print("[peer inbox: {d} dropped message(s) could not be retained; the room log keeps the originals]\n", .{g_dropped}) catch return error.OutOfMemory; + for (0..g_len) |i| { const p = itemAt(i); const flag = if (p.device and p.dm) " · device DM" else if (p.device) " · device" else if (p.dm) " · DM" else ""; - const line = std.fmt.allocPrint(arena, "[peer message from {s}{s}]: {s}\n", .{ fromSlice(p), flag, textSlice(p) }) catch continue; - buf.appendSlice(arena, line) catch {}; + buf.writer.print("[peer message from {s}{s}]: {s}\n", .{ fromSlice(p), flag, textSlice(p) }) catch return error.OutOfMemory; } - buf.appendSlice(arena, "(inbox cleared — reply with peer_message; omit session for the room, or name one peer to DM)") catch {}; - g_head = 0; - g_len = 0; - return buf.items; + buf.writer.writeAll("(inbox cleared — reply with peer_message; omit session for the room, or name one peer to DM)") catch return error.OutOfMemory; + const result = try buf.toOwnedSlice(); + clear(); + return result; } /// Digest parked bodies so a save skips only when the mailbox is unchanged. pub fn mixFingerprint(f: anytype) void { f.num(g_len); + f.num(g_dropped); var i: usize = 0; while (i < g_len) : (i += 1) { const p = itemAt(i); @@ -247,11 +288,11 @@ test "inbox ring: park, one-line wake, overflow drops oldest, takeAll clears" { try testing.expect(std.mem.indexOf(u8, wake, "parked") != null); try testing.expect(std.mem.indexOf(u8, wake, "when relevant") != null); try testing.expect(std.mem.indexOfScalar(u8, wake, '\n') == null); - const body = takeAll(a); + const body = try takeAll(a); try testing.expect(std.mem.indexOf(u8, body, "[peer message from session-aaa]: hold gui/src") != null); try testing.expect(std.mem.indexOf(u8, body, "[peer message from session-bbb · device DM]: your turn") != null); try testing.expectEqual(@as(usize, 0), unread()); - try testing.expectEqualStrings("inbox empty", takeAll(a)); + try testing.expectEqualStrings("inbox empty", try takeAll(a)); var i: usize = 0; while (i < inbox_cap + 2) : (i += 1) { @@ -260,7 +301,7 @@ test "inbox ring: park, one-line wake, overflow drops oldest, takeAll clears" { _ = parkHeard(&.{msg(from, text, "")}, &.{}); } try testing.expectEqual(inbox_cap, unread()); - const overflow = takeAll(a); + const overflow = try takeAll(a); try testing.expect(std.mem.indexOf(u8, overflow, "line 0") == null); try testing.expect(std.mem.indexOf(u8, overflow, "line 1") == null); try testing.expect(std.mem.indexOf(u8, overflow, try std.fmt.allocPrint(a, "line {d}", .{inbox_cap + 1})) != null); @@ -313,5 +354,5 @@ test "inbox JSON round-trip: restoreJson rebuilds the ring" { const parsed = try std.json.parseFromSliceLeaky(std.json.Value, a, aw.writer.buffered(), .{}); restoreJson(parsed); try testing.expectEqual(@as(usize, 1), unread()); - try testing.expect(std.mem.indexOf(u8, takeAll(a), "hold the tree") != null); + try testing.expect(std.mem.indexOf(u8, try takeAll(a), "hold the tree") != null); } diff --git a/src/peer_inbox_content_test.zig b/src/peer_inbox_content_test.zig new file mode 100644 index 00000000..a09cfb9c --- /dev/null +++ b/src/peer_inbox_content_test.zig @@ -0,0 +1,166 @@ +const std = @import("std"); +const testing = std.testing; +const inbox = @import("peer_inbox.zig"); +const Message = @import("presence_chan.zig").Message; +const util = @import("util.zig"); + +fn msg(from: []const u8, text: []const u8) Message { + return .{ .from_session = from, .text = text }; +} + +fn expectEmpty() !void { + try testing.expectEqual(@as(usize, 0), inbox.unread()); + try testing.expectEqual(@as(usize, 0), inbox.dropped()); + try testing.expect(!inbox.pending()); +} + +test "inbox content: complete long unicode sender and multiline whitespace body" { + inbox.resetForTest(); + defer inbox.clear(); + var arena = std.heap.ArenaAllocator.init(testing.allocator); + defer arena.deinit(); + const a = arena.allocator(); + const from = "sender-" ++ util.repeatBytes("長い名前🙂", 12) ++ "-end"; + // The former 200-byte boundary falls inside a multibyte character. + const body = util.repeatBytes(" ", 199) ++ "🙂\n\t preserved indentation\r\n" ++ + util.repeatBytes("資料と絵文字🌿\n", 40) ++ " final line\t \n\n"; + try testing.expectEqual(@as(usize, 1), inbox.parkHeard(&.{msg(from, body)}, &.{})); + try testing.expect(inbox.pending()); + const result = try inbox.takeAll(a); + const expected = try std.fmt.allocPrint(a, "[peer message from {s}]: {s}\n", .{ from, body }); + try testing.expect(std.mem.indexOf(u8, result, expected) != null); + try testing.expect(std.unicode.utf8ValidateSlice(result)); + try expectEmpty(); +} + +test "inbox content: parked local and device messages own their source bytes" { + inbox.resetForTest(); + defer inbox.clear(); + var arena = std.heap.ArenaAllocator.init(testing.allocator); + defer arena.deinit(); + const a = arena.allocator(); + const sender = "owned-sender-" ++ util.repeatBytes("x", 80); + const body = "owned body\n\t" ++ util.repeatBytes("whole message 🙂 ", 30) ++ " trailing "; + { + const from = try testing.allocator.dupe(u8, sender); + defer testing.allocator.free(from); + const text = try testing.allocator.dupe(u8, body); + defer testing.allocator.free(text); + var device = msg(from, text); + device.to = "recipient"; + try testing.expectEqual(@as(usize, 2), inbox.parkHeard(&.{msg(from, text)}, &.{device})); + @memset(from, '!'); + @memset(text, '?'); + } + const result = try inbox.takeAll(a); + const local = try std.fmt.allocPrint(a, "[peer message from {s}]: {s}\n", .{ sender, body }); + const device = try std.fmt.allocPrint(a, "[peer message from {s} · device DM]: {s}\n", .{ sender, body }); + try testing.expect(std.mem.indexOf(u8, result, local) != null); + try testing.expect(std.mem.indexOf(u8, result, device) != null); + try expectEmpty(); +} + +test "inbox content: repeated wraparound retains FIFO tail and reports every eviction" { + inbox.resetForTest(); + defer inbox.clear(); + var arena = std.heap.ArenaAllocator.init(testing.allocator); + defer arena.deinit(); + const a = arena.allocator(); + const total = inbox.inbox_cap * 3 + 3; + const lost = total - inbox.inbox_cap; + for (0..total) |i| { + const text = try std.fmt.allocPrint(a, "", .{i}); + try testing.expectEqual(@as(usize, 1), inbox.parkHeard(&.{msg("writer", text)}, &.{})); + try testing.expectEqual(@min(i + 1, inbox.inbox_cap), inbox.unread()); + try testing.expectEqual((i + 1) -| inbox.inbox_cap, inbox.dropped()); + } + try testing.expect(inbox.pending()); + const result = try inbox.takeAll(a); + for (0..lost) |i| { + const text = try std.fmt.allocPrint(a, "", .{i}); + try testing.expect(std.mem.indexOf(u8, result, text) == null); + } + var offset: usize = 0; + for (lost..total) |i| { + const text = try std.fmt.allocPrint(a, "[peer message from writer]: \n", .{i}); + const position = std.mem.indexOfPos(u8, result, offset, text) orelse return error.TestExpectedEqual; + offset = position + text.len; + } + // Match the count next to the loss label, not an unrelated message number. + const count_first = try std.fmt.allocPrint(a, "{d} dropped", .{lost}); + const label_first = try std.fmt.allocPrint(a, "dropped {d}", .{lost}); + try testing.expect(std.mem.indexOf(u8, result, count_first) != null or + std.mem.indexOf(u8, result, label_first) != null); + try expectEmpty(); + try testing.expectEqualStrings("inbox empty", try inbox.takeAll(a)); + try expectEmpty(); +} + +test "inbox content: explicit clear resets overflow and subsequent read starts fresh" { + inbox.resetForTest(); + defer inbox.clear(); + var arena = std.heap.ArenaAllocator.init(testing.allocator); + defer arena.deinit(); + const a = arena.allocator(); + for (0..inbox.inbox_cap + 2) |_| { + _ = inbox.parkHeard(&.{msg("old", "old body")}, &.{}); + } + try testing.expectEqual(@as(usize, 2), inbox.dropped()); + inbox.clear(); + try expectEmpty(); + try testing.expectEqualStrings("inbox empty", try inbox.takeAll(a)); + inbox.clear(); + try expectEmpty(); + _ = inbox.parkHeard(&.{msg("fresh", "fresh body")}, &.{}); + try testing.expectEqual(@as(usize, 0), inbox.dropped()); + const result = try inbox.takeAll(a); + try testing.expect(std.mem.indexOf(u8, result, "[peer message from fresh]: fresh body\n") != null); + try testing.expect(std.mem.indexOf(u8, result, "old body") == null); + try expectEmpty(); + try testing.expectEqualStrings("inbox empty", try inbox.takeAll(a)); +} + +test "inbox content: wake bounds preserve UTF-8 and loss-only reads remain actionable" { + inbox.resetForTest(); + defer inbox.clear(); + var arena = std.heap.ArenaAllocator.init(testing.allocator); + defer arena.deinit(); + const a = arena.allocator(); + const from = "x" ++ util.repeatBytes("🙂", 90) ++ "\nsecond line"; + _ = inbox.parkHeard(&.{msg(from, "body not in wake")}, &.{}); + const wake = inbox.formatWake(a); + try testing.expect(wake.len <= @import("peer_context.zig").inject_byte_cap); + try testing.expect(std.unicode.utf8ValidateSlice(wake)); + try testing.expect(std.mem.startsWith(u8, wake, "[peer]")); + try testing.expect(std.mem.indexOfScalar(u8, wake, '\n') == null); + try testing.expect(std.mem.indexOf(u8, wake, "body not in wake") == null); + try testing.expect(std.mem.endsWith(u8, wake, "action=inbox when relevant")); + inbox.clear(); + inbox.restoreDropped(.{ .integer = 4 }); + try testing.expect(inbox.pending()); + const loss_wake = inbox.formatWake(a); + try testing.expect(std.mem.indexOf(u8, loss_wake, "0 unread; 4 dropped") != null); + const result = try inbox.takeAll(a); + try testing.expect(std.mem.indexOf(u8, result, "4 dropped message(s)") != null); + try expectEmpty(); +} + +test "inbox content: malformed saved loss is ignored and addition saturates" { + inbox.resetForTest(); + defer inbox.clear(); + inbox.restoreDropped(.{ .integer = 3 }); + for ([_]std.json.Value{ .null, .{ .integer = -1 }, .{ .float = 2.5 }, .{ .string = "9" }, .{ .bool = true } }) |v| { + inbox.restoreDropped(v); + try testing.expectEqual(@as(usize, 3), inbox.dropped()); + } + for (0..3) |_| inbox.restoreDropped(.{ .integer = std.math.maxInt(i64) }); + try testing.expectEqual(std.math.maxInt(usize), inbox.dropped()); + var arena = std.heap.ArenaAllocator.init(testing.allocator); + defer arena.deinit(); + const a = arena.allocator(); + const saved = try std.json.Stringify.valueAlloc(a, inbox.dropped(), .{}); + const parsed = try std.json.parseFromSliceLeaky(std.json.Value, a, saved, .{}); + inbox.clear(); + inbox.restoreDropped(parsed); + try testing.expectEqual(std.math.maxInt(usize), inbox.dropped()); +} diff --git a/src/peer_inbox_failure_test.zig b/src/peer_inbox_failure_test.zig new file mode 100644 index 00000000..446f9272 --- /dev/null +++ b/src/peer_inbox_failure_test.zig @@ -0,0 +1,152 @@ +//! Allocation failures must not consume a pull inbox, including tool dispatch. +const std = @import("std"); +const testing = std.testing; +const inbox = @import("peer_inbox.zig"); +const channel = @import("peer_channel.zig"); +const Agent = @import("agent.zig").Agent; +const ToolCall = @import("tools.zig").ToolCall; + +const payload = @import("util.zig").repeatBytes("complete message body — keep this UTF-8 text, not a clipped preview.\n", 40); + +fn seed(a: std.mem.Allocator) !void { + inbox.resetForTest(); + for (0..inbox.inbox_cap + 2) |i| { + const from = try std.fmt.allocPrint(a, "session-with-a-name-longer-than-the-old-forty-eight-byte-limit-{d}", .{i}); + const body = try std.fmt.allocPrint(a, "BEGIN-{d}\n{s}END-{d}", .{ i, payload, i }); + const message = @import("presence_chan.zig").Message{ + .from_session = from, + .text = body, + .to = if (i % 2 == 0) "recipient" else "", + }; + if (i % 2 == 0) { + _ = inbox.parkHeard(&.{message}, &.{}); + } else { + _ = inbox.parkHeard(&.{}, &.{message}); + } + } + try expectPending(); +} + +fn expectPending() !void { + try testing.expect(inbox.pending()); + try testing.expectEqual(inbox.inbox_cap, inbox.unread()); + try testing.expectEqual(@as(usize, 2), inbox.dropped()); +} + +fn expectCleared() !void { + try testing.expect(!inbox.pending()); + try testing.expectEqual(@as(usize, 0), inbox.unread()); + try testing.expectEqual(@as(usize, 0), inbox.dropped()); +} + +fn expectedResult(a: std.mem.Allocator) ![]const u8 { + try seed(a); + const text = try inbox.takeAll(a); + // Validate the oracle too: comparing two equally clipped reads is not a + // full-body regression. Include every retained body, not only its tail. + for (2..inbox.inbox_cap + 2) |i| { + const body = try std.fmt.allocPrint(a, "BEGIN-{d}\n{s}END-{d}", .{ i, payload, i }); + try testing.expect(std.mem.indexOf(u8, text, body) != null); + const from = try std.fmt.allocPrint(a, "session-with-a-name-longer-than-the-old-forty-eight-byte-limit-{d}", .{i}); + try testing.expect(std.mem.indexOf(u8, text, from) != null); + } + try testing.expect(std.mem.indexOf(u8, text, "BEGIN-0\n") == null); + try testing.expect(std.mem.indexOf(u8, text, "BEGIN-1\n") == null); + try expectCleared(); + return text; +} + +fn inboxCall(a: std.mem.Allocator) !ToolCall { + var args: std.json.ObjectMap = .empty; + try args.put(a, "action", .{ .string = "inbox" }); + return .{ .id = "inbox-test", .name = "peer_message", .input = .{ .object = args } }; +} + +test "peer inbox allocation failure: every formatting failure retains bodies and overflow for retry" { + defer inbox.resetForTest(); + var setup = std.heap.ArenaAllocator.init(testing.allocator); + defer setup.deinit(); + const expected = try expectedResult(setup.allocator()); + + // Fail each successive allocation/reallocation, including ones after + // earlier lines have already been formatted. Arenas mirror production + // ownership and reclaim any partial formatting on both paths. + var fail_index: usize = 0; + while (fail_index < 1024) : (fail_index += 1) { + try seed(setup.allocator()); + var storage = std.heap.ArenaAllocator.init(testing.allocator); + defer storage.deinit(); + var failing = testing.FailingAllocator.init(storage.allocator(), .{ .fail_index = fail_index }); + const text = inbox.takeAll(failing.allocator()) catch |err| { + try testing.expectEqual(error.OutOfMemory, err); + try testing.expect(failing.has_induced_failure); + try expectPending(); + const retry = try inbox.takeAll(setup.allocator()); + try testing.expectEqualStrings(expected, retry); + try expectCleared(); + continue; + }; + try testing.expect(!failing.has_induced_failure); + try testing.expect(fail_index > 0); + try testing.expectEqualStrings(expected, text); + try expectCleared(); + return; + } + return error.AllocationSweepDidNotFinish; +} + +test "peer inbox allocation failure: actual action=inbox dispatch reports error without consuming mail" { + defer inbox.resetForTest(); + var setup = std.heap.ArenaAllocator.init(testing.allocator); + defer setup.deinit(); + const expected = try expectedResult(setup.allocator()); + const call = try inboxCall(setup.allocator()); + + var fail_index: usize = 0; + while (fail_index < 1024) : (fail_index += 1) { + try seed(setup.allocator()); + var storage = std.heap.ArenaAllocator.init(testing.allocator); + defer storage.deinit(); + var failing = testing.FailingAllocator.init(storage.allocator(), .{ .fail_index = fail_index }); + // The inbox branch only needs the allocator; other Agent fields must + // not be touched (no presence lookup or filesystem side effects). + var agent: Agent = undefined; + agent.arena = failing.allocator(); + const result = try channel.handleMessage(&agent, call); + if (failing.has_induced_failure) { + try testing.expect(result.is_error); + try testing.expect(result.text.len > 0); + try expectPending(); + agent.arena = setup.allocator(); + const retry = try channel.handleMessage(&agent, call); + try testing.expect(!retry.is_error); + try testing.expectEqualStrings(expected, retry.text); + try expectCleared(); + continue; + } + try testing.expect(fail_index > 0); + try testing.expect(!result.is_error); + try testing.expectEqualStrings(expected, result.text); + try expectCleared(); + return; + } + return error.AllocationSweepDidNotFinish; +} + +test "peer inbox allocation failure: empty helper and inbox dispatch require no allocation" { + inbox.resetForTest(); + defer inbox.resetForTest(); + var setup = std.heap.ArenaAllocator.init(testing.allocator); + defer setup.deinit(); + const call = try inboxCall(setup.allocator()); + var failing = testing.FailingAllocator.init(testing.allocator, .{ .fail_index = 0 }); + try testing.expectEqualStrings("inbox empty", try inbox.takeAll(failing.allocator())); + var agent: Agent = undefined; + agent.arena = failing.allocator(); + const result = try channel.handleMessage(&agent, call); + try testing.expect(!result.is_error); + try testing.expectEqualStrings("inbox empty", result.text); + try testing.expect(!failing.has_induced_failure); + try testing.expectEqual(@as(usize, 0), failing.alloc_index); + try expectCleared(); +} diff --git a/src/peer_inbox_sequence_test.zig b/src/peer_inbox_sequence_test.zig new file mode 100644 index 00000000..be383603 --- /dev/null +++ b/src/peer_inbox_sequence_test.zig @@ -0,0 +1,152 @@ +//! Seeded state-machine regression for ADR 0111. The oracle is a plain FIFO +//! of immutable corpus entries, not a second ring or a read from the inbox. +const std = @import("std"); +const testing = std.testing; +const inbox = @import("peer_inbox.zig"); +const Message = @import("presence_chan.zig").Message; +const util = @import("util.zig"); + +const Entry = struct { from: []const u8, text: []const u8, dm: bool, device: bool }; +const senders = [_][]const u8{ "", " \t\r\n", "zero\x00sender", "sender-" ++ util.repeatBytes("長🙂", 20) }; +const bodies = [_][]const u8{ "", "\x00", " \t\r\n ", "BEGIN\n" ++ util.repeatBytes("complete🙂 payload\n", 50) ++ "END \t" }; +const flags = [_][]const u8{ "", " · DM", " · device", " · device DM" }; + +const Model = struct { + queue: std.ArrayList(Entry) = .empty, + lost: usize = 0, + + fn clear(self: *Model) void { + self.queue.clearRetainingCapacity(); + self.lost = 0; + } + + fn append(self: *Model, entry: Entry) !void { + // Deliberately use a shifting list and the specified capacity, not the + // production head/length arithmetic or its capacity constant. + if (self.queue.items.len == 8) { + _ = self.queue.orderedRemove(0); + self.lost += 1; + } + try self.queue.append(testing.allocator, entry); + } + + fn check(self: *const Model, a: std.mem.Allocator) !void { + try testing.expectEqual(self.queue.items.len, inbox.unread()); + try testing.expectEqual(self.lost, inbox.dropped()); + try testing.expectEqual(self.queue.items.len != 0 or self.lost != 0, inbox.pending()); + const saved = try snapshot(a); + try testing.expectEqual(self.queue.items.len, saved.array.items.len); + for (self.queue.items, saved.array.items) |entry, value| { + const obj = value.object; + try testing.expectEqualStrings(entry.from, obj.get("from").?.string); + try testing.expectEqualStrings(entry.text, obj.get("text").?.string); + try testing.expectEqual(entry.dm, obj.get("dm").?.bool); + try testing.expectEqual(entry.device, obj.get("device").?.bool); + } + } + + fn read(self: *Model, a: std.mem.Allocator) !void { + var expected: std.Io.Writer.Allocating = .init(a); + defer expected.deinit(); + if (self.queue.items.len == 0 and self.lost == 0) { + try expected.writer.writeAll("inbox empty"); + } else { + if (self.lost != 0) try expected.writer.print("[peer inbox: {d} dropped message(s) could not be retained; the room log keeps the originals]\n", .{self.lost}); + for (self.queue.items) |entry| { + const flag: usize = @as(usize, @intFromBool(entry.dm)) + 2 * @as(usize, @intFromBool(entry.device)); + try expected.writer.print("[peer message from {s}{s}]: {s}\n", .{ entry.from, flags[flag], entry.text }); + } + try expected.writer.writeAll("(inbox cleared — reply with peer_message; omit session for the room, or name one peer to DM)"); + } + try testing.expectEqualStrings(expected.writer.buffered(), try inbox.takeAll(a)); + self.clear(); + try self.check(a); + } +}; + +fn snapshot(a: std.mem.Allocator) !std.json.Value { + var out: std.Io.Writer.Allocating = .init(a); + defer out.deinit(); + var json: std.json.Stringify = .{ .writer = &out.writer }; + try inbox.writeJson(&json); + return std.json.parseFromSliceLeaky(std.json.Value, a, out.writer.buffered(), .{ .allocate = .alloc_always }); +} + +fn next(state: *u64) usize { + state.* = state.* *% 6364136223846793005 +% 1442695040888963407; + return @intCast(state.* >> 32); +} + +fn parkBatch(model: *Model, state: *u64, count: usize) !void { + for (0..count) |_| { + // Every batch mixes local and device traffic; independently chosen + // target bits exercise all four flag combinations, including empties. + const local = Entry{ .from = senders[next(state) % 4], .text = bodies[next(state) % 4], .dm = next(state) % 2 != 0, .device = false }; + const device = Entry{ .from = senders[next(state) % 4], .text = bodies[next(state) % 4], .dm = next(state) % 2 != 0, .device = true }; + const lm = Message{ .from_session = local.from, .text = local.text, .to = if (local.dm) "recipient" else "" }; + const dm = Message{ .from_session = device.from, .text = device.text, .to = if (device.dm) "recipient" else "" }; + try testing.expectEqual(@as(usize, 2), inbox.parkHeard(&.{lm}, &.{dm})); + try model.append(local); + try model.append(device); + } +} + +test "peer inbox sequence: seeded FIFO oracle across overflow persistence failed reads and retry" { + inbox.setStorageAllocatorForTest(testing.allocator); + defer inbox.setStorageAllocatorForTest(std.heap.page_allocator); + for ([_]u64{ 1, 865, 893, 0xdeadbeefcafef00d }) |seed| { + inbox.clear(); + var model: Model = .{}; + defer model.queue.deinit(testing.allocator); + var state = seed; + for (0..240) |step| { + // All temporary snapshots/results are reclaimed every transition; + // the oracle and inbox retain only eight bounded corpus entries. + var arena = std.heap.ArenaAllocator.init(testing.allocator); + defer arena.deinit(); + const a = arena.allocator(); + // A scripted prefix guarantees each transition, then seeded mixes + // explore differing queue occupancy and histories of loss. + const op = if (step < 10) step else next(&state) % 10; + switch (op) { + 0, 1, 2, 3 => try parkBatch(&model, &state, 1 + next(&state) % 13), + 4 => { + const saved = try snapshot(a); + const loss_json = try std.json.Stringify.valueAlloc(a, model.lost, .{}); + const loss = try std.json.parseFromSliceLeaky(std.json.Value, a, loss_json, .{}); + inbox.clear(); + inbox.restoreJson(saved); + inbox.restoreDropped(loss); + }, + 5 => { + var failing = testing.FailingAllocator.init(testing.allocator, .{ .fail_index = 0 }); + if (model.queue.items.len != 0 or model.lost != 0) { + try testing.expectError(error.OutOfMemory, inbox.takeAll(failing.allocator())); + try testing.expect(failing.has_induced_failure); + try model.check(a); + } else { + try testing.expectEqualStrings("inbox empty", try inbox.takeAll(failing.allocator())); + try testing.expect(!failing.has_induced_failure); + } + try model.read(a); + }, + 6 => try model.read(a), + 7 => { + inbox.clear(); + model.clear(); + }, + 8 => { + const loss = next(&state) % 17; + inbox.restoreDropped(.{ .integer = @intCast(loss) }); + model.lost += loss; + }, + 9 => try testing.expectEqual(@as(usize, 0), inbox.parkHeard(&.{}, &.{})), + else => unreachable, + } + try model.check(a); + } + var arena = std.heap.ArenaAllocator.init(testing.allocator); + defer arena.deinit(); + try model.read(arena.allocator()); + } +} diff --git a/src/peer_inbox_snapshot_test.zig b/src/peer_inbox_snapshot_test.zig new file mode 100644 index 00000000..52bd3497 --- /dev/null +++ b/src/peer_inbox_snapshot_test.zig @@ -0,0 +1,146 @@ +//! Adversarial mailbox snapshots through the session serialization boundary. +const std = @import("std"); +const testing = std.testing; +const Allocator = std.mem.Allocator; +const Value = std.json.Value; +const Agent = @import("agent.zig").Agent; +const inbox = @import("peer_inbox.zig"); +const session = @import("session_peer.zig"); +const presence = @import("presence.zig"); +const main = @import("main.zig"); +const Fingerprint = @import("session_writer.zig").Fingerprint; +const util = @import("util.zig"); + +const Fixture = struct { + arena: std.heap.ArenaAllocator, + unattended: bool, + + fn init() Fixture { + const old = main.unattended; + main.unattended = false; + inbox.setStorageAllocatorForTest(testing.allocator); + presence.resetRoomCursorForTest(); + return .{ .arena = .init(testing.allocator), .unattended = old }; + } + + fn deinit(self: *Fixture) void { + inbox.setStorageAllocatorForTest(std.heap.page_allocator); + presence.resetRoomCursorForTest(); + main.unattended = self.unattended; + self.arena.deinit(); + } + + fn restore(self: *Fixture, value: Value) void { + // Exercise mailbox dispatch without markCaughtUp's private channel + // latch: cursor dispatch has separate session_peer coverage. + var obj = value.object; + _ = obj.swapRemove("chan_off"); + _ = obj.swapRemove("device_off"); + var root: Agent = undefined; + root.arena = self.arena.allocator(); + root.io = testing.io; + root.messages = std.json.Array.init(root.arena); + session.restore(&root, obj); + } + + fn parse(self: *Fixture, text: []const u8) !Value { + return std.json.parseFromSliceLeaky(Value, self.arena.allocator(), text, .{ .allocate = .alloc_always }); + } + + fn save(self: *Fixture) !Value { + var aw: std.Io.Writer.Allocating = .init(self.arena.allocator()); + defer aw.deinit(); + var s: std.json.Stringify = .{ .writer = &aw.writer }; + try s.beginObject(); + try session.writeFields(&s); + try s.endObject(); + return self.parse(aw.writer.buffered()); + } +}; + +fn fingerprint() u64 { + var f = Fingerprint.init(); + session.mixFingerprint(&f); + return f.final(); +} + +test "peer inbox snapshot: malformed entries do not consume capacity or count as evictions" { + var fixture = Fixture.init(); + defer fixture.deinit(); + var source = try fixture.parse("{\"peer_inbox\":[],\"peer_inbox_dropped\":4}"); + const items = &source.object.getPtr("peer_inbox").?.array; + const malformed = try fixture.parse("[null,7,false,[],{}, {\"from\":9,\"text\":\"bad\"},{\"from\":\"bad\",\"text\":null},{\"from\":\"missing text\"},{\"text\":\"missing sender\"}]"); + for (0..inbox.inbox_cap + 3) |i| { + for (malformed.array.items) |bad| try items.append(bad); + const raw = try std.fmt.allocPrint(fixture.arena.allocator(), "{{\"from\":\"peer\",\"text\":\"{d}\"}}", .{i}); + try items.append(try fixture.parse(raw)); + } + fixture.restore(source); + try testing.expectEqual(inbox.inbox_cap, inbox.unread()); + try testing.expectEqual(@as(usize, 7), inbox.dropped()); + const saved = try fixture.save(); + for (saved.object.get("peer_inbox").?.array.items, 3..) |item, i| { + const expected = try std.fmt.allocPrint(fixture.arena.allocator(), "{d}", .{i}); + try testing.expectEqualStrings(expected, item.object.get("text").?.string); + } +} + +test "peer inbox snapshot: empty and escaped bytes survive and invalid flags default false" { + var fixture = Fixture.init(); + defer fixture.deinit(); + const source = try fixture.parse( + \\{"peer_inbox":[{"from":"","text":""},{"from":"a\u0000\n\t\u00e9\ud83d\ude80","text":"\u0000\u0001\b\f\n\r\t\"\\\u96ea\ud83d\ude80","dm":"true","device":1},{"from":"peer","text":"flags","dm":null,"device":{}},{"from":"peer","text":"true flags","dm":true,"device":true}]} + ); + fixture.restore(source); + for (0..3) |_| { + const saved = try fixture.save(); + const items = saved.object.get("peer_inbox").?.array.items; + try testing.expectEqual(@as(usize, 4), items.len); + for (items, source.object.get("peer_inbox").?.array.items, 0..) |item, original, i| { + try testing.expectEqualStrings(original.object.get("from").?.string, item.object.get("from").?.string); + try testing.expectEqualStrings(original.object.get("text").?.string, item.object.get("text").?.string); + try testing.expectEqual(i == 3, item.object.get("dm").?.bool); + try testing.expectEqual(i == 3, item.object.get("device").?.bool); + } + try testing.expectEqualStrings("a\x00\n\té🚀", items[1].object.get("from").?.string); + try testing.expectEqualStrings("\x00\x01\x08\x0c\n\r\t\"\\雪🚀", items[1].object.get("text").?.string); + fixture.restore(saved); + } +} + +test "peer inbox snapshot: unsigned dropped count is stable through repeated saves and restores" { + if (@bitSizeOf(usize) < 64) return error.SkipZigTest; + var fixture = Fixture.init(); + defer fixture.deinit(); + fixture.restore(try fixture.parse("{\"peer_inbox\":[],\"peer_inbox_dropped\":18446744073709551614}")); + const expected: usize = std.math.maxInt(usize) - 1; + const initial = fingerprint(); + for (0..5) |_| { + try testing.expectEqual(expected, inbox.dropped()); + try testing.expectEqual(@as(usize, 0), inbox.unread()); + const saved = try fixture.save(); + try testing.expectEqualStrings("18446744073709551614", saved.object.get("peer_inbox_dropped").?.number_string); + fixture.restore(saved); + try testing.expectEqual(expected, inbox.dropped()); + try testing.expectEqual(initial, fingerprint()); + } +} + +test "peer inbox snapshot: fingerprints cover suffixes flags and dropped-only changes" { + var fixture = Fixture.init(); + defer fixture.deinit(); + const a = fixture.arena.allocator(); + const body = &util.repeatBytes("b", 220); + const sender = &util.repeatBytes("s", 60); + var hashes: [6]u64 = undefined; + for (&hashes, 0..) |*hash, i| { + const raw = try std.fmt.allocPrint(a, "{{\"peer_inbox\":[{{\"from\":\"{s}{s}\",\"text\":\"{s}{s}\",\"dm\":{s},\"device\":{s}}}],\"peer_inbox_dropped\":{d}}}", .{ sender, if (i == 1) "X" else "A", body, if (i == 2) "X" else "A", if (i == 3) "true" else "false", if (i == 4) "true" else "false", @as(usize, if (i == 5) 1 else 0) }); + fixture.restore(try fixture.parse(raw)); + hash.* = fingerprint(); + const saved = try fixture.save(); + inbox.clear(); + fixture.restore(saved); + try testing.expectEqual(hash.*, fingerprint()); + for (hashes[0..i]) |previous| try testing.expect(previous != hash.*); + } +} diff --git a/src/peer_inbox_storage_test.zig b/src/peer_inbox_storage_test.zig new file mode 100644 index 00000000..d3ca97f2 --- /dev/null +++ b/src/peer_inbox_storage_test.zig @@ -0,0 +1,131 @@ +//! Storage ownership and allocation failures, with leak-tracked backing memory. +const std = @import("std"); +const testing = std.testing; +const inbox = @import("peer_inbox.zig"); +const Message = @import("presence_chan.zig").Message; +const payload = @import("util.zig").repeatBytes("full body 🙂\n", 200); + +fn park(i: usize) !void { + const from = try std.fmt.allocPrint(testing.allocator, "sender-{d}", .{i}); + defer testing.allocator.free(from); + const text = try std.fmt.allocPrint(testing.allocator, "body-{d}:{s}", .{ i, payload }); + defer testing.allocator.free(text); + const m: Message = .{ .from_session = from, .text = text, .to = if (i % 2 == 0) "recipient" else "" }; + const heard = if (i % 2 == 0) inbox.parkHeard(&.{m}, &.{}) else inbox.parkHeard(&.{}, &.{m}); + try testing.expectEqual(@as(usize, 1), heard); +} + +fn snapshot() ![]const u8 { + var writer: std.Io.Writer.Allocating = .init(testing.allocator); + defer writer.deinit(); + var json: std.json.Stringify = .{ .writer = &writer.writer }; + try inbox.writeJson(&json); + return writer.toOwnedSlice(); +} + +fn parkingFailure(full: bool, allocation: usize) !void { + var failing = testing.FailingAllocator.init(testing.allocator, .{}); + inbox.setStorageAllocatorForTest(failing.allocator()); + // Must release all allocations while failing is still alive. + defer inbox.setStorageAllocatorForTest(std.heap.page_allocator); + if (full) for (0..inbox.inbox_cap + 2) |i| try park(i); + const before = try snapshot(); + defer testing.allocator.free(before); + const old_len = inbox.unread(); + const old_loss = inbox.dropped(); + failing.fail_index = failing.alloc_index + allocation; + try park(99); + try testing.expect(failing.has_induced_failure); + try testing.expectEqual(old_len, inbox.unread()); + try testing.expectEqual(old_loss + 1, inbox.dropped()); + try testing.expect(inbox.pending()); + const after = try snapshot(); + defer testing.allocator.free(after); + // Byte-exact JSON includes FIFO order, full payloads, senders and flags. + try testing.expectEqualStrings(before, after); + const result = try inbox.takeAll(testing.allocator); + defer testing.allocator.free(result); + try testing.expect(std.mem.indexOf(u8, result, "body-99:") == null); + try testing.expect(!inbox.pending()); + try testing.expectEqual(failing.allocated_bytes, failing.freed_bytes); +} + +test "inbox storage: first parking allocation fails on empty ring" { + try parkingFailure(false, 0); +} + +test "inbox storage: second parking allocation fails on empty ring without leaking sender" { + try parkingFailure(false, 1); +} + +test "inbox storage: first parking allocation fails on wrapped full ring without eviction" { + try parkingFailure(true, 0); +} + +test "inbox storage: second parking allocation fails on wrapped full ring without eviction" { + try parkingFailure(true, 1); +} + +test "inbox storage: eviction restore clear and allocator switch release owned bytes" { + var tracked = testing.FailingAllocator.init(testing.allocator, .{}); + inbox.setStorageAllocatorForTest(tracked.allocator()); + defer inbox.setStorageAllocatorForTest(std.heap.page_allocator); + for (0..inbox.inbox_cap * 3) |i| try park(i); + const saved = try snapshot(); + defer testing.allocator.free(saved); + { + const parsed = try std.json.parseFromSlice(std.json.Value, testing.allocator, saved, .{ .allocate = .alloc_always }); + defer parsed.deinit(); + inbox.restoreJson(parsed.value); + } + const restored = try snapshot(); + defer testing.allocator.free(restored); + try testing.expectEqualStrings(saved, restored); + try testing.expectEqual(@as(usize, 0), inbox.dropped()); + inbox.clear(); + inbox.clear(); + try testing.expectEqual(tracked.allocated_bytes, tracked.freed_bytes); + try park(100); + inbox.setStorageAllocatorForTest(testing.allocator); + try testing.expectEqual(tracked.allocated_bytes, tracked.freed_bytes); + try testing.expect(!inbox.pending()); + try park(101); + inbox.restoreJson(.null); + try testing.expect(!inbox.pending()); +} + +test "inbox storage: output allocation sweep frees partial reads and retains storage until retry" { + inbox.setStorageAllocatorForTest(testing.allocator); + defer inbox.setStorageAllocatorForTest(std.heap.page_allocator); + for (0..inbox.inbox_cap + 2) |i| try park(i); + const expected = try inbox.takeAll(testing.allocator); + defer testing.allocator.free(expected); + for (0..128) |fail_index| { + for (0..inbox.inbox_cap + 2) |i| try park(i); + const before = try snapshot(); + defer testing.allocator.free(before); + var failing = testing.FailingAllocator.init(testing.allocator, .{ .fail_index = fail_index }); + const result = inbox.takeAll(failing.allocator()) catch |err| { + try testing.expectEqual(error.OutOfMemory, err); + try testing.expect(failing.has_induced_failure); + try testing.expectEqual(failing.allocated_bytes, failing.freed_bytes); + try testing.expectEqual(inbox.inbox_cap, inbox.unread()); + try testing.expectEqual(@as(usize, 2), inbox.dropped()); + const after = try snapshot(); + defer testing.allocator.free(after); + try testing.expectEqualStrings(before, after); + const retry = try inbox.takeAll(testing.allocator); + defer testing.allocator.free(retry); + try testing.expectEqualStrings(expected, retry); + try testing.expect(!inbox.pending()); + continue; + }; + defer failing.allocator().free(result); + try testing.expect(!failing.has_induced_failure); + try testing.expect(fail_index > 0); + try testing.expectEqualStrings(expected, result); + try testing.expect(!inbox.pending()); + return; + } + return error.AllocationSweepDidNotFinish; +} diff --git a/src/session_peer.zig b/src/session_peer.zig index 48abe1b4..22688110 100644 --- a/src/session_peer.zig +++ b/src/session_peer.zig @@ -71,6 +71,8 @@ pub fn writeFields(s: *std.json.Stringify) !void { try s.write(cur.device); try s.objectField("peer_inbox"); try peer_inbox.writeJson(s); + try s.objectField("peer_inbox_dropped"); + try s.write(peer_inbox.dropped()); } fn u64Field(obj: std.json.ObjectMap, name: []const u8) ?u64 { @@ -110,8 +112,11 @@ pub fn restore(root: *Agent, obj: std.json.ObjectMap) void { // Pre-0014 file: no cursor. Seek rather than replay the last 10 lines. presence.seekRoomsToTail(root.io, root.arena); } - if (obj.get("peer_inbox")) |v| peer_inbox.restoreJson(v); - if (peer_inbox.unread() > 0) injectWake(root); + // restoreJson clears both bodies and loss accounting, even for a legacy + // session with no mailbox. Add the saved count AFTER array evictions. + peer_inbox.restoreJson(obj.get("peer_inbox") orelse .null); + if (obj.get("peer_inbox_dropped")) |v| peer_inbox.restoreDropped(v); + if (peer_inbox.pending()) injectWake(root); } const testing = std.testing; @@ -177,7 +182,7 @@ test "writeFields / restoreJson: inbox bodies survive a process restart" { try testing.expectEqual(@as(usize, 0), peer_inbox.unread()); peer_inbox.restoreJson(parsed.object.get("peer_inbox").?); try testing.expectEqual(@as(usize, 2), peer_inbox.unread()); - const body = peer_inbox.takeAll(a); + const body = try peer_inbox.takeAll(a); try testing.expect(std.mem.indexOf(u8, body, "hold gui/src") != null); try testing.expect(std.mem.indexOf(u8, body, "your turn") != null); } @@ -204,3 +209,102 @@ test "legacy session without cursor fields seeks rather than replaying (restore try testing.expectEqual(@as(usize, 1), root.messages.items.len); try testing.expect(isHumanUserTurn(root.messages.items[0])); } + +fn restoreTestRoot(a: Allocator) Agent { + var root: Agent = undefined; + root.arena = a; + root.io = testing.io; + root.messages = std.json.Array.init(a); + return root; +} + +test "session peer restore dispatch: full body roundtrip, legacy reset, loss-only wake and one-shot" { + var arena_state = std.heap.ArenaAllocator.init(testing.allocator); + defer arena_state.deinit(); + const a = arena_state.allocator(); + peer_inbox.clear(); + defer peer_inbox.clear(); + presence.resetRoomCursorForTest(); + defer presence.resetRoomCursorForTest(); + const unattended = main_mod.unattended; + main_mod.unattended = false; + defer main_mod.unattended = unattended; + const text = &@import("util.zig").repeatBytes("long body beyond the old clipping boundary: ", 20); + const sender = &@import("util.zig").repeatBytes("sender-beyond-the-old-fixed-buffer-", 3); + const Message = @import("presence_chan.zig").Message; + for (0..peer_inbox.inbox_cap + 2) |_| { + _ = peer_inbox.parkHeard(&.{Message{ .from_session = sender, .text = text, .to = "me" }}, &.{}); + } + var aw: std.Io.Writer.Allocating = .init(a); + var s: std.json.Stringify = .{ .writer = &aw.writer }; + try s.beginObject(); + try writeFields(&s); + try s.endObject(); + const parsed = try std.json.parseFromSliceLeaky(Value, a, aw.writer.buffered(), .{}); + try testing.expectEqual(@as(i64, 2), parsed.object.get("peer_inbox_dropped").?.integer); + const saved = parsed.object.get("peer_inbox").?.array.items; + try testing.expectEqualStrings(text, saved[0].object.get("text").?.string); + try testing.expectEqualStrings(sender, saved[0].object.get("from").?.string); + peer_inbox.clear(); + var root = restoreTestRoot(a); + restore(&root, parsed.object); + try testing.expectEqual(peer_inbox.inbox_cap, peer_inbox.unread()); + try testing.expectEqual(@as(usize, 2), peer_inbox.dropped()); + try testing.expectEqual(@as(usize, 1), root.messages.items.len); + const body = try peer_inbox.takeAll(a); + try testing.expect(std.mem.indexOf(u8, body, text) != null); + try testing.expect(std.mem.indexOf(u8, body, sender) != null); + + // A legacy array resets previously accumulated loss accounting. + const legacy = try std.json.parseFromSliceLeaky(Value, a, "{\"chan_off\":0,\"peer_inbox\":[{\"from\":\"old\",\"text\":\"legacy body\"}]}", .{}); + peer_inbox.restoreDropped(.{ .integer = 9 }); + restore(&root, legacy.object); + try testing.expectEqual(@as(usize, 1), peer_inbox.unread()); + try testing.expectEqual(@as(usize, 0), peer_inbox.dropped()); + peer_inbox.restoreDropped(.{ .integer = 9 }); + const empty = try std.json.parseFromSliceLeaky(Value, a, "{}", .{}); + restore(&root, empty.object); + try testing.expect(!peer_inbox.pending()); + try testing.expectEqual(@as(usize, 0), root.messages.items.len); + + const loss = try std.json.parseFromSliceLeaky(Value, a, "{\"chan_off\":0,\"peer_inbox_dropped\":3}", .{}); + var before = session_writer.Fingerprint.init(); + mixFingerprint(&before); + restore(&root, loss.object); + try testing.expectEqual(@as(usize, 0), peer_inbox.unread()); + try testing.expectEqual(@as(usize, 3), peer_inbox.dropped()); + try testing.expectEqual(@as(usize, 1), root.messages.items.len); + try testing.expect(!isHumanUserTurn(root.messages.items[0])); + var after = session_writer.Fingerprint.init(); + mixFingerprint(&after); + try testing.expect(before.h.final() != after.h.final()); + restore(&root, loss.object); + try testing.expectEqual(@as(usize, 1), root.messages.items.len); + try testing.expectEqual(@as(usize, 3), peer_inbox.dropped()); + main_mod.unattended = true; + restore(&root, parsed.object); + try testing.expect(!peer_inbox.pending()); + try testing.expectEqual(@as(usize, 0), root.messages.items.len); +} + +test "session peer restore dispatch: saved loss adds to oversized array evictions" { + var arena_state = std.heap.ArenaAllocator.init(testing.allocator); + defer arena_state.deinit(); + const a = arena_state.allocator(); + peer_inbox.clear(); + defer peer_inbox.clear(); + presence.resetRoomCursorForTest(); + defer presence.resetRoomCursorForTest(); + const unattended = main_mod.unattended; + main_mod.unattended = false; + defer main_mod.unattended = unattended; + var parsed = try std.json.parseFromSliceLeaky(Value, a, "{\"chan_off\":0,\"peer_inbox\":[],\"peer_inbox_dropped\":5}", .{}); + const item = try std.json.parseFromSliceLeaky(Value, a, "{\"from\":\"peer\",\"text\":\"body\"}", .{}); + const items = parsed.object.getPtr("peer_inbox").?; + for (0..peer_inbox.inbox_cap + 2) |_| try items.array.append(item); + var root = restoreTestRoot(a); + restore(&root, parsed.object); + try testing.expectEqual(peer_inbox.inbox_cap, peer_inbox.unread()); + try testing.expectEqual(@as(usize, 7), peer_inbox.dropped()); + try testing.expectEqual(@as(usize, 1), root.messages.items.len); +}