@@ -122,6 +122,19 @@ type roundState struct {
122122 // filters Phase 10's synthetic ID out of this counter explicitly.
123123 spontaneousUserCreatedThreadCount int
124124 spontaneousUserCreatedThreadIDs []string
125+
126+ // Phases 16/17: PRODUCTION queue path (session-scoped prompt queue) against
127+ // a live Zed. Unlike the other phases (which drive the low-level
128+ // SendChatMessage primitive), these exercise EnqueueQueuedPrompt →
129+ // processPendingPromptsForSession → processPromptQueue / processInterruptPrompt
130+ // → sendQueuedPromptToSession — the path all production agent sends now use.
131+ // Queue prompts get generated request_ids (not "req-phaseN-"), so these
132+ // phases are driven by polling the store, not the completion switch.
133+ queueThreadID string // established thread the queue phases reuse
134+ phase16Deferred bool // interrupt=false was HELD while busy (no concurrent interaction)
135+ phase16Delivered bool // the deferred message was delivered as the next turn once idle
136+ phase17Interrupted bool // interrupt=true cancelled the running turn (interaction went interrupted)
137+ phase17Delivered bool // the interrupt message was delivered
125138}
126139
127140// phase15AddSample records a single message_added event tied to phase 15's
@@ -231,6 +244,11 @@ func (d *testDriver) syncEventCallback(sessionID string, syncMsg *types.SyncMess
231244 d .round .phase15ThreadID = acpThreadID
232245 log .Printf ("[%s] Phase 15: Captured thread ID: %s" , d .round .agentName , truncate (acpThreadID , 16 ))
233246 }
247+ // Capture the thread created for phase 16 (production queue path).
248+ if d .phase == 16 && d .round .queueThreadID == "" {
249+ d .round .queueThreadID = acpThreadID
250+ log .Printf ("[%s] Phase 16: Captured thread ID: %s" , d .round .agentName , truncate (acpThreadID , 16 ))
251+ }
234252 }
235253
236254 case "user_created_thread" :
@@ -770,8 +788,16 @@ func (d *testDriver) advanceAfterCompletion(completedPhase int) {
770788 d .mu .Unlock ()
771789 d .runPhase13 ()
772790 case 15 :
773- // Phase 15 completed normally — final phase of the round.
774- d .advanceToNextRound ()
791+ // Phase 15 done — advance to the production queue-path phases.
792+ d .mu .Lock ()
793+ d .phase = 16
794+ d .mu .Unlock ()
795+ d .runPhase16 ()
796+ case 16 :
797+ // The phase-16 initial message (which established the queue thread) has
798+ // completed. Now run the queue defer + interrupt sub-tests, which drive
799+ // the production queue path and poll the store for outcomes.
800+ go d .runQueuePhases ()
775801 }
776802}
777803
@@ -1277,6 +1303,191 @@ func (d *testDriver) runPhase15() {
12771303 )
12781304}
12791305
1306+ // runPhase16 establishes a thread for the production queue-path phases by
1307+ // sending an ordinary message. When it completes (case 16), runQueuePhases()
1308+ // drives the queue itself. Establishing the thread up front keeps the queue
1309+ // tests off the first-message boot-race path so they exercise the steady-state
1310+ // busy-defer / interrupt behaviour cleanly.
1311+ func (d * testDriver ) runPhase16 () {
1312+ agent := d .round .agentName
1313+ log .Printf ("\n ==================================================" )
1314+ log .Printf (" [%s] PHASE 16: Production queue path — establish thread" , agent )
1315+ log .Printf ("==================================================" )
1316+ d .startPhaseTimeout (16 )
1317+ d .sendChatMessage ("What is 6 + 6? Reply with just the number." , d .round .reqID ("phase16" ), agent )
1318+ }
1319+
1320+ // findInteractionByPromptID returns (interactionID, state) for the interaction
1321+ // the queue created from the given prompt-history entry, or ("","") if none yet.
1322+ func (d * testDriver ) findInteractionByPromptID (promptID string ) (string , string ) {
1323+ for _ , in := range d .store .GetAllInteractions () {
1324+ if in .PromptID == promptID {
1325+ return in .ID , string (in .State )
1326+ }
1327+ }
1328+ return "" , ""
1329+ }
1330+
1331+ // waitInteractionStreaming polls until the interaction for promptID is actively
1332+ // streaming (waiting state with some assistant content) — the same precondition
1333+ // Phase 8 uses before interrupting, so the second message lands mid-turn rather
1334+ // than before the turn has started (cancelling a not-yet-started turn leaves the
1335+ // follow-up wedged). Returns true once streaming, false on timeout.
1336+ func (d * testDriver ) waitInteractionStreaming (promptID string , timeout time.Duration ) bool {
1337+ deadline := time .Now ().Add (timeout )
1338+ for time .Now ().Before (deadline ) {
1339+ for _ , in := range d .store .GetAllInteractions () {
1340+ if in .PromptID == promptID && in .ResponseMessage != "" {
1341+ return true
1342+ }
1343+ }
1344+ time .Sleep (300 * time .Millisecond )
1345+ }
1346+ return false
1347+ }
1348+
1349+ // waitInteractionState polls until the interaction for promptID reaches one of
1350+ // wantStates, returning true on success or false on timeout.
1351+ func (d * testDriver ) waitInteractionState (promptID string , wantStates map [string ]bool , timeout time.Duration ) bool {
1352+ deadline := time .Now ().Add (timeout )
1353+ for time .Now ().Before (deadline ) {
1354+ if _ , st := d .findInteractionByPromptID (promptID ); st != "" && wantStates [st ] {
1355+ return true
1356+ }
1357+ time .Sleep (500 * time .Millisecond )
1358+ }
1359+ return false
1360+ }
1361+
1362+ // runQueuePhases drives the production send path against live Zed:
1363+ //
1364+ // Phase 16 (defer): enqueue interrupt=false while the agent is busy and assert
1365+ // it is HELD (no concurrent interaction) then delivered as the next turn once
1366+ // idle — the concurrent-mid-turn incident, now fixed.
1367+ // Phase 17 (interrupt): enqueue interrupt=true while busy and assert it cancels
1368+ // the running turn (interaction → interrupted) and its own message is delivered.
1369+ //
1370+ // Driven by polling the store because queue prompts get generated request_ids
1371+ // that don't flow through the phase-completion switch.
1372+ func (d * testDriver ) runQueuePhases () {
1373+ agent := d .round .agentName
1374+ d .mu .Lock ()
1375+ threadID := d .round .queueThreadID
1376+ d .mu .Unlock ()
1377+
1378+ // Resolve the child session that owns the established thread — that's what we
1379+ // enqueue onto (its ZedThreadID drives sendQueuedPromptToSession).
1380+ var sessID string
1381+ for _ , s := range d .store .GetAllSessions () {
1382+ if s .Metadata .ZedThreadID == threadID {
1383+ sessID = s .ID
1384+ break
1385+ }
1386+ }
1387+ if sessID == "" {
1388+ log .Printf ("[%s] Phase 16: ABORT — no session found for queue thread %s" , agent , truncate (threadID , 12 ))
1389+ os .Exit (1 )
1390+ }
1391+ waiting := map [string ]bool {"waiting" : true }
1392+ complete := map [string ]bool {"complete" : true }
1393+ interrupted := map [string ]bool {"interrupted" : true }
1394+
1395+ // ---- Phase 16: busy-defer + deliver-when-idle (interrupt=false) ----
1396+ log .Printf ("\n ==================================================" )
1397+ log .Printf (" [%s] PHASE 16: Queue busy-defer (interrupt=false)" , agent )
1398+ log .Printf ("==================================================" )
1399+ d .startPhaseTimeout (16 )
1400+
1401+ promptA , err := d .srv .EnqueueQueuedPrompt (sessID , "Count slowly from 1 to 40, writing one number per line with a short fact about each." , false )
1402+ if err != nil {
1403+ log .Printf ("[%s] Phase 16: ABORT — enqueue A failed: %v" , agent , err )
1404+ os .Exit (1 )
1405+ }
1406+ if ! d .waitInteractionState (promptA , waiting , 15 * time .Second ) {
1407+ log .Printf ("[%s] Phase 16: ABORT — turn A never became waiting" , agent )
1408+ os .Exit (1 )
1409+ }
1410+ // Wait until A is actively streaming so the session is genuinely busy when we
1411+ // enqueue B (avoids racing a fast turn that completes before B is queued).
1412+ if ! d .waitInteractionStreaming (promptA , 30 * time .Second ) {
1413+ log .Printf ("[%s] Phase 16: ABORT — turn A never started streaming" , agent )
1414+ os .Exit (1 )
1415+ }
1416+ // Agent is now busy on A. Enqueue B (interrupt=false): must be HELD.
1417+ promptB , err := d .srv .EnqueueQueuedPrompt (sessID , "Now also mention whales." , false )
1418+ if err != nil {
1419+ log .Printf ("[%s] Phase 16: ABORT — enqueue B failed: %v" , agent , err )
1420+ os .Exit (1 )
1421+ }
1422+ if inID , _ := d .findInteractionByPromptID (promptB ); inID != "" {
1423+ log .Printf ("[%s] Phase 16: FAIL — B was dispatched concurrently (interaction %s) instead of deferred" , agent , truncate (inID , 12 ))
1424+ } else {
1425+ d .mu .Lock ()
1426+ d .round .phase16Deferred = true
1427+ d .mu .Unlock ()
1428+ log .Printf ("[%s] Phase 16: ✅ interrupt=false HELD while busy (no concurrent interaction)" , agent )
1429+ }
1430+ // Once A completes, the production completion handler drains the queue and B
1431+ // is delivered as the next turn.
1432+ if d .waitInteractionState (promptB , complete , 75 * time .Second ) {
1433+ d .mu .Lock ()
1434+ d .round .phase16Delivered = true
1435+ d .mu .Unlock ()
1436+ log .Printf ("[%s] Phase 16: ✅ deferred message delivered as next turn once idle" , agent )
1437+ } else {
1438+ log .Printf ("[%s] Phase 16: FAIL — deferred message never delivered" , agent )
1439+ }
1440+
1441+ // ---- Phase 17: interrupt (interrupt=true cancels + delivers) ----
1442+ d .mu .Lock ()
1443+ d .phase = 17
1444+ d .mu .Unlock ()
1445+ log .Printf ("\n ==================================================" )
1446+ log .Printf (" [%s] PHASE 17: Queue interrupt (interrupt=true)" , agent )
1447+ log .Printf ("==================================================" )
1448+ d .startPhaseTimeout (17 )
1449+
1450+ promptX , err := d .srv .EnqueueQueuedPrompt (sessID , "Count slowly from 1 to 60, one number per line, with a short sentence about each." , false )
1451+ if err != nil {
1452+ log .Printf ("[%s] Phase 17: ABORT — enqueue X failed: %v" , agent , err )
1453+ os .Exit (1 )
1454+ }
1455+ if ! d .waitInteractionState (promptX , waiting , 15 * time .Second ) {
1456+ log .Printf ("[%s] Phase 17: ABORT — turn X never became waiting" , agent )
1457+ os .Exit (1 )
1458+ }
1459+ // Interrupt only once X is actively streaming (cancelling a not-yet-started
1460+ // turn leaves the follow-up wedged — same precondition Phase 8 relies on).
1461+ if ! d .waitInteractionStreaming (promptX , 30 * time .Second ) {
1462+ log .Printf ("[%s] Phase 17: ABORT — turn X never started streaming" , agent )
1463+ os .Exit (1 )
1464+ }
1465+ // Interrupt while X is streaming: cancel-then-send via processInterruptPrompt.
1466+ promptY , err := d .srv .EnqueueQueuedPrompt (sessID , "Stop counting — just say 'done'." , true )
1467+ if err != nil {
1468+ log .Printf ("[%s] Phase 17: ABORT — enqueue Y (interrupt) failed: %v" , agent , err )
1469+ os .Exit (1 )
1470+ }
1471+ if d .waitInteractionState (promptX , interrupted , 30 * time .Second ) {
1472+ d .mu .Lock ()
1473+ d .round .phase17Interrupted = true
1474+ d .mu .Unlock ()
1475+ log .Printf ("[%s] Phase 17: ✅ interrupt cancelled the running turn (X → interrupted)" , agent )
1476+ } else {
1477+ log .Printf ("[%s] Phase 17: FAIL — running turn X was not interrupted" , agent )
1478+ }
1479+ if d .waitInteractionState (promptY , complete , 60 * time .Second ) {
1480+ d .mu .Lock ()
1481+ d .round .phase17Delivered = true
1482+ d .mu .Unlock ()
1483+ log .Printf ("[%s] Phase 17: ✅ interrupt message delivered" , agent )
1484+ } else {
1485+ log .Printf ("[%s] Phase 17: FAIL — interrupt message never delivered" , agent )
1486+ }
1487+
1488+ d .advanceToNextRound ()
1489+ }
1490+
12801491// --- Per-round validation ---
12811492
12821493func (d * testDriver ) validateRound () roundResult {
@@ -1812,6 +2023,21 @@ func (d *testDriver) validateRound() roundResult {
18122023 }
18132024 }
18142025
2026+ // Phase 16: production queue path — busy-defer + deliver-when-idle.
2027+ if ! d .round .phase16Deferred {
2028+ errors = append (errors , "Phase 16: interrupt=false was NOT held while busy (concurrent dispatch)" )
2029+ }
2030+ if ! d .round .phase16Delivered {
2031+ errors = append (errors , "Phase 16: deferred queue message was never delivered once idle" )
2032+ }
2033+ // Phase 17: production queue path — interrupt cancels + delivers.
2034+ if ! d .round .phase17Interrupted {
2035+ errors = append (errors , "Phase 17: interrupt=true did NOT cancel the running turn" )
2036+ }
2037+ if ! d .round .phase17Delivered {
2038+ errors = append (errors , "Phase 17: interrupt queue message was never delivered" )
2039+ }
2040+
18152041 // --- SUMMARY ---
18162042 passed := len (errors ) == 0
18172043 if ! passed {
@@ -1836,6 +2062,8 @@ func (d *testDriver) validateRound() roundResult {
18362062 log .Printf ("[%s] Phase 13: Helix-initiated cancel - PASSED" , agent )
18372063 log .Printf ("[%s] Phase 14: Cancel no-op - PASSED" , agent )
18382064 log .Printf ("[%s] Phase 15: Streaming patches arrive incrementally - PASSED" , agent )
2065+ log .Printf ("[%s] Phase 16: Queue busy-defer (production path) - PASSED" , agent )
2066+ log .Printf ("[%s] Phase 17: Queue interrupt (production path) - PASSED" , agent )
18392067 }
18402068
18412069 totalCompletions := 0
0 commit comments