From d870780cd27ede177a4a5859a6ca1999bcabc5cd Mon Sep 17 00:00:00 2001 From: Jarvis Date: Sun, 30 Aug 2026 23:19:47 +1000 Subject: [PATCH] fix: Queued-agent polling repeatedly loads and transfers full growing run output --- cmd/serve_jobs_v2.go | 13 ++- cmd/serve_jobs_v2_test.go | 17 ++++ internal/tools/queue_agent.go | 22 ++++- internal/tools/queue_agent_test.go | 127 +++++++++++++++++++---------- 4 files changed, 130 insertions(+), 49 deletions(-) diff --git a/cmd/serve_jobs_v2.go b/cmd/serve_jobs_v2.go index 135bed946..6c8362b8b 100644 --- a/cmd/serve_jobs_v2.go +++ b/cmd/serve_jobs_v2.go @@ -2055,6 +2055,11 @@ func (m *jobsV2Manager) GetRun(id string) (jobsV2Run, error) { return scanRunV2(row) } +func (m *jobsV2Manager) GetRunSummary(id string) (jobsV2Run, error) { + row := m.db.QueryRow(`SELECT `+jobsV2RunSummaryColumns+` FROM job_runs_v2 WHERE id = ?`, id) + return scanRunSummaryV2(row) +} + func (m *jobsV2Manager) ListRuns(jobID string, limit, offset int) ([]jobsV2Run, int, error) { return m.listRuns(jobID, limit, offset, true) } @@ -2894,7 +2899,13 @@ func (s *serveServer) handleRunV2ByID(w http.ResponseWriter, r *http.Request) { writeOpenAIError(w, http.StatusMethodNotAllowed, "invalid_request_error", "method not allowed") return } - run, err := s.jobsV2.GetRun(runID) + var run jobsV2Run + var err error + if queryBool(r, "summary") { + run, err = s.jobsV2.GetRunSummary(runID) + } else { + run, err = s.jobsV2.GetRun(runID) + } if err != nil { writeOpenAIError(w, http.StatusNotFound, "invalid_request_error", "run not found") return diff --git a/cmd/serve_jobs_v2_test.go b/cmd/serve_jobs_v2_test.go index 1d9662a33..468a40957 100644 --- a/cmd/serve_jobs_v2_test.go +++ b/cmd/serve_jobs_v2_test.go @@ -366,6 +366,23 @@ func TestJobsV2RunsSummaryOmitsOutputPayload(t *testing.T) { t.Fatalf("summary lost metadata: %+v", summary) } + summaryDetailReq := httptest.NewRequest(http.MethodGet, "/v2/runs/run_summary_payload?summary=true", nil) + summaryDetailRR := httptest.NewRecorder() + srv.handleRunV2ByID(summaryDetailRR, summaryDetailReq) + if summaryDetailRR.Code != http.StatusOK { + t.Fatalf("summary detail status = %d, want 200 body=%s", summaryDetailRR.Code, summaryDetailRR.Body.String()) + } + var summaryDetail jobsV2Run + if err := json.Unmarshal(summaryDetailRR.Body.Bytes(), &summaryDetail); err != nil { + t.Fatalf("decode summary detail: %v", err) + } + if summaryDetail.Stdout != "" || summaryDetail.Stderr != "" || summaryDetail.Thinking != "" || summaryDetail.Response != "" { + t.Fatalf("summary detail included output payloads: stdout=%d stderr=%d thinking=%d response=%d", len(summaryDetail.Stdout), len(summaryDetail.Stderr), len(summaryDetail.Thinking), len(summaryDetail.Response)) + } + if summaryDetail.Status != jobsV2RunFailed || summaryDetail.Error != "sample error" || summaryDetail.ExitReason != exitReasonException { + t.Fatalf("summary detail lost status metadata: %+v", summaryDetail) + } + detailReq := httptest.NewRequest(http.MethodGet, "/v2/runs/run_summary_payload", nil) detailRR := httptest.NewRecorder() srv.handleRunV2ByID(detailRR, detailReq) diff --git a/internal/tools/queue_agent.go b/internal/tools/queue_agent.go index 73cc61a3c..bfe1ce582 100644 --- a/internal/tools/queue_agent.go +++ b/internal/tools/queue_agent.go @@ -444,12 +444,12 @@ func (c *jobsBackedAgentClient) waitForRun(ctx context.Context, runID string, po pollInterval = defaultQueuedAgentPollInterval * time.Second } for { - run, err := c.getRun(ctx, runID) + run, err := c.getRunSummary(ctx, runID) if err != nil { return jobsV2AgentRunResponse{}, err } if isQueuedAgentTerminalStatus(run.Status) { - return run, nil + return c.getRun(ctx, runID) } timer := time.NewTimer(pollInterval) select { @@ -472,6 +472,17 @@ func (c *jobsBackedAgentClient) getRun(ctx context.Context, runID string) (jobsV return run, nil } +func (c *jobsBackedAgentClient) getRunSummary(ctx context.Context, runID string) (jobsV2AgentRunResponse, error) { + var run jobsV2AgentRunResponse + if err := c.doJSON(ctx, http.MethodGet, "/v2/runs/"+url.PathEscape(runID)+"?summary=true", nil, &run); err != nil { + return jobsV2AgentRunResponse{}, err + } + if run.ID == "" { + run.ID = runID + } + return run, nil +} + func (c *jobsBackedAgentClient) waitForJob(ctx context.Context, jobID string, pollInterval time.Duration) (jobsV2AgentRunResponse, error) { if pollInterval <= 0 { pollInterval = defaultQueuedAgentPollInterval * time.Second @@ -482,7 +493,10 @@ func (c *jobsBackedAgentClient) waitForJob(ctx context.Context, jobID string, po return jobsV2AgentRunResponse{}, err } if found && isQueuedAgentTerminalStatus(run.Status) { - return run, nil + if run.ID == "" { + return jobsV2AgentRunResponse{}, fmt.Errorf("jobs server returned terminal run without id") + } + return c.getRun(ctx, run.ID) } timer := time.NewTimer(pollInterval) select { @@ -532,7 +546,7 @@ func (c *jobsBackedAgentClient) reconcileRunForJob(ctx context.Context, jobID st func (c *jobsBackedAgentClient) latestRunForJob(ctx context.Context, jobID string) (jobsV2AgentRunResponse, bool, error) { var runs jobsV2AgentRunsListResponse - path := "/v2/runs?limit=1&offset=0&job_id=" + url.QueryEscape(jobID) + path := "/v2/runs?limit=1&offset=0&summary=true&job_id=" + url.QueryEscape(jobID) if err := c.doJSON(ctx, http.MethodGet, path, nil, &runs); err != nil { return jobsV2AgentRunResponse{}, false, err } diff --git a/internal/tools/queue_agent_test.go b/internal/tools/queue_agent_test.go index b700f2871..46e8b230e 100644 --- a/internal/tools/queue_agent_test.go +++ b/internal/tools/queue_agent_test.go @@ -250,39 +250,61 @@ func TestQueueAgentNotifyWhenDonePersistsTrustedOrigin(t *testing.T) { } func TestWaitForJobsPollsUntilTerminal(t *testing.T) { - var polls int32 + var summaryPolls int32 + var fullRequests int32 + payload := strings.Repeat("x", 1<<20) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet || r.URL.Path != "/v2/runs" { - t.Fatalf("unexpected request: %s %s", r.Method, r.URL.Path) - } - if r.URL.Query().Get("job_id") != "job_123" { - t.Fatalf("job_id query = %q, want job_123", r.URL.Query().Get("job_id")) - } - if r.URL.Query().Get("limit") != "1" || r.URL.Query().Get("offset") != "0" { - t.Fatalf("unexpected pagination query: %s", r.URL.RawQuery) - } - count := atomic.AddInt32(&polls, 1) - if count == 1 { - writeJSON(t, w, jobsV2AgentRunsListResponse{Data: []jobsV2AgentRunResponse{{ID: "run_123", JobID: "job_123", Status: "running"}}}) - return + switch { + case r.Method == http.MethodGet && r.URL.Path == "/v2/runs": + if r.URL.Query().Get("job_id") != "job_123" { + t.Fatalf("job_id query = %q, want job_123", r.URL.Query().Get("job_id")) + } + if r.URL.Query().Get("limit") != "1" || r.URL.Query().Get("offset") != "0" { + t.Fatalf("unexpected pagination query: %s", r.URL.RawQuery) + } + if r.URL.Query().Get("summary") != "true" { + t.Fatalf("poll should request a summary, got query %q", r.URL.RawQuery) + } + count := atomic.AddInt32(&summaryPolls, 1) + if count == 1 { + writeJSON(t, w, jobsV2AgentRunsListResponse{Data: []jobsV2AgentRunResponse{{ID: "run_123", JobID: "job_123", Status: "running"}}}) + return + } + exitCode := 0 + turnCount := 1 + inputTokens := 10 + outputTokens := 3 + writeJSON(t, w, jobsV2AgentRunsListResponse{Data: []jobsV2AgentRunResponse{{ + ID: "run_123", + JobID: "job_123", + Status: "succeeded", + ExitReason: "natural_completion", + TurnCount: &turnCount, + InputTokens: &inputTokens, + OutputTokens: &outputTokens, + ExitCode: &exitCode, + StartedAt: "2026-06-07T07:15:51.314202856Z", + FinishedAt: "2026-06-07T07:16:49.259958355Z", + }}}) + case r.Method == http.MethodGet && r.URL.Path == "/v2/runs/run_123": + if r.URL.RawQuery != "" { + t.Fatalf("terminal output fetch should be full, got query %q", r.URL.RawQuery) + } + atomic.AddInt32(&fullRequests, 1) + exitCode := 0 + writeJSON(t, w, jobsV2AgentRunResponse{ + ID: "run_123", + JobID: "job_123", + Status: "succeeded", + Response: "STATUS: COMPLETE\nOK", + Stdout: payload, + ExitCode: &exitCode, + StartedAt: "2026-06-07T07:15:51.314202856Z", + FinishedAt: "2026-06-07T07:16:49.259958355Z", + }) + default: + t.Fatalf("unexpected request: %s %s", r.Method, r.URL.String()) } - exitCode := 0 - turnCount := 1 - inputTokens := 10 - outputTokens := 3 - writeJSON(t, w, jobsV2AgentRunsListResponse{Data: []jobsV2AgentRunResponse{{ - ID: "run_123", - JobID: "job_123", - Status: "succeeded", - ExitReason: "natural_completion", - TurnCount: &turnCount, - InputTokens: &inputTokens, - OutputTokens: &outputTokens, - Response: "STATUS: COMPLETE\nOK", - ExitCode: &exitCode, - StartedAt: "2026-06-07T07:15:51.314202856Z", - FinishedAt: "2026-06-07T07:16:49.259958355Z", - }}}) })) defer server.Close() @@ -301,36 +323,50 @@ func TestWaitForJobsPollsUntilTerminal(t *testing.T) { t.Fatalf("got %d results, want 1", len(results)) } result := results[0] - if result.JobID != "job_123" || result.Status != "succeeded" || result.Response != "STATUS: COMPLETE\nOK" { - t.Fatalf("unexpected result: %+v", result) + if result.JobID != "job_123" || result.Status != "succeeded" || result.Response != "STATUS: COMPLETE\nOK" || result.Stdout != payload { + t.Fatalf("unexpected result metadata or output lengths: job=%q status=%q response=%q stdout=%d", result.JobID, result.Status, result.Response, len(result.Stdout)) } if strings.Contains(out.Content, "run_id") { - t.Fatalf("wait output should not expose run_id: %s", out.Content) + t.Fatalf("wait output should not expose run_id") } if result.DurationSeconds == nil || *result.DurationSeconds < 57.9 || *result.DurationSeconds > 58.0 { t.Fatalf("duration = %#v, want about 57.9", result.DurationSeconds) } - if polls != 2 { - t.Fatalf("polls = %d, want 2", polls) + if summaryPolls != 2 { + t.Fatalf("summary polls = %d, want 2", summaryPolls) + } + if fullRequests != 1 { + t.Fatalf("full requests = %d, want 1", fullRequests) } } func TestWaitForJobsPollsSpecificRunUntilTerminal(t *testing.T) { - var polls int32 + var summaryPolls int32 + var fullRequests int32 + payload := strings.Repeat("x", 1<<20) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch { - case r.Method == http.MethodGet && r.URL.Path == "/v2/runs/run_123": - count := atomic.AddInt32(&polls, 1) + case r.Method == http.MethodGet && r.URL.Path == "/v2/runs/run_123" && r.URL.Query().Get("summary") == "true": + count := atomic.AddInt32(&summaryPolls, 1) if count == 1 { writeJSON(t, w, jobsV2AgentRunResponse{ID: "run_123", JobID: "job_123", Status: "running"}) return } + writeJSON(t, w, jobsV2AgentRunResponse{ + ID: "run_123", + JobID: "job_123", + Status: "succeeded", + StartedAt: "2026-06-07T07:15:51.314202856Z", + }) + case r.Method == http.MethodGet && r.URL.Path == "/v2/runs/run_123" && r.URL.RawQuery == "": + atomic.AddInt32(&fullRequests, 1) exitCode := 0 writeJSON(t, w, jobsV2AgentRunResponse{ ID: "run_123", JobID: "job_123", Status: "succeeded", Response: "STATUS: COMPLETE\nOK", + Stdout: payload, ExitCode: &exitCode, StartedAt: "2026-06-07T07:15:51.314202856Z", FinishedAt: "2026-06-07T07:16:49.259958355Z", @@ -338,7 +374,7 @@ func TestWaitForJobsPollsSpecificRunUntilTerminal(t *testing.T) { case r.Method == http.MethodGet && r.URL.Path == "/v2/runs": t.Fatalf("wait_for_jobs should poll the specific run, got list query %q", r.URL.RawQuery) default: - t.Fatalf("unexpected request: %s %s", r.Method, r.URL.Path) + t.Fatalf("unexpected request: %s %s", r.Method, r.URL.String()) } })) defer server.Close() @@ -358,11 +394,14 @@ func TestWaitForJobsPollsSpecificRunUntilTerminal(t *testing.T) { t.Fatalf("got %d results, want 1", len(results)) } result := results[0] - if result.JobID != "job_123" || result.Status != "succeeded" || result.Response != "STATUS: COMPLETE\nOK" { - t.Fatalf("unexpected result: %+v", result) + if result.JobID != "job_123" || result.Status != "succeeded" || result.Response != "STATUS: COMPLETE\nOK" || result.Stdout != payload { + t.Fatalf("unexpected result metadata or output lengths: job=%q status=%q response=%q stdout=%d", result.JobID, result.Status, result.Response, len(result.Stdout)) + } + if summaryPolls != 2 { + t.Fatalf("summary polls = %d, want 2", summaryPolls) } - if polls != 2 { - t.Fatalf("polls = %d, want 2", polls) + if fullRequests != 1 { + t.Fatalf("full requests = %d, want 1", fullRequests) } }