diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/message_steering/pi_output.ts b/apps/server/src/orchestration-v2/testkit/fixtures/message_steering/pi_output.ts index cb48f13643843fadd74cf059273eec62ff601fc4..1ccd33eb00d102ee317e6241c4c17de33669a69c 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/message_steering/pi_output.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/message_steering/pi_output.ts @@ -36,7 +36,7 @@ export function assertPiMessageSteeringOutput( assert.deepEqual( prompts.map((frame) => [frameField(frame, "message"), frameField(frame, "streamingBehavior")]), [ - [MESSAGE_STEERING_INITIAL_PROMPT, undefined], + [MESSAGE_STEERING_INITIAL_PROMPT, "steer"], [MESSAGE_STEERING_STEER_PROMPT, "steer"], ], ); diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/message_steering/pi_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/message_steering/pi_transcript.ndjson index 32ffdd3987c4b7777722b79bb17a2ae44c327386..907bf6618b41c9d788a011ca64822a72d7be726c 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/message_steering/pi_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/message_steering/pi_transcript.ndjson @@ -12,7 +12,7 @@ {"type":"expect_outbound","label":"set_session_name","frame":{"type":"set_session_name","name":"Replay fixture: message_steering","id":"t3-5"}} {"type":"emit_inbound","label":"session_info_changed","frame":{"type":"session_info_changed","name":"Replay fixture: message_steering"}} {"type":"emit_inbound","label":"response:set_session_name","frame":{"id":"t3-5","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: steering fixture initial response"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: steering fixture initial response","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/multi_turn/pi_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/multi_turn/pi_transcript.ndjson index 7e32afc0a5b2ec6d272c3d54cccc5fd2d529b48e..5841abd415c1d93c45342867dc4614526c65c67c 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/multi_turn/pi_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/multi_turn/pi_transcript.ndjson @@ -12,7 +12,7 @@ {"type":"expect_outbound","label":"set_session_name","frame":{"type":"set_session_name","name":"Replay fixture: multi_turn","id":"t3-5"}} {"type":"emit_inbound","label":"session_info_changed","frame":{"type":"session_info_changed","name":"Replay fixture: multi_turn"}} {"type":"emit_inbound","label":"response:set_session_name","frame":{"id":"t3-5","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: first fixture turn complete"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: first fixture turn complete","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} @@ -41,7 +41,7 @@ {"type":"emit_inbound","label":"response:get_entries","frame":{"id":"t3-7","type":"response","command":"get_entries","success":true,"data":{"entries":[{"type":"model_change","id":"5e37f535","parentId":"f3bce6ef","timestamp":"2026-01-01T00:00:00.437Z","provider":"openrouter","modelId":"deepseek/deepseek-v4-flash"},{"type":"session_info","id":"16d089a9","parentId":"5e37f535","timestamp":"2026-01-01T00:00:00.440Z","name":"Replay fixture: multi_turn"},{"type":"message","id":"ea6f5949","parentId":"16d089a9","timestamp":"2026-01-01T00:00:00.448Z","message":{"role":"system","content":"","sections":{"preamble":"","tools":"","rules":"","docs":"","cwd":""},"timestamp":1790898591750,"toolsAdded":[{"name":"read","description":"Read the contents of a file. Supports text files and images (jpg, png, gif, webp, bmp). Images are sent as attachments. For text files, output is truncated to 2000 lines or 50KB (whichever is hit first). Use offset/limit for large files. When you need the full file, continue with offset until complete.","parameters":{"type":"object","required":["path"],"properties":{"path":{"type":"string","description":"Path to the file to read (relative or absolute)"},"offset":{"type":"number","description":"Line number to start reading from (1-indexed)"},"limit":{"type":"number","description":"Maximum number of lines to read"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"bash","description":"Execute a bash command in the current working directory. Returns stdout and stderr. Output is truncated to last 2000 lines or 50KB (whichever is hit first). If truncated, full output is saved to a temp file. Optionally provide a timeout in seconds.","parameters":{"type":"object","required":["command"],"properties":{"command":{"type":"string","description":"Shell command to execute"},"timeout":{"type":"number","description":"Timeout in seconds (optional, no default timeout)"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"edit","description":"Edit a single file using exact text replacement. Every edits[].oldText must match a unique, non-overlapping region of the original file. If two changes affect the same block or nearby lines, merge them into one edit instead of emitting overlapping edits. Do not include large unchanged regions just to connect distant changes.","parameters":{"type":"object","required":["path","edits"],"properties":{"path":{"type":"string","description":"Path to the file to edit (relative or absolute)"},"edits":{"type":"array","items":{"type":"object","required":["oldText","newText"],"properties":{"oldText":{"type":"string","description":"Exact text for one targeted replacement. It must be unique in the original file and must not overlap with any other edits[].oldText in the same call."},"newText":{"type":"string","description":"Replacement text for this targeted edit."}}},"description":"One or more targeted replacements. Each edit is matched against the original file, not incrementally. Do not include overlapping or nested edits. If two changes touch the same block or nearby lines, merge them into one edit instead."}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"write","description":"Write content to a file. Creates the file if it doesn't exist, overwrites if it does. Automatically creates parent directories.","parameters":{"type":"object","required":["path","content"],"properties":{"path":{"type":"string","description":"Path to the file to write (relative or absolute)"},"content":{"type":"string","description":"Content to write to the file"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}}]}},{"type":"message","id":"ff895990","parentId":"ea6f5949","timestamp":"2026-01-01T00:00:00.448Z","message":{"role":"user","content":[{"type":"text","text":"Respond with exactly: first fixture turn complete"}],"timestamp":1767225600444}},{"type":"message","id":"5c0d3865","parentId":"ff895990","timestamp":"2026-01-01T00:00:01.723Z","message":{"role":"assistant","content":[{"type":"thinking","thinking":"The user wants me to respond with exactly: \"first fixture turn complete\"","thinkingSignature":""},{"type":"text","text":"first fixture turn complete"}],"api":"openai-completions","provider":"openrouter","model":"deepseek/deepseek-v4-flash","usage":{"input":1787,"output":21,"cacheRead":0,"cacheWrite":0,"reasoning":18,"totalTokens":1808,"cost":{"input":0.00007480382,"output":0.0000017581199999999999,"cacheRead":0,"cacheWrite":0,"total":0.00007656194}},"stopReason":"stop","timestamp":1767225600465,"responseId":"gen-1790898592-RtbvUBJEB0AHaUCPMqrg","rawStopReason":"stop","thinkingLevel":"high"}}],"leafId":"5c0d3865"}}} {"type":"expect_outbound","label":"get_session_stats","frame":{"type":"get_session_stats","id":"t3-8"}} {"type":"emit_inbound","label":"response:get_session_stats","frame":{"id":"t3-8","type":"response","command":"get_session_stats","success":true,"data":{"sessionFile":"/pi-sessions/session-1.jsonl","sessionId":"00000000-0000-4000-8000-000000000002","userMessages":1,"assistantMessages":1,"toolCalls":0,"toolResults":0,"totalMessages":3,"tokens":{"input":1787,"output":21,"cacheRead":0,"cacheWrite":0,"total":1808},"cost":0.00007656194,"contextUsage":{"tokens":1808,"contextWindow":1048576,"percent":0.17242431640625}}}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: second fixture turn complete"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: second fixture turn complete","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/pi_compaction/pi_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/pi_compaction/pi_transcript.ndjson index 43aadfd2d881ae37914b346b753589cd02571ab0..39243d970f6d92163dc4997740462b8c9cc3ae91 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/pi_compaction/pi_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/pi_compaction/pi_transcript.ndjson @@ -12,7 +12,7 @@ {"type":"expect_outbound","label":"set_session_name","frame":{"type":"set_session_name","name":"Replay fixture: pi_compaction","id":"t3-5"}} {"type":"emit_inbound","label":"session_info_changed","frame":{"type":"session_info_changed","name":"Replay fixture: pi_compaction"}} {"type":"emit_inbound","label":"response:set_session_name","frame":{"id":"t3-5","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Remember the opaque marker compaction-marker-4Z7Q for later. Respond with exactly: compaction marker stored"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Remember the opaque marker compaction-marker-4Z7Q for later. Respond with exactly: compaction marker stored","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} @@ -56,7 +56,7 @@ {"type":"emit_inbound","label":"response:get_entries","frame":{"id":"t3-11","type":"response","command":"get_entries","success":true,"data":{"entries":[],"leafId":"9923aaa9"}}} {"type":"expect_outbound","label":"get_session_stats","frame":{"type":"get_session_stats","id":"t3-12"}} {"type":"emit_inbound","label":"response:get_session_stats","frame":{"id":"t3-12","type":"response","command":"get_session_stats","success":true,"data":{"sessionFile":"/pi-sessions/session-1.jsonl","sessionId":"00000000-0000-4000-8000-000000000002","userMessages":1,"assistantMessages":1,"toolCalls":0,"toolResults":0,"totalMessages":3,"tokens":{"input":1803,"output":36,"cacheRead":0,"cacheWrite":0,"total":1839},"cost":0.0000784875,"contextUsage":{"tokens":1839,"contextWindow":1048576,"percent":0.17538070678710938}}}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: second compaction fixture turn"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: second compaction fixture turn","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} @@ -82,7 +82,7 @@ {"type":"emit_inbound","label":"response:get_entries","frame":{"id":"t3-14","type":"response","command":"get_entries","success":true,"data":{"entries":[{"type":"message","id":"de423e78","parentId":"9923aaa9","timestamp":"2026-01-01T00:00:02.437Z","message":{"role":"user","content":[{"type":"text","text":"Respond with exactly: second compaction fixture turn"}],"timestamp":1767225602435}},{"type":"message","id":"70c802be","parentId":"de423e78","timestamp":"2026-01-01T00:00:03.192Z","message":{"role":"assistant","content":[{"type":"thinking","thinking":"The user wants me to respond with exactly \"second compaction fixture turn\".","thinkingSignature":""},{"type":"text","text":"second compaction fixture turn"}],"api":"openai-completions","provider":"openrouter","model":"deepseek/deepseek-v4-flash","usage":{"input":58,"output":20,"cacheRead":1792,"cacheWrite":0,"reasoning":19,"totalTokens":1870,"cost":{"input":0.00000242788,"output":0.0000016744,"cacheRead":0.000015002623999999998,"cacheWrite":0,"total":0.000019104903999999998}},"stopReason":"stop","timestamp":1767225602438,"responseId":"gen-1790898593-ijLmDH6S1P8UCX8xelRy","rawStopReason":"stop","thinkingLevel":"high"}}],"leafId":"70c802be"}}} {"type":"expect_outbound","label":"get_session_stats","frame":{"type":"get_session_stats","id":"t3-15"}} {"type":"emit_inbound","label":"response:get_session_stats","frame":{"id":"t3-15","type":"response","command":"get_session_stats","success":true,"data":{"sessionFile":"/pi-sessions/session-1.jsonl","sessionId":"00000000-0000-4000-8000-000000000002","userMessages":2,"assistantMessages":2,"toolCalls":0,"toolResults":0,"totalMessages":5,"tokens":{"input":1861,"output":56,"cacheRead":1792,"cacheWrite":0,"total":3709},"cost":0.00009759240400000001,"contextUsage":{"tokens":1870,"contextWindow":1048576,"percent":0.17833709716796875}}}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: third compaction fixture turn"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: third compaction fixture turn","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} @@ -120,7 +120,7 @@ {"type":"emit_inbound","label":"response:get_entries","frame":{"id":"t3-21","type":"response","command":"get_entries","success":true,"data":{"entries":[{"type":"compaction","id":"f5058450","parentId":"705e1e5c","timestamp":"2026-01-01T00:00:34.035Z","summary":"## Goal\nThe user asked the assistant to remember the opaque marker \"compaction-marker-4Z7Q\" for later use in the conversation.\n\n## Constraints & Preferences\n- The assistant was required to respond with exactly \"compaction marker stored\" (which it did).\n\n## Progress\n### Done\n- [x] Acknowledged and stored the marker \"compaction-marker-4Z7Q\"\n- [x] Responded with exactly \"compaction marker stored\"\n\n### In Progress\n- (none)\n\n### Blocked\n- (none)\n\n## Key Decisions\n- **Remember the marker**: The value \"compaction-marker-4Z7Q\" is to be retained as the opaque marker for later reference.\n\n## Next Steps\n1. When the user later references \"the marker\" or asks to use it, recall that the marker value is \"compaction-marker-4Z7Q\".\n2. Use this marker in whatever context the user requests (e.g., compaction operations).\n\n## Critical Context\n- Opaque marker value: **compaction-marker-4Z7Q**\n\n---\n\n**Turn Context (split turn):**\n\n# Conversation Summary\n\n## Original Request\nThe user asked the assistant to \"Respond with exactly: second compaction fixture turn\" — i.e., to reply using only that exact phrase.\n\n## Progress So Far\n- The user issued a single directive requiring a verbatim response of the phrase \"second compaction fixture turn.\"\n- No additional work, decisions, or content has been provided in this segment.\n\n## Context Needed to Continue\n- The required exact response text is: **second compaction fixture turn**\n- The user's instruction is to output nothing beyond that exact phrase.","firstKeptEntryId":"70c802be","tokensBefore":1901,"details":{"readFiles":[],"modifiedFiles":[]},"usage":{"input":0,"output":1003,"cacheRead":693,"cacheWrite":0,"reasoning":660,"totalTokens":1696,"cost":{"input":0,"output":0.00008397116,"cacheRead":0.000005801795999999999,"cacheWrite":0,"total":0.000089772956}},"fromHook":false,"systemMessage":{"role":"system","content":"","sections":{"preamble":"","tools":"","rules":"","docs":"","cwd":""},"toolsAdded":[{"name":"read","description":"Read the contents of a file. Supports text files and images (jpg, png, gif, webp, bmp). Images are sent as attachments. For text files, output is truncated to 2000 lines or 50KB (whichever is hit first). Use offset/limit for large files. When you need the full file, continue with offset until complete.","parameters":{"type":"object","required":["path"],"properties":{"path":{"type":"string","description":"Path to the file to read (relative or absolute)"},"offset":{"type":"number","description":"Line number to start reading from (1-indexed)"},"limit":{"type":"number","description":"Maximum number of lines to read"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"bash","description":"Execute a bash command in the current working directory. Returns stdout and stderr. Output is truncated to last 2000 lines or 50KB (whichever is hit first). If truncated, full output is saved to a temp file. Optionally provide a timeout in seconds.","parameters":{"type":"object","required":["command"],"properties":{"command":{"type":"string","description":"Shell command to execute"},"timeout":{"type":"number","description":"Timeout in seconds (optional, no default timeout)"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"edit","description":"Edit a single file using exact text replacement. Every edits[].oldText must match a unique, non-overlapping region of the original file. If two changes affect the same block or nearby lines, merge them into one edit instead of emitting overlapping edits. Do not include large unchanged regions just to connect distant changes.","parameters":{"type":"object","required":["path","edits"],"properties":{"path":{"type":"string","description":"Path to the file to edit (relative or absolute)"},"edits":{"type":"array","items":{"type":"object","required":["oldText","newText"],"properties":{"oldText":{"type":"string","description":"Exact text for one targeted replacement. It must be unique in the original file and must not overlap with any other edits[].oldText in the same call."},"newText":{"type":"string","description":"Replacement text for this targeted edit."}}},"description":"One or more targeted replacements. Each edit is matched against the original file, not incrementally. Do not include overlapping or nested edits. If two changes touch the same block or nearby lines, merge them into one edit instead."}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"write","description":"Write content to a file. Creates the file if it doesn't exist, overwrites if it does. Automatically creates parent directories.","parameters":{"type":"object","required":["path","content"],"properties":{"path":{"type":"string","description":"Path to the file to write (relative or absolute)"},"content":{"type":"string","description":"Content to write to the file"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}}],"timestamp":1790898625260}}],"leafId":"f5058450"}}} {"type":"expect_outbound","label":"get_session_stats","frame":{"type":"get_session_stats","id":"t3-22"}} {"type":"emit_inbound","label":"response:get_session_stats","frame":{"id":"t3-22","type":"response","command":"get_session_stats","success":true,"data":{"sessionFile":"/pi-sessions/session-1.jsonl","sessionId":"00000000-0000-4000-8000-000000000002","userMessages":3,"assistantMessages":3,"toolCalls":0,"toolResults":0,"totalMessages":7,"tokens":{"input":1950,"output":1079,"cacheRead":4277,"cacheWrite":0,"total":7306},"cost":0.000207767924,"contextUsage":{"tokens":null,"contextWindow":1048576,"percent":null}}}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Return the opaque marker I asked you to remember. Respond with only the marker."}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Return the opaque marker I asked you to remember. Respond with only the marker.","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/provider_thread_resume/pi_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/provider_thread_resume/pi_transcript.ndjson index cba3bd8a396d17be4efba55582747dfb07b2b1ab..c86c6f198f582824fdfcd7135e5c0a20715f7a7c 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/provider_thread_resume/pi_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/provider_thread_resume/pi_transcript.ndjson @@ -12,7 +12,7 @@ {"type":"expect_outbound","label":"set_session_name","frame":{"type":"set_session_name","name":"Replay fixture: provider_thread_resume","id":"t3-5"}} {"type":"emit_inbound","label":"session_info_changed","frame":{"type":"session_info_changed","name":"Replay fixture: provider_thread_resume"}} {"type":"emit_inbound","label":"response:set_session_name","frame":{"id":"t3-5","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: provider thread resume fixture first turn complete"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: provider thread resume fixture first turn complete","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} @@ -74,7 +74,7 @@ {"type":"expect_outbound","label":"set_session_name@p2","frame":{"type":"set_session_name","name":"Replay fixture: provider_thread_resume","id":"t3-6"}} {"type":"emit_inbound","label":"session_info_changed@p2","frame":{"type":"session_info_changed","name":"Replay fixture: provider_thread_resume"}} {"type":"emit_inbound","label":"response:set_session_name@p2","frame":{"id":"t3-6","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt@p2","frame":{"type":"prompt","message":"Using the conversation history available in this resumed thread, first repeat the exact final answer you gave in the previous turn. Then on a new line write exactly: provider thread resume fixture second turn complete"}} +{"type":"expect_outbound","label":"prompt@p2","frame":{"type":"prompt","message":"Using the conversation history available in this resumed thread, first repeat the exact final answer you gave in the previous turn. Then on a new line write exactly: provider thread resume fixture second turn complete","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt@p2","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start@p2","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start@p2","frame":{"type":"turn_start"}} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/simple/pi_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/simple/pi_transcript.ndjson index ba7f84de2878843678725928703ce99a0aaccacb..61c3eb45d7ca5e6e90c06b50a48574b41fd1a52a 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/simple/pi_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/simple/pi_transcript.ndjson @@ -12,7 +12,7 @@ {"type":"expect_outbound","label":"set_session_name","frame":{"type":"set_session_name","name":"Replay fixture: simple","id":"t3-5"}} {"type":"emit_inbound","label":"session_info_changed","frame":{"type":"session_info_changed","name":"Replay fixture: simple"}} {"type":"emit_inbound","label":"response:set_session_name","frame":{"id":"t3-5","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with the following text: fixture simple ok"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with the following text: fixture simple ok","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback/pi_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback/pi_transcript.ndjson index 2fc3def568e6b75415ea1d654adfde7e51858017..44aeb73c4ceee945244a8bbcd9adcf5110aa8abb 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback/pi_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback/pi_transcript.ndjson @@ -12,7 +12,7 @@ {"type":"expect_outbound","label":"set_session_name","frame":{"type":"set_session_name","name":"Replay fixture: thread_rollback","id":"t3-5"}} {"type":"emit_inbound","label":"session_info_changed","frame":{"type":"session_info_changed","name":"Replay fixture: thread_rollback"}} {"type":"emit_inbound","label":"response:set_session_name","frame":{"id":"t3-5","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: rollback fixture first turn complete"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: rollback fixture first turn complete","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} @@ -52,7 +52,7 @@ {"type":"emit_inbound","label":"response:get_entries","frame":{"id":"t3-7","type":"response","command":"get_entries","success":true,"data":{"entries":[{"type":"model_change","id":"cc15a6b4","parentId":"00c31b78","timestamp":"2026-01-01T00:00:00.344Z","provider":"openrouter","modelId":"deepseek/deepseek-v4-flash"},{"type":"session_info","id":"2c1dcf96","parentId":"cc15a6b4","timestamp":"2026-01-01T00:00:00.349Z","name":"Replay fixture: thread_rollback"},{"type":"message","id":"aca5528d","parentId":"2c1dcf96","timestamp":"2026-01-01T00:00:00.357Z","message":{"role":"system","content":"","sections":{"preamble":"","tools":"","rules":"","docs":"","cwd":""},"timestamp":1790898591831,"toolsAdded":[{"name":"read","description":"Read the contents of a file. Supports text files and images (jpg, png, gif, webp, bmp). Images are sent as attachments. For text files, output is truncated to 2000 lines or 50KB (whichever is hit first). Use offset/limit for large files. When you need the full file, continue with offset until complete.","parameters":{"type":"object","required":["path"],"properties":{"path":{"type":"string","description":"Path to the file to read (relative or absolute)"},"offset":{"type":"number","description":"Line number to start reading from (1-indexed)"},"limit":{"type":"number","description":"Maximum number of lines to read"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"bash","description":"Execute a bash command in the current working directory. Returns stdout and stderr. Output is truncated to last 2000 lines or 50KB (whichever is hit first). If truncated, full output is saved to a temp file. Optionally provide a timeout in seconds.","parameters":{"type":"object","required":["command"],"properties":{"command":{"type":"string","description":"Shell command to execute"},"timeout":{"type":"number","description":"Timeout in seconds (optional, no default timeout)"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"edit","description":"Edit a single file using exact text replacement. Every edits[].oldText must match a unique, non-overlapping region of the original file. If two changes affect the same block or nearby lines, merge them into one edit instead of emitting overlapping edits. Do not include large unchanged regions just to connect distant changes.","parameters":{"type":"object","required":["path","edits"],"properties":{"path":{"type":"string","description":"Path to the file to edit (relative or absolute)"},"edits":{"type":"array","items":{"type":"object","required":["oldText","newText"],"properties":{"oldText":{"type":"string","description":"Exact text for one targeted replacement. It must be unique in the original file and must not overlap with any other edits[].oldText in the same call."},"newText":{"type":"string","description":"Replacement text for this targeted edit."}}},"description":"One or more targeted replacements. Each edit is matched against the original file, not incrementally. Do not include overlapping or nested edits. If two changes touch the same block or nearby lines, merge them into one edit instead."}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"write","description":"Write content to a file. Creates the file if it doesn't exist, overwrites if it does. Automatically creates parent directories.","parameters":{"type":"object","required":["path","content"],"properties":{"path":{"type":"string","description":"Path to the file to write (relative or absolute)"},"content":{"type":"string","description":"Content to write to the file"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}}]}},{"type":"message","id":"70559bb2","parentId":"aca5528d","timestamp":"2026-01-01T00:00:00.357Z","message":{"role":"user","content":[{"type":"text","text":"Respond with exactly: rollback fixture first turn complete"}],"timestamp":1767225600354}},{"type":"message","id":"c906b4e2","parentId":"70559bb2","timestamp":"2026-01-01T00:00:02.034Z","message":{"role":"assistant","content":[{"type":"thinking","thinking":"The user is asking me to respond with exactly: \"rollback fixture first turn complete\"\n\nLet me check if there are any rollback-related fixtures or tests in the workspace first, as this seems like a test instruction.","thinkingSignature":""},{"type":"text","text":"rollback fixture first turn complete"}],"api":"openai-completions","provider":"openrouter","model":"deepseek/deepseek-v4-flash","usage":{"input":1791,"output":52,"cacheRead":0,"cacheWrite":0,"reasoning":54,"totalTokens":1843,"cost":{"input":0.00007497126,"output":0.00000435344,"cacheRead":0,"cacheWrite":0,"total":0.0000793247}},"stopReason":"stop","timestamp":1767225600369,"responseId":"gen-1790898592-a7T4i6GUUneRyLxy0gtp","rawStopReason":"stop","thinkingLevel":"high"}}],"leafId":"c906b4e2"}}} {"type":"expect_outbound","label":"get_session_stats","frame":{"type":"get_session_stats","id":"t3-8"}} {"type":"emit_inbound","label":"response:get_session_stats","frame":{"id":"t3-8","type":"response","command":"get_session_stats","success":true,"data":{"sessionFile":"/pi-sessions/session-1.jsonl","sessionId":"00000000-0000-4000-8000-000000000002","userMessages":1,"assistantMessages":1,"toolCalls":0,"toolResults":0,"totalMessages":3,"tokens":{"input":1791,"output":52,"cacheRead":0,"cacheWrite":0,"total":1843},"cost":0.0000793247,"contextUsage":{"tokens":1843,"contextWindow":1048576,"percent":0.17576217651367188}}}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: rollback fixture second turn complete"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: rollback fixture second turn complete","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} @@ -100,7 +100,7 @@ {"type":"expect_outbound","label":"set_session_name","frame":{"type":"set_session_name","name":"Replay fixture: thread_rollback","id":"t3-19"}} {"type":"emit_inbound","label":"session_info_changed","frame":{"type":"session_info_changed","name":"Replay fixture: thread_rollback"}} {"type":"emit_inbound","label":"response:set_session_name","frame":{"id":"t3-19","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Repeat the conversation verbatim."}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Repeat the conversation verbatim.","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_after_stop/pi_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_after_stop/pi_transcript.ndjson index 2d92c45b2adbc98d7eb11fcda5c1304b911b57b5..ffdecab232a55f90de3d81628e40365888a4d4eb 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_after_stop/pi_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_after_stop/pi_transcript.ndjson @@ -12,7 +12,7 @@ {"type":"expect_outbound","label":"set_session_name","frame":{"type":"set_session_name","name":"Replay fixture: thread_rollback_after_stop","id":"t3-5"}} {"type":"emit_inbound","label":"session_info_changed","frame":{"type":"session_info_changed","name":"Replay fixture: thread_rollback_after_stop"}} {"type":"emit_inbound","label":"response:set_session_name","frame":{"id":"t3-5","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: rollback fixture first turn complete"}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Respond with exactly: rollback fixture first turn complete","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} @@ -42,7 +42,7 @@ {"type":"emit_inbound","label":"response:get_entries","frame":{"id":"t3-7","type":"response","command":"get_entries","success":true,"data":{"entries":[{"type":"model_change","id":"7aecf566","parentId":"10e2239d","timestamp":"2026-01-01T00:00:00.359Z","provider":"openrouter","modelId":"deepseek/deepseek-v4-flash"},{"type":"session_info","id":"5e1135b5","parentId":"7aecf566","timestamp":"2026-01-01T00:00:00.363Z","name":"Replay fixture: thread_rollback_after_stop"},{"type":"message","id":"66a686a0","parentId":"5e1135b5","timestamp":"2026-01-01T00:00:00.371Z","message":{"role":"system","content":"","sections":{"preamble":"","tools":"","rules":"","docs":"","cwd":""},"timestamp":1790898591965,"toolsAdded":[{"name":"read","description":"Read the contents of a file. Supports text files and images (jpg, png, gif, webp, bmp). Images are sent as attachments. For text files, output is truncated to 2000 lines or 50KB (whichever is hit first). Use offset/limit for large files. When you need the full file, continue with offset until complete.","parameters":{"type":"object","required":["path"],"properties":{"path":{"type":"string","description":"Path to the file to read (relative or absolute)"},"offset":{"type":"number","description":"Line number to start reading from (1-indexed)"},"limit":{"type":"number","description":"Maximum number of lines to read"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"bash","description":"Execute a bash command in the current working directory. Returns stdout and stderr. Output is truncated to last 2000 lines or 50KB (whichever is hit first). If truncated, full output is saved to a temp file. Optionally provide a timeout in seconds.","parameters":{"type":"object","required":["command"],"properties":{"command":{"type":"string","description":"Shell command to execute"},"timeout":{"type":"number","description":"Timeout in seconds (optional, no default timeout)"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"edit","description":"Edit a single file using exact text replacement. Every edits[].oldText must match a unique, non-overlapping region of the original file. If two changes affect the same block or nearby lines, merge them into one edit instead of emitting overlapping edits. Do not include large unchanged regions just to connect distant changes.","parameters":{"type":"object","required":["path","edits"],"properties":{"path":{"type":"string","description":"Path to the file to edit (relative or absolute)"},"edits":{"type":"array","items":{"type":"object","required":["oldText","newText"],"properties":{"oldText":{"type":"string","description":"Exact text for one targeted replacement. It must be unique in the original file and must not overlap with any other edits[].oldText in the same call."},"newText":{"type":"string","description":"Replacement text for this targeted edit."}}},"description":"One or more targeted replacements. Each edit is matched against the original file, not incrementally. Do not include overlapping or nested edits. If two changes touch the same block or nearby lines, merge them into one edit instead."}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}},{"name":"write","description":"Write content to a file. Creates the file if it doesn't exist, overwrites if it does. Automatically creates parent directories.","parameters":{"type":"object","required":["path","content"],"properties":{"path":{"type":"string","description":"Path to the file to write (relative or absolute)"},"content":{"type":"string","description":"Content to write to the file"}}},"constrainedSampling":{"type":"json_schema","strict":"prefer"}}]}},{"type":"message","id":"0d296cbc","parentId":"66a686a0","timestamp":"2026-01-01T00:00:00.372Z","message":{"role":"user","content":[{"type":"text","text":"Respond with exactly: rollback fixture first turn complete"}],"timestamp":1767225600369}},{"type":"message","id":"b184b5a8","parentId":"0d296cbc","timestamp":"2026-01-01T00:00:01.622Z","message":{"role":"assistant","content":[{"type":"thinking","thinking":"The user wants me to respond with exactly \"rollback fixture first turn complete\".","thinkingSignature":""},{"type":"text","text":"rollback fixture first turn complete"}],"api":"openai-completions","provider":"openrouter","model":"deepseek/deepseek-v4-flash","usage":{"input":1795,"output":24,"cacheRead":0,"cacheWrite":0,"reasoning":21,"totalTokens":1819,"cost":{"input":0.00007513869999999999,"output":0.00000200928,"cacheRead":0,"cacheWrite":0,"total":0.00007714798}},"stopReason":"stop","timestamp":1767225600386,"responseId":"gen-1790898592-vpyqfefcvhCaD41XnfZc","rawStopReason":"stop","thinkingLevel":"high"}}],"leafId":"b184b5a8"}}} {"type":"expect_outbound","label":"get_session_stats","frame":{"type":"get_session_stats","id":"t3-8"}} {"type":"emit_inbound","label":"response:get_session_stats","frame":{"id":"t3-8","type":"response","command":"get_session_stats","success":true,"data":{"sessionFile":"/pi-sessions/session-1.jsonl","sessionId":"00000000-0000-4000-8000-000000000002","userMessages":1,"assistantMessages":1,"toolCalls":0,"toolResults":0,"totalMessages":3,"tokens":{"input":1795,"output":24,"cacheRead":0,"cacheWrite":0,"total":1819},"cost":0.00007714798,"contextUsage":{"tokens":1819,"contextWindow":1048576,"percent":0.17347335815429688}}}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Run this exact local command: `node -e \"console.log('interrupt fixture tool started'); setTimeout(() => {}, 30000)\"`. Do not answer until it completes, then respond exactly: interrupt fixture should not finish naturally."}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Run this exact local command: `node -e \"console.log('interrupt fixture tool started'); setTimeout(() => {}, 30000)\"`. Do not answer until it completes, then respond exactly: interrupt fixture should not finish naturally.","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} @@ -164,7 +164,7 @@ {"type":"expect_outbound","label":"set_session_name@p2","frame":{"type":"set_session_name","name":"Replay fixture: thread_rollback_after_stop","id":"t3-6"}} {"type":"emit_inbound","label":"session_info_changed@p2","frame":{"type":"session_info_changed","name":"Replay fixture: thread_rollback_after_stop"}} {"type":"emit_inbound","label":"response:set_session_name@p2","frame":{"id":"t3-6","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt@p2","frame":{"type":"prompt","message":"Respond with exactly: rollback fixture second turn complete"}} +{"type":"expect_outbound","label":"prompt@p2","frame":{"type":"prompt","message":"Respond with exactly: rollback fixture second turn complete","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt@p2","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start@p2","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start@p2","frame":{"type":"turn_start"}} @@ -223,7 +223,7 @@ {"type":"expect_outbound","label":"set_session_name@p2","frame":{"type":"set_session_name","name":"Replay fixture: thread_rollback_after_stop","id":"t3-17"}} {"type":"emit_inbound","label":"session_info_changed@p2","frame":{"type":"session_info_changed","name":"Replay fixture: thread_rollback_after_stop"}} {"type":"emit_inbound","label":"response:set_session_name@p2","frame":{"id":"t3-17","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt@p2","frame":{"type":"prompt","message":"Repeat the conversation verbatim."}} +{"type":"expect_outbound","label":"prompt@p2","frame":{"type":"prompt","message":"Repeat the conversation verbatim.","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt@p2","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start@p2","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start@p2","frame":{"type":"turn_start"}} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/turn_interrupt_mid_tool/pi_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/turn_interrupt_mid_tool/pi_transcript.ndjson index 14e710c1ff9d10c1189dedaa5058a2bf7780907d..d4db23f4fd8e37a5b66d489444aac17d6ffa805d 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/turn_interrupt_mid_tool/pi_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/turn_interrupt_mid_tool/pi_transcript.ndjson @@ -12,7 +12,7 @@ {"type":"expect_outbound","label":"set_session_name","frame":{"type":"set_session_name","name":"Replay fixture: turn_interrupt_mid_tool","id":"t3-5"}} {"type":"emit_inbound","label":"session_info_changed","frame":{"type":"session_info_changed","name":"Replay fixture: turn_interrupt_mid_tool"}} {"type":"emit_inbound","label":"response:set_session_name","frame":{"id":"t3-5","type":"response","command":"set_session_name","success":true}} -{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Run this exact local command: `node -e \"console.log('interrupt fixture tool started'); setTimeout(() => {}, 30000)\"`. Do not answer until it completes, then respond exactly: interrupt fixture should not finish naturally."}} +{"type":"expect_outbound","label":"prompt","frame":{"type":"prompt","message":"Run this exact local command: `node -e \"console.log('interrupt fixture tool started'); setTimeout(() => {}, 30000)\"`. Do not answer until it completes, then respond exactly: interrupt fixture should not finish naturally.","streamingBehavior":"steer"}} {"type":"emit_inbound","label":"response:prompt","frame":{"type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}} {"type":"emit_inbound","label":"agent_start","frame":{"type":"agent_start"}} {"type":"emit_inbound","label":"turn_start","frame":{"type":"turn_start"}} diff --git a/packages/provider-pi/src/server/adapter.test.ts b/packages/provider-pi/src/server/adapter.test.ts index cab0043ab185fac51b75e37619f68e74f1d5cd91..98a956cc5a08e0c4ac295c084c035a28a0888283 100644 --- a/packages/provider-pi/src/server/adapter.test.ts +++ b/packages/provider-pi/src/server/adapter.test.ts @@ -90,6 +90,8 @@ interface FakePi { readonly vetoNextSwitch: () => void; /** Fields overriding the recorded idle state in the next `get_state` acks, in order. */ readonly queueState: (data: Record) => void; + /** Change Pi's streaming state before its agent_start event reaches T3. */ + readonly setStreaming: (streaming: boolean) => void; /** Hold the next `get_state` response until the test resolves it. */ readonly deferNextState: () => void; /** Resolve the held `get_state` request. */ @@ -154,6 +156,7 @@ const makeFakePi: Effect.Effect = Effect.gen(function* () { let sessionFile = FAKE_SESSION_FILE; let sessionGeneration = 0; let models: ReadonlyArray = []; + let isStreaming: boolean | null = null; let stdinBuffer = ""; const emit = (record: PiRpcRecord, chunkBytes?: number) => @@ -166,7 +169,8 @@ const makeFakePi: Effect.Effect = Effect.gen(function* () { }); const respondTo = (record: PiRpcRecord): PiRpcRecord | null => { - if (typeof record["id"] !== "string") return null; + if (typeof record["id"] !== "string" && !(record["type"] === "prompt" && isStreaming !== null)) + return null; const base = { type: "response", id: record["id"], @@ -181,7 +185,24 @@ const makeFakePi: Effect.Effect = Effect.gen(function* () { } // Queued data overrides fields of the recorded idle state, so a test // that only cares about the session file still gets a real shape. - return { ...base, data: { ...recordedIdleState(sessionFile), ...stateQueue.shift() } }; + return { + ...base, + data: { + ...recordedIdleState(sessionFile), + isStreaming: isStreaming ?? false, + ...stateQueue.shift(), + }, + }; + case "prompt": + if (isStreaming && record["streamingBehavior"] === undefined) { + return { + ...base, + success: false, + error: + "Agent is already processing. Specify streamingBehavior ('steer' or 'followUp') to queue the message.", + }; + } + return base; case "get_available_models": return { ...base, data: { models } }; case "new_session": { @@ -312,6 +333,9 @@ const makeFakePi: Effect.Effect = Effect.gen(function* () { vetoSwitch = true; }, queueState: (data) => stateQueue.push(data), + setStreaming: (streaming) => { + isStreaming = streaming; + }, queueStats: (data) => statsQueue.push(data), closeStdout: Queue.end(stdout), lastSpawn: () => lastSpawn, @@ -488,6 +512,37 @@ const expectModelFailure = (errorMessage: string) => }).pipe(Effect.scoped, Effect.provide(layerTest)); describe("PiAdapterV2", () => { + it.effect.each([false, true])( + "completes a resumed user turn before native wake events arrive, Pi streaming=%s", + (streaming) => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const { runtime, takeEvent } = yield* openRuntime(fake); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + fake.setStreaming(streaming); + yield* runtime.resumeThread({ providerThread }); + yield* startTurn(runtime, providerThread); + yield* fake.takeRequest("prompt"); + yield* fake.emit({ type: "agent_start" }); + yield* fake.emit({ + type: "message_end", + message: { + role: "assistant", + content: [{ type: "text", text: "Resumed prompt completed" }], + stopReason: "stop", + }, + }); + fake.queueState({ isStreaming: false }); + yield* fake.emit({ type: "agent_settled" }); + const terminal = yield* takeEvent((event) => event.type === "turn.terminal"); + assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "completed"); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + it.effect.each([false, true])( "serializes rollback fork hooks with native wake ownership, wake=%s", (nativeWake) => diff --git a/packages/provider-pi/src/server/adapter.ts b/packages/provider-pi/src/server/adapter.ts index 96d2783ecf3f184751c47b484521a79ed5af4a87..722ac13b3989a06a597798bea0e89d7045b9958d 100644 --- a/packages/provider-pi/src/server/adapter.ts +++ b/packages/provider-pi/src/server/adapter.ts @@ -2697,10 +2697,12 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( kind: "turn_start", }); } else if (payload !== null) { + // Pi may already be streaming before its wake reaches T3. + // Let Pi atomically queue the prompt or start an idle run. yield* connection.send({ type: "prompt", message: payload.message, - ...(wake === null ? {} : { streamingBehavior: "steer" }), + streamingBehavior: "steer", ...(payload.images.length === 0 ? {} : { images: payload.images }), }); pendingPromptResponses.push({