MUL-5832: separate directory waits from working signals (#6553)

* fix(daemon): separate directory waits from working status Co-authored-by: multica-agent <[email protected]> * fix(status): address directory wait review feedback Co-authored-by: multica-agent <[email protected]> --------- Co-authored-by: Eve <[email protected]> Co-authored-by: multica-agent <[email protected]>
2026-11-11 02:46:45 +08:00 · 2026-08-07 12:40:12 +08:00
parent 45027c1db6
commit 4900a4a85b
14 changed files with 393 additions and 71 deletions
@@ -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)
@@ -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
@@ -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       // "<prefix>/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      // "<prefix>/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
@@ -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()

@@ -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,
@@ -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)
 	}
 }

@@ -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) {
@@ -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
@@ -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)
+}
@@ -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 {
@@ -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
@@ -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 {
@@ -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
@@ -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,