From 4900a4a85b6b2bad882f2c9c73d93092f46bfe4f Mon Sep 17 00:00:00 2001 From: Multica Eve Date: Fri, 7 Aug 2026 12:40:12 +0800 Subject: [PATCH] MUL-5832: separate directory waits from working signals (#6553) * fix(daemon): separate directory waits from working status Co-authored-by: multica-agent * fix(status): address directory wait review feedback Co-authored-by: multica-agent --------- Co-authored-by: Eve Co-authored-by: multica-agent --- server/cmd/server/runtime_sweeper.go | 3 +- server/cmd/server/runtime_sweeper_test.go | 74 ++++++++++++ server/internal/daemon/daemon.go | 35 +++++- server/internal/daemon/daemon_test.go | 57 +++++++++ server/internal/daemon/health.go | 45 ++++--- server/internal/daemon/health_test.go | 28 ++++- .../internal/daemon/local_directory_test.go | 6 + server/internal/handler/squad.go | 30 ++--- .../handler/squad_member_status_test.go | 110 ++++++++++++++++-- server/internal/service/task.go | 14 ++- server/pkg/db/generated/agent.sql.go | 25 +++- server/pkg/db/generated/squad.sql.go | 6 +- server/pkg/db/queries/agent.sql | 25 +++- server/pkg/db/queries/squad.sql | 6 +- 14 files changed, 393 insertions(+), 71 deletions(-) diff --git a/server/cmd/server/runtime_sweeper.go b/server/cmd/server/runtime_sweeper.go index 72dd75718..3bc09ed85 100644 --- a/server/cmd/server/runtime_sweeper.go +++ b/server/cmd/server/runtime_sweeper.go @@ -398,7 +398,8 @@ func broadcastFailedTasks(ctx context.Context, queries *db.Queries, taskSvc *ser } } -// reconcileAgentStatus refreshes agent status from the current active task set. +// reconcileAgentStatus refreshes agent status from the current working task +// set. A no-op returns no row, so the fallback emits no redundant status event. // Used only by the test-fallback path of broadcastFailedTasks above. func reconcileAgentStatus(ctx context.Context, queries *db.Queries, bus *events.Bus, agentID pgtype.UUID) { agent, err := queries.RefreshAgentStatusFromTasks(ctx, agentID) diff --git a/server/cmd/server/runtime_sweeper_test.go b/server/cmd/server/runtime_sweeper_test.go index 6c82ee33b..55866dbbf 100644 --- a/server/cmd/server/runtime_sweeper_test.go +++ b/server/cmd/server/runtime_sweeper_test.go @@ -8,6 +8,7 @@ import ( "time" "github.com/multica-ai/multica/server/internal/events" + "github.com/multica-ai/multica/server/internal/service" db "github.com/multica-ai/multica/server/pkg/db/generated" ) @@ -113,6 +114,7 @@ func TestRefreshAgentStatusFromTasks(t *testing.T) { t.Cleanup(func() { cleanupSweeperFixture(t, issueID, agentID) }) queries := db.New(testPool) + taskService := service.NewTaskService(queries, testPool, nil, events.New()) if _, err := testPool.Exec(ctx, `UPDATE agent SET status = 'idle' WHERE id = $1`, agentID); err != nil { t.Fatalf("failed to seed idle agent status: %v", err) @@ -126,6 +128,28 @@ func TestRefreshAgentStatusFromTasks(t *testing.T) { t.Fatalf("expected dispatched task to refresh agent status to working, got %q", agent.Status) } + if _, err := taskService.MarkTaskWaitingLocalDirectory(ctx, parseUUID(taskID), "test path busy"); err != nil { + t.Fatalf("MarkTaskWaitingLocalDirectory failed: %v", err) + } + agent, err = queries.GetAgent(ctx, parseUUID(agentID)) + if err != nil { + t.Fatalf("GetAgent after local-directory wait failed: %v", err) + } + if agent.Status != "idle" { + t.Fatalf("expected waiter-only agent status idle, got %q", agent.Status) + } + + if _, err := taskService.StartTask(ctx, parseUUID(taskID)); err != nil { + t.Fatalf("StartTask from local-directory wait failed: %v", err) + } + agent, err = queries.GetAgent(ctx, parseUUID(agentID)) + if err != nil { + t.Fatalf("GetAgent after StartTask failed: %v", err) + } + if agent.Status != "working" { + t.Fatalf("expected running task to restore agent status working, got %q", agent.Status) + } + if _, err := testPool.Exec(ctx, ` UPDATE agent_task_queue SET status = 'cancelled', completed_at = now() @@ -146,6 +170,56 @@ func TestRefreshAgentStatusFromTasks(t *testing.T) { } } +func TestStartTaskSkipsUnchangedAgentStatusWriteAndBroadcast(t *testing.T) { + if testPool == nil { + t.Skip("no database connection") + } + + ctx := context.Background() + issueID, agentID, taskID := setupSweeperTestFixture(t, "dispatched") + t.Cleanup(func() { cleanupSweeperFixture(t, issueID, agentID) }) + + var updatedAtBefore time.Time + if err := testPool.QueryRow(ctx, ` + UPDATE agent + SET status = 'working', updated_at = now() - interval '1 hour' + WHERE id = $1 + RETURNING updated_at + `, agentID).Scan(&updatedAtBefore); err != nil { + t.Fatalf("seed working agent status: %v", err) + } + + bus := events.New() + statusEvents := 0 + bus.Subscribe("agent:status", func(events.Event) { + statusEvents++ + }) + taskService := service.NewTaskService(db.New(testPool), testPool, nil, bus) + + if _, err := taskService.StartTask(ctx, parseUUID(taskID)); err != nil { + t.Fatalf("StartTask from dispatched failed: %v", err) + } + + var ( + status string + updatedAtAfter time.Time + ) + if err := testPool.QueryRow(ctx, ` + SELECT status, updated_at FROM agent WHERE id = $1 + `, agentID).Scan(&status, &updatedAtAfter); err != nil { + t.Fatalf("load agent after StartTask: %v", err) + } + if status != "working" { + t.Fatalf("agent status after StartTask = %q, want working", status) + } + if !updatedAtAfter.Equal(updatedAtBefore) { + t.Fatalf("unchanged status rewrote updated_at: before=%s after=%s", updatedAtBefore, updatedAtAfter) + } + if statusEvents != 0 { + t.Fatalf("unchanged status broadcasts = %d, want 0", statusEvents) + } +} + // TestSweepStaleTasksBroadcastsWithWorkspaceID verifies that when the task sweeper // fails a stale running task, the task:failed event is broadcast with the correct // WorkspaceID so it reaches frontend WebSocket clients (events without WorkspaceID diff --git a/server/internal/daemon/daemon.go b/server/internal/daemon/daemon.go index 3b62bda53..cb0327b4a 100644 --- a/server/internal/daemon/daemon.go +++ b/server/internal/daemon/daemon.go @@ -433,11 +433,21 @@ type Daemon struct { // trySelfReload now calls restartTargetBinary every check tick — without // the cache that is up to two uncached `brew --prefix` forks per tick. brewTargetOnce sync.Once - brewInstall bool // resolved once: was this binary installed via brew? - brewTarget string // "/bin/multica" when brewInstall and the prefix resolved - updating atomic.Bool // prevents concurrent update attempts - activeTasks atomic.Int64 // number of tasks currently in handleTask; exposed via /health - ready atomic.Bool // false until preflight completes; gates /health status (starting -> running) + brewInstall bool // resolved once: was this binary installed via brew? + brewTarget string // "/bin/multica" when brewInstall and the prefix resolved + updating atomic.Bool // prevents concurrent update attempts + // activeTasks is the ownership-safe count of tasks currently in handleTask. + // It deliberately includes preparation and local-directory waiters because + // restart/update barriers must not kill any claimed task. + activeTasks atomic.Int64 + // runningTasks counts live provider execution sessions, beginning only after + // backend.Execute returns. It can briefly lag the server-side running state, + // which starts during preparation before provider launch. resourceWaitTasks + // counts tasks blocked on a local_directory path mutex. Both are diagnostic + // /health dimensions and must never replace activeTasks in safety barriers. + runningTasks atomic.Int64 + resourceWaitTasks atomic.Int64 + ready atomic.Bool // false until preflight completes; gates /health status (starting -> running) // reloadPendingReason explains why a confirmed multica version change hasn't // restarted the daemon yet (a task was running at the barrier check). Set // and cleared by trySelfReload, read by /health. Diagnostic only. @@ -4846,7 +4856,13 @@ func (d *Daemon) acquireLocalDirectoryLockIfNeeded(ctx context.Context, task Tas prepareLeaseOnce sync.Once cancelledByPoll <-chan struct{} stopPrepareLease func() + waitCounted bool ) + defer func() { + if waitCounted { + d.resourceWaitTasks.Add(-1) + } + }() defer func() { if stopPrepareLease != nil { stopPrepareLease() @@ -4854,6 +4870,11 @@ func (d *Daemon) acquireLocalDirectoryLockIfNeeded(ctx context.Context, task Tas }() onWait := func(holder string) { + // LocalPathLocker invokes onWait synchronously and at most once for an + // Acquire call. Count the actual mutex wait even if the best-effort + // server status update below fails. + d.resourceWaitTasks.Add(1) + waitCounted = true reason := fmt.Sprintf("local_directory %s", assignment.AbsPath) if holder != "" { reason = fmt.Sprintf("%s (held by task %s)", reason, shortID(holder)) @@ -6729,6 +6750,10 @@ func (d *Daemon) executeAndDrain(ctx context.Context, backend agent.Backend, pro taskLog.Debug("backend execute returned error", "error", err) return agent.Result{}, 0, err } + // This counter intentionally starts at the narrower provider-session + // boundary, not at the earlier server-side StartTask transition. + d.runningTasks.Add(1) + defer d.runningTasks.Add(-1) taskLog.Debug("backend started, draining messages") // Bound the drain loop only when there is a wall-clock cap. With a positive diff --git a/server/internal/daemon/daemon_test.go b/server/internal/daemon/daemon_test.go index 38193b603..b564c11a7 100644 --- a/server/internal/daemon/daemon_test.go +++ b/server/internal/daemon/daemon_test.go @@ -2802,6 +2802,63 @@ func (blockingBackend) Execute(_ context.Context, _ string, _ agent.ExecOptions) return &agent.Session{Messages: msgCh, Result: resCh}, nil } +type countedRunningBackend struct { + release <-chan struct{} +} + +func (b countedRunningBackend) Execute(_ context.Context, _ string, _ agent.ExecOptions) (*agent.Session, error) { + msgCh := make(chan agent.Message) + resCh := make(chan agent.Result, 1) + go func() { + <-b.release + close(msgCh) + resCh <- agent.Result{Status: "completed"} + }() + return &agent.Session{Messages: msgCh, Result: resCh}, nil +} + +func TestExecuteAndDrainTracksRunningTaskCount(t *testing.T) { + t.Parallel() + + d := newTestDaemon(t) + release := make(chan struct{}) + done := make(chan error, 1) + go func() { + _, _, err := d.executeAndDrain( + context.Background(), + countedRunningBackend{release: release}, + "p", + agent.ExecOptions{}, + slog.Default(), + "task-counted-running", + "", + new(atomic.Int32), + ) + done <- err + }() + + deadline := time.Now().Add(time.Second) + for d.runningTasks.Load() != 1 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if got := d.runningTasks.Load(); got != 1 { + t.Fatalf("running task count while backend is live = %d, want 1", got) + } + + close(release) + select { + case err := <-done: + if err != nil { + t.Fatalf("executeAndDrain: %v", err) + } + case <-time.After(time.Second): + t.Fatal("executeAndDrain did not return after backend release") + } + if got := d.runningTasks.Load(); got != 0 { + t.Fatalf("running task count after backend exit = %d, want 0", got) + } +} + func TestExecuteAndDrain_ContextCancelled_ReportsCancelled(t *testing.T) { t.Parallel() diff --git a/server/internal/daemon/health.go b/server/internal/daemon/health.go index e3db8fca6..ca20bd3e9 100644 --- a/server/internal/daemon/health.go +++ b/server/internal/daemon/health.go @@ -25,14 +25,19 @@ type HealthResponse struct { // lifecycle CLI (`daemon start/stop`) acts on the host process namespace, // so a foreign-OS daemon can't be started/stopped by the app even though // /health is reachable. See #3916. - OS string `json:"os"` - Uptime string `json:"uptime"` - DaemonID string `json:"daemon_id"` - DeviceName string `json:"device_name"` - ServerURL string `json:"server_url"` - CLIVersion string `json:"cli_version"` - ActiveTaskCount int64 `json:"active_task_count"` - Agents []string `json:"agents"` + OS string `json:"os"` + Uptime string `json:"uptime"` + DaemonID string `json:"daemon_id"` + DeviceName string `json:"device_name"` + ServerURL string `json:"server_url"` + CLIVersion string `json:"cli_version"` + // ActiveTaskCount remains the compatibility/safety count of every claimed + // handleTask lifecycle. The additive counters split actual provider + // execution from local-directory parking for throughput and diagnostics. + ActiveTaskCount int64 `json:"active_task_count"` + RunningTaskCount int64 `json:"running_task_count"` + ResourceWaitTaskCount int64 `json:"resource_wait_task_count"` + Agents []string `json:"agents"` // SkippedAgents maps a provider that WAS discovered on this machine to the // reason the last registration round dropped it (version undetectable, // below the minimum supported version). Purely diagnostic, and omitted when @@ -108,17 +113,19 @@ func (d *Daemon) healthHandler(startedAt time.Time) http.HandlerFunc { } resp := HealthResponse{ - Status: status, - PID: os.Getpid(), - OS: runtime.GOOS, - Uptime: time.Since(startedAt).Truncate(time.Second).String(), - DaemonID: d.cfg.DaemonID, - DeviceName: d.cfg.DeviceName, - ServerURL: d.cfg.ServerBaseURL, - CLIVersion: d.cfg.CLIVersion, - ActiveTaskCount: d.activeTasks.Load(), - Agents: agents, - SkippedAgents: d.skippedAgentsSnapshot(), + Status: status, + PID: os.Getpid(), + OS: runtime.GOOS, + Uptime: time.Since(startedAt).Truncate(time.Second).String(), + DaemonID: d.cfg.DaemonID, + DeviceName: d.cfg.DeviceName, + ServerURL: d.cfg.ServerBaseURL, + CLIVersion: d.cfg.CLIVersion, + ActiveTaskCount: d.activeTasks.Load(), + RunningTaskCount: d.runningTasks.Load(), + ResourceWaitTaskCount: d.resourceWaitTasks.Load(), + Agents: agents, + SkippedAgents: d.skippedAgentsSnapshot(), ReloadPendingReason: d.reloadPending(), Workspaces: wsList, diff --git a/server/internal/daemon/health_test.go b/server/internal/daemon/health_test.go index 1797ec1a4..30eb37ed7 100644 --- a/server/internal/daemon/health_test.go +++ b/server/internal/daemon/health_test.go @@ -15,7 +15,7 @@ import ( "github.com/multica-ai/multica/server/internal/daemon/repocache" ) -func TestHealthHandlerReportsCLIVersionAndActiveTaskCount(t *testing.T) { +func TestHealthHandlerReportsCLIVersionAndTaskCounts(t *testing.T) { t.Parallel() d := &Daemon{ @@ -28,7 +28,9 @@ func TestHealthHandlerReportsCLIVersionAndActiveTaskCount(t *testing.T) { workspaces: map[string]*workspaceState{}, logger: slog.Default(), } - d.activeTasks.Store(3) + d.activeTasks.Store(2) + d.runningTasks.Store(1) + d.resourceWaitTasks.Store(1) d.ready.Store(true) // preflight done -> status should be "running" req := httptest.NewRequest(http.MethodGet, "/health", nil) @@ -41,7 +43,9 @@ func TestHealthHandlerReportsCLIVersionAndActiveTaskCount(t *testing.T) { // Decode into a raw map so the test locks in the exact wire-level JSON // keys — the desktop TS client depends on snake_case (cli_version, - // active_task_count), so a silent struct-tag rename must fail here. + // active_task_count), so a silent struct-tag rename must fail here. The + // execution/wait split is additive: active_task_count keeps its ownership + // semantics for old clients and restart barriers. var raw map[string]any if err := json.Unmarshal(rec.Body.Bytes(), &raw); err != nil { t.Fatalf("decode raw response: %v", err) @@ -50,9 +54,15 @@ func TestHealthHandlerReportsCLIVersionAndActiveTaskCount(t *testing.T) { t.Errorf("cli_version key: got %v, want %q", got, want) } // JSON numbers decode to float64 through map[string]any. - if got, want := raw["active_task_count"], float64(3); got != want { + if got, want := raw["active_task_count"], float64(2); got != want { t.Errorf("active_task_count key: got %v, want %v", got, want) } + if got, want := raw["running_task_count"], float64(1); got != want { + t.Errorf("running_task_count key: got %v, want %v", got, want) + } + if got, want := raw["resource_wait_task_count"], float64(1); got != want { + t.Errorf("resource_wait_task_count key: got %v, want %v", got, want) + } if got, want := raw["status"], "running"; got != want { t.Errorf("status key: got %v, want %q", got, want) } @@ -72,8 +82,14 @@ func TestHealthHandlerReportsCLIVersionAndActiveTaskCount(t *testing.T) { if resp.CLIVersion != "v9.9.9" { t.Errorf("CLIVersion: got %q, want %q", resp.CLIVersion, "v9.9.9") } - if resp.ActiveTaskCount != 3 { - t.Errorf("ActiveTaskCount: got %d, want 3", resp.ActiveTaskCount) + if resp.ActiveTaskCount != 2 { + t.Errorf("ActiveTaskCount: got %d, want 2", resp.ActiveTaskCount) + } + if resp.RunningTaskCount != 1 { + t.Errorf("RunningTaskCount: got %d, want 1", resp.RunningTaskCount) + } + if resp.ResourceWaitTaskCount != 1 { + t.Errorf("ResourceWaitTaskCount: got %d, want 1", resp.ResourceWaitTaskCount) } } diff --git a/server/internal/daemon/local_directory_test.go b/server/internal/daemon/local_directory_test.go index d16f3dead..5e6ade895 100644 --- a/server/internal/daemon/local_directory_test.go +++ b/server/internal/daemon/local_directory_test.go @@ -704,6 +704,9 @@ func TestAcquireLocalDirectoryLock_ParentCancellationReportsWaitFailure(t *testi select { case <-parked: + if got := d.resourceWaitTasks.Load(); got != 1 { + t.Fatalf("resource wait count while parked = %d, want 1", got) + } cancel() case <-time.After(2 * time.Second): t.Fatal("task never entered local_directory wait") @@ -724,6 +727,9 @@ func TestAcquireLocalDirectoryLock_ParentCancellationReportsWaitFailure(t *testi if got := failCalls.Load(); got != 1 { t.Fatalf("fail callback calls = %d, want 1", got) } + if got := d.resourceWaitTasks.Load(); got != 0 { + t.Fatalf("resource wait count after cancellation = %d, want 0", got) + } } func TestAcquireLocalDirectoryLock_EarlyFailureReportsWithCancelledParent(t *testing.T) { diff --git a/server/internal/handler/squad.go b/server/internal/handler/squad.go index 9b0a88d69..07a29ede1 100644 --- a/server/internal/handler/squad.go +++ b/server/internal/handler/squad.go @@ -605,13 +605,13 @@ func deriveSquadMemberStatus( archived bool, runtimeStatus pgtype.Text, lastSeen pgtype.Timestamptz, - hasActiveTask bool, + hasWorkingTask bool, now time.Time, ) string { if archived { return "archived" } - if hasActiveTask { + if hasWorkingTask { return "working" } if !runtimeStatus.Valid { @@ -630,10 +630,10 @@ func deriveSquadMemberStatus( } // ListSquadMemberStatus returns one entry per squad member with derived -// status, the issues each agent member is currently running, and the last -// observed runtime activity. The endpoint is read-only and inherits the -// workspace-membership guard from the route middleware — any member of the -// workspace can read it. +// status, the issues each agent member is currently running or waiting to run, +// and the last observed runtime activity. The endpoint is read-only and +// inherits the workspace-membership guard from the route middleware — any +// member of the workspace can read it. func (h *Handler) ListSquadMemberStatus(w http.ResponseWriter, r *http.Request) { squad, _, ok := h.loadSquadInWorkspace(w, r) if !ok { @@ -655,7 +655,7 @@ func (h *Handler) ListSquadMemberStatus(w http.ResponseWriter, r *http.Request) type memberAcc struct { response SquadMemberStatusResponse archived bool - hasActiveTask bool + hasWorkingTask bool runtimeStatus pgtype.Text runtimeSeenAt pgtype.Timestamptz latestActiveAt pgtype.Timestamptz @@ -685,13 +685,15 @@ func (h *Handler) ListSquadMemberStatus(w http.ResponseWriter, r *http.Request) continue } - // A dispatched/running task occupies an agent slot even when it - // has no associated issue (chat / quick-create tasks set - // agent_task_queue.issue_id = NULL). The `working` bucket is - // defined by task presence, not by whether we can render an - // issue link, so flag the agent here regardless of issue_id. + // Keep waiting_local_directory rows available for issue visibility, + // but only dispatched/running work drives the `working` bucket. A + // working task may have no issue (chat / quick-create), so decide the + // bucket independently from whether an issue link can be rendered. if row.TaskID.Valid { - entry.hasActiveTask = true + if row.TaskStatus.Valid && + (row.TaskStatus.String == "dispatched" || row.TaskStatus.String == "running") { + entry.hasWorkingTask = true + } if row.TaskIssueID.Valid { brief := SquadActiveIssueBrief{ @@ -725,7 +727,7 @@ func (h *Handler) ListSquadMemberStatus(w http.ResponseWriter, r *http.Request) entry.archived, entry.runtimeStatus, entry.runtimeSeenAt, - entry.hasActiveTask, + entry.hasWorkingTask, now, ) entry.response.Status = &status diff --git a/server/internal/handler/squad_member_status_test.go b/server/internal/handler/squad_member_status_test.go index adfa29e1c..5c6a06bbb 100644 --- a/server/internal/handler/squad_member_status_test.go +++ b/server/internal/handler/squad_member_status_test.go @@ -1,6 +1,10 @@ package handler import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" "testing" "time" @@ -19,15 +23,15 @@ func TestDeriveSquadMemberStatus(t *testing.T) { tsNone := pgtype.Timestamptz{} cases := []struct { - name string - archived bool - runtimeStatus pgtype.Text - lastSeen pgtype.Timestamptz - hasActiveTask bool - want string + name string + archived bool + runtimeStatus pgtype.Text + lastSeen pgtype.Timestamptz + hasWorkingTask bool + want string }{ - {"active wins over offline runtime", false, offline, tsAgo(time.Hour), true, "working"}, - {"active wins over missing runtime", false, missing, tsNone, true, "working"}, + {"working wins over offline runtime", false, offline, tsAgo(time.Hour), true, "working"}, + {"working wins over missing runtime", false, missing, tsNone, true, "working"}, {"online runtime, no task", false, online, tsAgo(2 * time.Second), false, "idle"}, {"offline runtime, recent heartbeat", false, offline, tsAgo(2 * time.Minute), false, "unstable"}, {"offline runtime, stale heartbeat", false, offline, tsAgo(2 * time.Hour), false, "offline"}, @@ -43,10 +47,98 @@ func TestDeriveSquadMemberStatus(t *testing.T) { for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { - got := deriveSquadMemberStatus(tc.archived, tc.runtimeStatus, tc.lastSeen, tc.hasActiveTask, now) + got := deriveSquadMemberStatus(tc.archived, tc.runtimeStatus, tc.lastSeen, tc.hasWorkingTask, now) if got != tc.want { t.Fatalf("deriveSquadMemberStatus = %q, want %q", got, tc.want) } }) } } + +func TestListSquadMemberStatusRetainsWaiterIssueWithoutWorkingStatus(t *testing.T) { + if testHandler == nil || testPool == nil { + t.Skip("database not available") + } + + ctx := context.Background() + agentID := createHandlerTestAgent(t, "squad-status-local-directory-waiter", []byte(`{}`)) + squadID := createCommentTriggerPreviewSquad(t, "Squad Status Local Directory Waiter", agentID) + issueID := createCommentTriggerPreviewIssue(t, "Directory waiter remains visible", "agent", agentID) + + if _, err := testPool.Exec(ctx, ` + INSERT INTO squad_member (squad_id, member_type, member_id) + VALUES ($1, 'agent', $2) + `, squadID, agentID); err != nil { + t.Fatalf("insert squad member: %v", err) + } + t.Cleanup(func() { + testPool.Exec(context.Background(), `DELETE FROM squad_member WHERE squad_id = $1 AND member_id = $2`, squadID, agentID) + }) + + var taskID string + if err := testPool.QueryRow(ctx, ` + INSERT INTO agent_task_queue (agent_id, runtime_id, issue_id, status, priority, dispatched_at) + VALUES ($1, $2, $3, 'waiting_local_directory', 0, now()) + RETURNING id + `, agentID, handlerTestRuntimeID(t), issueID).Scan(&taskID); err != nil { + t.Fatalf("insert local-directory waiter: %v", err) + } + t.Cleanup(func() { + testPool.Exec(context.Background(), `DELETE FROM agent_task_queue WHERE id = $1`, taskID) + }) + + rows, err := testHandler.Queries.ListSquadMemberStatusRows(ctx, parseUUID(squadID)) + if err != nil { + t.Fatalf("ListSquadMemberStatusRows with waiter: %v", err) + } + if len(rows) != 1 || !rows[0].TaskID.Valid || rows[0].TaskStatus.String != "waiting_local_directory" { + t.Fatalf("waiter-only squad member rows = %+v, want the waiting task retained", rows) + } + + assertSquadMember := func(wantWorking bool) { + t.Helper() + w := httptest.NewRecorder() + testHandler.ListSquadMemberStatus(w, squadScopeReq( + "", + http.MethodGet, + "/api/squads/"+squadID+"/members/status", + nil, + map[string]string{"id": squadID}, + )) + if w.Code != http.StatusOK { + t.Fatalf("ListSquadMemberStatus: got %d: %s", w.Code, w.Body.String()) + } + var resp SquadMemberStatusListResponse + if err := json.NewDecoder(w.Body).Decode(&resp); err != nil { + t.Fatalf("decode squad member status: %v", err) + } + if len(resp.Members) != 1 { + t.Fatalf("squad members = %+v, want one member", resp.Members) + } + member := resp.Members[0] + if member.Status == nil { + t.Fatal("agent member status is nil") + } + if gotWorking := *member.Status == "working"; gotWorking != wantWorking { + t.Fatalf("member status = %q, want working=%t", *member.Status, wantWorking) + } + if len(member.ActiveIssues) != 1 || member.ActiveIssues[0].IssueID != issueID { + t.Fatalf("active issues = %+v, want waiting issue %s", member.ActiveIssues, issueID) + } + } + assertSquadMember(false) + + if _, err := testPool.Exec(ctx, ` + UPDATE agent_task_queue SET status = 'dispatched' WHERE id = $1 + `, taskID); err != nil { + t.Fatalf("promote waiter to dispatched: %v", err) + } + rows, err = testHandler.Queries.ListSquadMemberStatusRows(ctx, parseUUID(squadID)) + if err != nil { + t.Fatalf("ListSquadMemberStatusRows with dispatched task: %v", err) + } + if len(rows) != 1 || !rows[0].TaskID.Valid { + t.Fatalf("dispatched squad member rows = %+v, want one working task row", rows) + } + assertSquadMember(true) +} diff --git a/server/internal/service/task.go b/server/internal/service/task.go index 1878f6b72..ee5f63d73 100644 --- a/server/internal/service/task.go +++ b/server/internal/service/task.go @@ -3233,6 +3233,11 @@ func (s *TaskService) StartTask(ctx context.Context, taskID pgtype.UUID) (*db.Ag slog.Info("task started", "task_id", util.UUIDToString(task.ID), "issue_id", util.UUIDToString(task.IssueID)) s.captureTaskStarted(ctx, task) + // A local-directory waiter was reconciled out of the persisted working + // status while parked. Restore working as soon as it enters running; the + // normal dispatched -> running path is already working, so this is + // intentionally idempotent there. + s.ReconcileAgentStatus(ctx, task.AgentID) // Tell every connected workspace WS client that this task transitioned // (dispatched | waiting_local_directory) → running. Without this, the // workspace-wide `agentTaskSnapshot` query only refreshes on the 30s @@ -3316,6 +3321,11 @@ func (s *TaskService) MarkTaskWaitingLocalDirectory(ctx context.Context, taskID "issue_id", util.UUIDToString(task.IssueID), "reason", reason, ) + // waiting_local_directory is owned/queued work, not executing work. The + // claim path marked the agent working while the row was dispatched, so + // reconcile immediately when it parks instead of leaving that persisted + // status stale until a terminal transition. + s.ReconcileAgentStatus(ctx, task.AgentID) s.broadcastTaskEvent(ctx, protocol.EventTaskWaitingLocalDirectory, task) return &task, nil } @@ -4587,7 +4597,9 @@ func (s *TaskService) ReportProgress(ctx context.Context, taskID string, workspa }) } -// ReconcileAgentStatus refreshes agent status from the current active task set. +// ReconcileAgentStatus refreshes agent status from the current working task +// set. The query returns no row when the status is already correct, which +// avoids rewriting updated_at and broadcasting a zero-information event. func (s *TaskService) ReconcileAgentStatus(ctx context.Context, agentID pgtype.UUID) { agent, err := s.Queries.RefreshAgentStatusFromTasks(ctx, agentID) if err != nil { diff --git a/server/pkg/db/generated/agent.sql.go b/server/pkg/db/generated/agent.sql.go index 257d171da..c100954e0 100644 --- a/server/pkg/db/generated/agent.sql.go +++ b/server/pkg/db/generated/agent.sql.go @@ -1547,6 +1547,10 @@ SELECT count(*) FROM agent_task_queue WHERE agent_id = $1 AND status IN ('dispatched', 'running', 'waiting_local_directory') ` +// waiting_local_directory remains capacity-bearing until resume admission has +// an atomic reservation/CAS gate. Consequently a waiter-only agent is reported +// idle by RefreshAgentStatusFromTasks but still cannot claim additional work; +// removing it here alone could exceed max_concurrent_tasks when it resumes. func (q *Queries) CountRunningTasks(ctx context.Context, agentID pgtype.UUID) (int64, error) { row := q.db.QueryRow(ctx, countRunningTasks, agentID) var count int64 @@ -6033,16 +6037,25 @@ func (q *Queries) RecoverOrphanedTasksForRuntime(ctx context.Context, runtimeID } const refreshAgentStatusFromTasks = `-- name: RefreshAgentStatusFromTasks :one +WITH desired AS ( + SELECT CASE WHEN EXISTS ( + SELECT 1 FROM agent_task_queue q + WHERE q.agent_id = $1 AND q.status IN ('dispatched', 'running') + ) THEN 'working' ELSE 'idle' END AS status +) UPDATE agent AS a -SET status = CASE WHEN EXISTS ( - SELECT 1 FROM agent_task_queue q - WHERE q.agent_id = a.id AND q.status IN ('dispatched', 'running', 'waiting_local_directory') -) THEN 'working' ELSE 'idle' END, +SET status = desired.status, updated_at = now() -WHERE a.id = $1 -RETURNING id, workspace_id, name, avatar_url, runtime_mode, runtime_config, visibility, status, max_concurrent_tasks, owner_id, created_at, updated_at, description, runtime_id, instructions, archived_at, archived_by, custom_env, custom_args, mcp_config, model, thinking_level, composio_toolkit_allowlist, permission_mode, kind, system_key, disabled_runtime_skills, service_tier +FROM desired +WHERE a.id = $1 AND a.status IS DISTINCT FROM desired.status +RETURNING a.id, a.workspace_id, a.name, a.avatar_url, a.runtime_mode, a.runtime_config, a.visibility, a.status, a.max_concurrent_tasks, a.owner_id, a.created_at, a.updated_at, a.description, a.runtime_id, a.instructions, a.archived_at, a.archived_by, a.custom_env, a.custom_args, a.mcp_config, a.model, a.thinking_level, a.composio_toolkit_allowlist, a.permission_mode, a.kind, a.system_key, a.disabled_runtime_skills, a.service_tier ` +// Persisted agent.status has no queued/resource-wait bucket. Keep dispatched +// as working because the daemon is actively preparing that task, but do not +// let a task parked on a local_directory mutex masquerade as executing work. +// The status guard makes a correct status a no-op: :one returns no rows, so +// callers neither rewrite updated_at nor publish a redundant status event. func (q *Queries) RefreshAgentStatusFromTasks(ctx context.Context, id pgtype.UUID) (Agent, error) { row := q.db.QueryRow(ctx, refreshAgentStatusFromTasks, id) var i Agent diff --git a/server/pkg/db/generated/squad.sql.go b/server/pkg/db/generated/squad.sql.go index 620a9f426..49c3feca9 100644 --- a/server/pkg/db/generated/squad.sql.go +++ b/server/pkg/db/generated/squad.sql.go @@ -410,10 +410,12 @@ type ListSquadMemberStatusRowsRow struct { } // Per-row join used to build the squad-members status view. One row per -// (squad_member × active_task); members with no active task return a +// (squad_member × in_flight_task); members with no in-flight task return a // single row with NULL task_* columns. Human members and agent members // with no agent row also return one row with NULL agent_/runtime_ columns. -// The handler aggregates rows by member_id. +// waiting_local_directory stays in the row set so its issue remains visible, +// but the handler only treats dispatched/running rows as working because the +// squad status vocabulary has no queued bucket. func (q *Queries) ListSquadMemberStatusRows(ctx context.Context, squadID pgtype.UUID) ([]ListSquadMemberStatusRowsRow, error) { rows, err := q.db.Query(ctx, listSquadMemberStatusRows, squadID) if err != nil { diff --git a/server/pkg/db/queries/agent.sql b/server/pkg/db/queries/agent.sql index 8be051822..649dcfce4 100644 --- a/server/pkg/db/queries/agent.sql +++ b/server/pkg/db/queries/agent.sql @@ -1234,6 +1234,10 @@ ORDER BY chat_finalize_deferred_at LIMIT @max_per_tick::int; -- name: CountRunningTasks :one +-- waiting_local_directory remains capacity-bearing until resume admission has +-- an atomic reservation/CAS gate. Consequently a waiter-only agent is reported +-- idle by RefreshAgentStatusFromTasks but still cannot claim additional work; +-- removing it here alone could exceed max_concurrent_tasks when it resumes. SELECT count(*) FROM agent_task_queue WHERE agent_id = $1 AND status IN ('dispatched', 'running', 'waiting_local_directory'); @@ -1762,14 +1766,23 @@ WHERE id = $1 RETURNING *; -- name: RefreshAgentStatusFromTasks :one +-- Persisted agent.status has no queued/resource-wait bucket. Keep dispatched +-- as working because the daemon is actively preparing that task, but do not +-- let a task parked on a local_directory mutex masquerade as executing work. +-- The status guard makes a correct status a no-op: :one returns no rows, so +-- callers neither rewrite updated_at nor publish a redundant status event. +WITH desired AS ( + SELECT CASE WHEN EXISTS ( + SELECT 1 FROM agent_task_queue q + WHERE q.agent_id = $1 AND q.status IN ('dispatched', 'running') + ) THEN 'working' ELSE 'idle' END AS status +) UPDATE agent AS a -SET status = CASE WHEN EXISTS ( - SELECT 1 FROM agent_task_queue q - WHERE q.agent_id = a.id AND q.status IN ('dispatched', 'running', 'waiting_local_directory') -) THEN 'working' ELSE 'idle' END, +SET status = desired.status, updated_at = now() -WHERE a.id = $1 -RETURNING *; +FROM desired +WHERE a.id = $1 AND a.status IS DISTINCT FROM desired.status +RETURNING a.*; -- name: GetAgentBySystemKey :one -- Resolves a workspace's built-in agent by its stable system_key. This is the diff --git a/server/pkg/db/queries/squad.sql b/server/pkg/db/queries/squad.sql index b89f9cf70..567c8489c 100644 --- a/server/pkg/db/queries/squad.sql +++ b/server/pkg/db/queries/squad.sql @@ -135,10 +135,12 @@ WHERE assignee_type = 'squad' AND assignee_id = $1; -- name: ListSquadMemberStatusRows :many -- Per-row join used to build the squad-members status view. One row per --- (squad_member × active_task); members with no active task return a +-- (squad_member × in_flight_task); members with no in-flight task return a -- single row with NULL task_* columns. Human members and agent members -- with no agent row also return one row with NULL agent_/runtime_ columns. --- The handler aggregates rows by member_id. +-- waiting_local_directory stays in the row set so its issue remains visible, +-- but the handler only treats dispatched/running rows as working because the +-- squad status vocabulary has no queued bucket. SELECT sm.id AS squad_member_id, sm.member_type AS member_type,