diff --git a/docs/changes/unreleased/1104-session-drawing-race.md b/docs/changes/unreleased/1104-session-drawing-race.md new file mode 100644 index 0000000000..c97f923411 --- /dev/null +++ b/docs/changes/unreleased/1104-session-drawing-race.md @@ -0,0 +1,9 @@ +--- +kind: internal +title: The carry-ladder session tests wait for the mark's drawing instead of racing it +pr: 1104 +surface: [engine] +invalidates: + - "`TestASuccessfulHandoffJournalsTheRungThatSuppliedTheBrief` could fail inside a full `make check` with the draft as the brief. The checkpoint fixtures now wait for the drawing beside the turn to land, and the race has a regression test." + - "The race looked like an errand stealing a scripted step. It is not: the drawing is a sidecar, so the round count before it lands drifts with the scheduler and the writer's ask could land past the end of the script." +--- diff --git a/internal/session/carryladder_test.go b/internal/session/carryladder_test.go index da232d40a1..27c79f268e 100644 --- a/internal/session/carryladder_test.go +++ b/internal/session/carryladder_test.go @@ -46,7 +46,7 @@ func TestAFaultedHandoffWriterIsJournaledWithItsWordsAndTheRungIsNamed(t *testin ran := make(ranNodes, 2) stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), asked) + events, err := agent.Submit(watchedContext(agent), asked) if err != nil { t.Fatalf("Submit: %v", err) } @@ -108,7 +108,7 @@ func TestASuccessfulHandoffJournalsTheRungThatSuppliedTheBrief(t *testing.T) { ran := make(ranNodes, 2) stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), asked) + events, err := agent.Submit(watchedContext(agent), asked) if err != nil { t.Fatalf("Submit: %v", err) } @@ -163,7 +163,7 @@ func TestTheLineSaysWhenNothingButTheAskWentWithTheWork(t *testing.T) { agent := checkpointAgent(t, blind) ran := make(ranNodes, 2) stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), asked) + events, err := agent.Submit(watchedContext(agent), asked) if err != nil { t.Fatalf("Submit: %v", err) } @@ -191,7 +191,7 @@ func TestTheLineSaysWhenNothingButTheAskWentWithTheWork(t *testing.T) { checkpointChainSketch, "Finish the four pieces.", written)} second := checkpointAgent(t, carrying) stubbedGraph(second, func(*TaskNode) {}) - events, err = second.Submit(context.Background(), asked) + events, err = second.Submit(watchedContext(second), asked) if err != nil { t.Fatalf("Submit: %v", err) } @@ -515,3 +515,107 @@ func TestTheOneModelPromiseTravelsToATaskNodeAndItsCrewOnlyErrands(t *testing.T) t.Errorf("the node's errand ran on %q, want its own model %q", model, child.model) } } + +// A READING BESIDE THE TURN IS WAITED ON, NOT RACED. +// +// The mark's drawing is a sidecar (sidecar.go): it is STARTED at the boundary +// that crosses the net and the turn goes straight on to its next request, so HOW +// MANY ROUNDS PASS between the net firing and the drawing landing is the +// scheduler's to decide, not the road's — [checkpointSlack] says the same in its +// own words, and sizes the script against it with a price's worth of rounds. +// +// A SCRIPTED TURN'S ROUNDS COST NOTHING, so on a busy machine the turn runs them +// off faster than the drawing's goroutine is scheduled onto a core. The gap then +// outruns the slack the script carries, the writer's ask lands PAST the end of +// the script, the scripted completer answers it with its past-the-end line +// (`(unscripted)`, which is not prose), the writer's rung reads degenerate, and +// the carry ladder falls onto the draft the running model wrote. That is the +// race TestASuccessfulHandoffJournalsTheRungThatSuppliedTheBrief lost in a full +// make check: `the worker's brief is not the written handoff: Finish the four +// pieces...` — the draft, not the mastermind's document. +// +// WHAT FIXES IT IS THE FIXTURE WAITING FOR THE READING, which is the order a +// real turn has: a real model spends seconds on a step and the small readings +// beside it land inside it ([watchReadings], [answerWhenQuiet]). This test holds +// the drawing back by a fixed wall time — standing in for a scheduler that does +// not reach the drawing's goroutine in time — and insists on the two facts that +// follow: the turn does not spend its script racing the drawing, and the +// writer's rung still supplies the brief. +func TestAHandoverWaitsForTheDrawingBesideItInsteadOfRacingIt(t *testing.T) { + const asked = "work through the four things I listed and report back" + const draft = "Finish the four pieces, and the auth test is the one still failing." + const written = "Finish the currency module. The auth test is the one still failing and the yaml " + + "route has been ruled out. It is done when the whole suite is green." + + inner := &scriptedCompleter{steps: handoffSteps(checkpointMarkAt(checkpointMarks)+checkpointSlack, + checkpointChainSketch, draft, written)} + answerTheReadingsOffTheQueue(inner) + // THE DRAWING IS HELD AND THE NAMER IS NOT, which is the race exactly: the + // namer's call is an errand the aside answers at once, and it is the DRAWING + // whose goroutine the scheduler is slow to reach. + completer := &lateDrawing{Completer: inner, held: 200 * time.Millisecond} + + agent := checkpointAgent(t, completer) + ran := make(ranNodes, 2) + stubbedGraph(agent, func(node *TaskNode) { ran <- node }) + + events, err := agent.Submit(watchedContext(agent), asked) + if err != nil { + t.Fatalf("Submit: %v", err) + } + collect(t, events) + + // THE TURN DID NOT SPEND ITS SCRIPT RACING THE DRAWING. A turn that is not + // held makes one more round for every instant the drawing's goroutine waits; + // a turn that waits on the reading makes exactly the one boundary the net's + // landing buys it. [checkpointMarkAt] of the last mark is where the net + // fires, and a mark's worth of rounds is more than the held turn's gap has + // ever been — while a turn that raced the drawing runs well past it. + if rounds := conversationRounds(inner); rounds > checkpointMarkAt(checkpointMarks)+checkpointMarks { + t.Fatalf("the turn made %d rounds while the drawing was in flight; it should "+ + "have waited for the reading, not spent its script racing it", rounds) + } + node := ran.await(t) + if !strings.Contains(node.spec.brief, written) { + t.Fatalf("the worker's brief is not the written handoff:\n%s", node.spec.brief) + } +} + +// conversationRounds is how many of a scripted completer's QUEUED requests were +// the CONVERSATION'S own rounds: the turn's request opens on the session's +// system prompt, while the draft, the writer and the remains read all open on a +// user-role page (checkpoint.go). The errands beside the turn never reach the +// queue at all ([answerTheReadingsOffTheQueue]). +func conversationRounds(completer *scriptedCompleter) int { + rounds := 0 + for index := 0; index < completer.requests(); index++ { + messages := completer.request(index) + if len(messages) > 0 && messages[0].Role == "system" { + rounds++ + } + } + return rounds +} + +// lateDrawing holds every mark's drawing back by a fixed wall time before it +// answers, standing in for a scheduler that does not get the drawing's goroutine +// onto a core before the turn has run several more rounds. It is the shape the +// real race has; the fixture's job is to make the turn WAIT for the reading +// rather than race it. +type lateDrawing struct { + Completer + held time.Duration +} + +func (l *lateDrawing) CompleteWithMessages(ctx context.Context, messages []ai.Message, options ...ai.Option) (*ai.Response, error) { + if askedForSketch(messages) { + timer := time.NewTimer(l.held) + defer timer.Stop() + select { + case <-timer.C: + case <-ctx.Done(): + return nil, ctx.Err() + } + } + return l.Completer.CompleteWithMessages(ctx, messages, options...) +} diff --git a/internal/session/checkpoint_quick_test.go b/internal/session/checkpoint_quick_test.go index aecb59e9e0..d19a56d9ab 100644 --- a/internal/session/checkpoint_quick_test.go +++ b/internal/session/checkpoint_quick_test.go @@ -46,7 +46,7 @@ func TestAWriteFreeTurnWithPartsIsTakenByAQuickTask(t *testing.T) { ran := make(ranNodes, 2) graph := stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), asked) + events, err := agent.Submit(watchedContext(agent), asked) if err != nil { t.Fatalf("Submit: %v", err) } @@ -277,7 +277,7 @@ func TestAQuickNodeFromADrawingTicksItsListRatherThanFaulting(t *testing.T) { ran := make(ranNodes, 2) graph := stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), asked) + events, err := agent.Submit(watchedContext(agent), asked) if err != nil { t.Fatalf("Submit: %v", err) } @@ -360,7 +360,7 @@ func TestAQuickCarryOnAsksForNoNameAndTheFullRoadStillDoes(t *testing.T) { completer.mu.Unlock() ran := make(ranNodes, 2) stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), asked) + events, err := agent.Submit(watchedContext(agent), asked) if err != nil { t.Fatalf("Submit: %v", err) } diff --git a/internal/session/checkpoint_test.go b/internal/session/checkpoint_test.go index 67c63c5bca..8e7bbc979a 100644 --- a/internal/session/checkpoint_test.go +++ b/internal/session/checkpoint_test.go @@ -702,6 +702,24 @@ func checkpointAgent(t *testing.T, completer Completer, mutate ...func(*Config)) extra(config) } }) + // THE DRAWING LANDS BEFORE THE NEXT WORD, and this fixture is the one that + // must say so. The mark's reading is a sidecar: it is started at the boundary + // that crosses the net and the turn goes straight on, so how many rounds pass + // between the net firing and the drawing landing is the SCHEDULER's to decide + // — a scripted turn's rounds cost nothing, so on a busy machine the turn runs + // them off faster than the reading's goroutine is scheduled, and the gap can + // outrun the slack a script carries ([checkpointSlack]). A script is then a + // test asserting the scheduler, and #392 one mechanism along: the drawing + // lands past the end, the writer's ask is answered with the scripted + // completer's past-the-end line, and the carry ladder falls onto the draft. + // + // [watchReadings] puts a watch on the turn's own context and [answerWhenQuiet] + // holds each request that carries the belt until every reading beside it has + // landed, which is the order a real turn has: a real model spends seconds on + // a step and the small readings land inside it. A script long enough to cross + // the mark is then long enough whatever else the machine is doing, and the + // tests here submit through [watchedContext] so the watch reaches the turn. + watchReadings(t, agent) return agent } @@ -940,7 +958,7 @@ func TestTheNoteRungsTellTheTurnAndAskNobodyAnything(t *testing.T) { agent := checkpointAgent(t, completer) graph := stubbedGraph(agent, func(node *TaskNode) {}) - events, err := agent.Submit(context.Background(), "work through the four things I listed and report back") + events, err := agent.Submit(watchedContext(agent), "work through the four things I listed and report back") if err != nil { t.Fatalf("Submit: %v", err) } @@ -1019,7 +1037,7 @@ func TestATenRoundReadingTurnIsToldAndNotMoved(t *testing.T) { agent := checkpointAgent(t, completer, func(config *Config) { config.Divide = true }) graph := stubbedGraph(agent, func(node *TaskNode) {}) - events, err := agent.Submit(context.Background(), "how do the home panels get their data") + events, err := agent.Submit(watchedContext(agent), "how do the home panels get their data") if err != nil { t.Fatalf("Submit: %v", err) } @@ -1063,7 +1081,7 @@ func TestADrawingWithPartsNoLongerMovesATurnAtAMark(t *testing.T) { agent := checkpointWritingAgent(t, completer, func(config *Config) { config.Divide = true }) graph := stubbedGraph(agent, func(node *TaskNode) {}) - events, err := agent.Submit(context.Background(), asked) + events, err := agent.Submit(watchedContext(agent), asked) if err != nil { t.Fatalf("Submit: %v", err) } @@ -1097,7 +1115,7 @@ func TestASketchSayingOneJobLetsTheTurnRunOn(t *testing.T) { agent := checkpointAgent(t, completer) graph := stubbedGraph(agent, func(node *TaskNode) {}) - events, err := agent.Submit(context.Background(), "work through the one thing I listed and report back") + events, err := agent.Submit(watchedContext(agent), "work through the one thing I listed and report back") if err != nil { t.Fatalf("Submit: %v", err) } @@ -1150,7 +1168,7 @@ func TestASidecarThatCannotBeReachedCarriesOnAndTheCeilingStillFires(t *testing. ran := make(ranNodes, 2) graph := stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), "work through the four things I listed and report back") + events, err := agent.Submit(watchedContext(agent), "work through the four things I listed and report back") if err != nil { t.Fatalf("Submit: %v", err) } @@ -1321,7 +1339,7 @@ func TestAtTheCeilingTheTurnEndsAndTheWorkMovesToOneWatchedTask(t *testing.T) { ran := make(ranNodes, 2) graph := stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), asked) + events, err := agent.Submit(watchedContext(agent), asked) if err != nil { t.Fatalf("Submit: %v", err) } @@ -1400,7 +1418,7 @@ func TestTheCeilingCarriesTheLastSketchIntoTheBrief(t *testing.T) { ran := make(ranNodes, 2) stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), "work through the four things I listed and report back") + events, err := agent.Submit(watchedContext(agent), "work through the four things I listed and report back") if err != nil { t.Fatalf("Submit: %v", err) } @@ -1431,7 +1449,7 @@ func TestTheCeilingHandsOverEvenWhenNobodyCanWriteTheBrief(t *testing.T) { ran := make(ranNodes, 2) graph := stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), asked) + events, err := agent.Submit(watchedContext(agent), asked) if err != nil { t.Fatalf("Submit: %v", err) } @@ -1471,7 +1489,7 @@ func TestTheHandoffBriefIsNotStreamedIntoTheRoom(t *testing.T) { ran := make(ranNodes, 2) stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), "work through the four things I listed and report back") + events, err := agent.Submit(watchedContext(agent), "work through the four things I listed and report back") if err != nil { t.Fatalf("Submit: %v", err) } @@ -1517,7 +1535,7 @@ func TestADowryOfMachineMarkupIsRefusedAndNeverBecomesTheName(t *testing.T) { ran := make(ranNodes, 2) graph := stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), asked) + events, err := agent.Submit(watchedContext(agent), asked) if err != nil { t.Fatalf("Submit: %v", err) } @@ -1587,7 +1605,7 @@ func TestAContinuationSayingNothingIsLeftDropsTheCeilingHandover(t *testing.T) { node.graph.complete(node, TaskDone) }) - events, err := agent.Submit(context.Background(), "write the eight files I listed and smoke-check them") + events, err := agent.Submit(watchedContext(agent), "write the eight files I listed and smoke-check them") if err != nil { t.Fatalf("Submit: %v", err) } @@ -1724,7 +1742,7 @@ func TestTheTurnsThatMustNeverBeCheckpointedAreNot(t *testing.T) { node.graph.complete(node, TaskDone) }) - events, err := agent.Submit(context.Background(), "work through the four things I listed and report back") + events, err := agent.Submit(watchedContext(agent), "work through the four things I listed and report back") if err != nil { t.Fatalf("Submit: %v", err) } @@ -1785,7 +1803,7 @@ func TestAnInterruptedTurnIsNotCheckpointed(t *testing.T) { node.graph.complete(node, TaskDone) }) - events, err := agent.Submit(context.Background(), "work through the four things I listed and report back") + events, err := agent.Submit(watchedContext(agent), "work through the four things I listed and report back") if err != nil { t.Fatalf("Submit: %v", err) } @@ -2064,7 +2082,7 @@ func TestEveryMarkReadIsJournaledWithItsDecisionAndItsCost(t *testing.T) { ran := make(ranNodes, 2) stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), "work through the four things I listed and report back") + events, err := agent.Submit(watchedContext(agent), "work through the four things I listed and report back") if err != nil { t.Fatalf("Submit: %v", err) } @@ -2159,7 +2177,7 @@ func TestAMarkNobodyCouldReadIsJournaledAsAFailure(t *testing.T) { ran := make(ranNodes, 2) stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), "work through the four things I listed and report back") + events, err := agent.Submit(watchedContext(agent), "work through the four things I listed and report back") if err != nil { t.Fatalf("Submit: %v", err) } @@ -2296,7 +2314,7 @@ func TestTheRunningModelsSayS0IsBelievedOnceAndThenMet(t *testing.T) { ran := make(ranNodes, 2) graph := stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), asked) + events, err := agent.Submit(watchedContext(agent), asked) if err != nil { t.Fatalf("Submit: %v", err) } @@ -2354,7 +2372,7 @@ func TestTheCeilingIsDroppedWhenTheReaderAgreesNothingRemains(t *testing.T) { node.graph.complete(node, TaskDone) }) - events, err := agent.Submit(context.Background(), "write the eight files I listed and smoke-check them") + events, err := agent.Submit(watchedContext(agent), "write the eight files I listed and smoke-check them") if err != nil { t.Fatalf("Submit: %v", err) } @@ -2392,7 +2410,7 @@ func TestASplitIsNeverDroppedByTheRunningModelsDeclaration(t *testing.T) { ran := make(ranNodes, 2) graph := stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), "work through the four things I listed and report back") + events, err := agent.Submit(watchedContext(agent), "work through the four things I listed and report back") if err != nil { t.Fatalf("Submit: %v", err) } @@ -2586,7 +2604,7 @@ func TestTheHandoffIsDraftedByTheRunnerAndWrittenByTheMastermind(t *testing.T) { ran := make(ranNodes, 2) stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), asked) + events, err := agent.Submit(watchedContext(agent), asked) if err != nil { t.Fatalf("Submit: %v", err) } @@ -2652,7 +2670,7 @@ func TestTheHandoffWriterStillWritesWhenTheDraftFailed(t *testing.T) { ran := make(ranNodes, 2) stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), asked) + events, err := agent.Submit(watchedContext(agent), asked) if err != nil { t.Fatalf("Submit: %v", err) } @@ -2717,7 +2735,7 @@ func TestADegenerateHandoffIsRegeneratedOnceAndThenGivenUpOn(t *testing.T) { ran := make(ranNodes, 2) stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), asked) + events, err := agent.Submit(watchedContext(agent), asked) if err != nil { t.Fatalf("Submit: %v", err) } @@ -2762,7 +2780,7 @@ func TestTheSpecTheCeilingBuildsIsFinishedAgainstThePersonsAsk(t *testing.T) { ran := make(ranNodes, 2) stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), asked) + events, err := agent.Submit(watchedContext(agent), asked) if err != nil { t.Fatalf("Submit: %v", err) } @@ -2967,7 +2985,7 @@ func TestATurnThatStopsShortOfTheAskIsReopened(t *testing.T) { agent := checkpointAgent(t, completer) stubbedGraph(agent, func(node *TaskNode) {}) - events, err := agent.Submit(context.Background(), "port the language server and get the golden tests passing") + events, err := agent.Submit(watchedContext(agent), "port the language server and get the golden tests passing") if err != nil { t.Fatalf("Submit: %v", err) } @@ -3005,7 +3023,7 @@ func TestATurnThatEndsOnAQuestionToThePersonIsNotReopened(t *testing.T) { agent := checkpointAgent(t, completer) stubbedGraph(agent, func(node *TaskNode) {}) - events, err := agent.Submit(context.Background(), "port the language server and get the golden tests passing") + events, err := agent.Submit(watchedContext(agent), "port the language server and get the golden tests passing") if err != nil { t.Fatalf("Submit: %v", err) } @@ -3050,7 +3068,7 @@ func TestAReopenedTurnStillMarksOnTheSameMeter(t *testing.T) { agent := checkpointAgent(t, completer, func(config *Config) { config.SessionFile = path }) stubbedGraph(agent, func(node *TaskNode) {}) - events, err := agent.Submit(context.Background(), "port the language server and get the golden tests passing") + events, err := agent.Submit(watchedContext(agent), "port the language server and get the golden tests passing") if err != nil { t.Fatalf("Submit: %v", err) } @@ -3095,7 +3113,7 @@ func TestATurnShorterThanTheFirstMarkIsNeverReadUnlessItChangedTheTree(t *testin agent := checkpointAgent(t, completer, func(config *Config) { config.SessionFile = path }) stubbedGraph(agent, func(node *TaskNode) {}) - events, err := agent.Submit(context.Background(), "list the folder and tell me what is in it") + events, err := agent.Submit(watchedContext(agent), "list the folder and tell me what is in it") if err != nil { t.Fatalf("Submit: %v", err) } @@ -3144,7 +3162,7 @@ func TestATurnThatWroteAndThenStoppedIsReadWhateverItCost(t *testing.T) { agent := checkpointAgent(t, completer) stubbedGraph(agent, func(node *TaskNode) {}) - events, err := agent.Submit(context.Background(), "port analysis.rs and get the tests passing") + events, err := agent.Submit(watchedContext(agent), "port analysis.rs and get the tests passing") if err != nil { t.Fatalf("Submit: %v", err) } @@ -3184,7 +3202,7 @@ func TestATurnThatCheckedWhatItWroteIsNotExposed(t *testing.T) { agent := checkpointAgent(t, completer) stubbedGraph(agent, func(node *TaskNode) {}) - events, err := agent.Submit(context.Background(), "port analysis.rs and get the tests passing") + events, err := agent.Submit(watchedContext(agent), "port analysis.rs and get the tests passing") if err != nil { t.Fatalf("Submit: %v", err) } @@ -3221,7 +3239,7 @@ func TestATurnWithNoRoundsIsNotExposedByAnEarlierTurnsWrite(t *testing.T) { // after it — which is exactly what the arm reads as exposure. writtenTurn(agent, "port analysis.rs", 3) - events, err := agent.Submit(context.Background(), "thanks") + events, err := agent.Submit(watchedContext(agent), "thanks") if err != nil { t.Fatalf("Submit: %v", err) } @@ -3610,7 +3628,7 @@ func TestACoordinationSketchStartsNothingAndRealPartsStillConvert(t *testing.T) agent := checkpointAgent(t, completer, func(config *Config) { config.Divide = true }) graph := stubbedGraph(agent, func(node *TaskNode) {}) - events, err := agent.Submit(context.Background(), asked) + events, err := agent.Submit(watchedContext(agent), asked) if err != nil { t.Fatalf("Submit: %v", err) } @@ -3643,7 +3661,7 @@ func TestACoordinationSketchStartsNothingAndRealPartsStillConvert(t *testing.T) ran := make(ranNodes, 2) graph := stubbedGraph(agent, func(node *TaskNode) { ran <- node }) - events, err := agent.Submit(context.Background(), "work through the four things I listed and report back") + events, err := agent.Submit(watchedContext(agent), "work through the four things I listed and report back") if err != nil { t.Fatalf("Submit: %v", err) } @@ -3769,7 +3787,7 @@ func TestATurnSpentWatchingItsOwnWorkIsNeverCheckpointed(t *testing.T) { agent := checkpointAgent(t, completer) graph := stubbedGraph(agent, func(node *TaskNode) {}) - events, err := agent.Submit(context.Background(), "keep an eye on the four pieces I have out") + events, err := agent.Submit(watchedContext(agent), "keep an eye on the four pieces I have out") if err != nil { t.Fatalf("Submit: %v", err) }