Verify each finding against the current code and only fix it if needed.

Nitpick comments:
In `@internal/intg/distr/zombie_recovery_test.go`:
- Line 54: The test bumped the scheduler timeout from 30s to 90s by calling
f.startScheduler(90 * time.Second), which masks flakiness; change this to a
deterministic fix: restore a reasonable default (e.g., 30 * time.Second) and
either (a) make the timeout a named constant (SchedulerStartTimeout) that can be
overridden by CI via an env var or test flag, or (b) avoid a large static sleep
by replacing the single long timeout usage in the test with a polling/wait
helper that retries until the scheduler is confirmed running (use existing
wait-for-condition helper or implement a short-interval loop with context
deadline). Update the call to f.startScheduler(...) to use the constant or the
wait helper and ensure any CI override is documented in the test setup.

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@internal/core/exec/distributed.go`:
- Around line 195-200: The current code fabricates an attempt key by falling
back from status.Root to status.DAGRun() which produces incorrect keys for
legacy sub-DAG statuses; update the logic in the block that computes the attempt
key (the lines using root, status.Root, status.DAGRun() and GenerateAttemptKey)
so that if status.Root is zero/unknown you return an empty string instead of
using status.DAGRun(); this prevents LeaseMatchesStatus from rejecting valid
leases on attempt-key mismatch—alternatively, thread an explicit root reference
from the caller into the function if callers can supply the true root instead of
constructing it here.

In `@internal/persis/filedistributed/active_distributed_run_store.go`:
- Around line 99-104: In ListAll (active_distributed_run_store.go) the error
path that calls readJSONFile(path, &record) currently returns on any non-ENOENT
error; change this to log or collect the decode/read error and continue the loop
so a single corrupted/truncated JSON file doesn't abort the entire
scan—specifically, inside the block handling readJSONFile errors, replace the
return nil, err with a non-fatal handling (e.g., processLogger.Warnf or similar)
that records path and err and then continue iterating, ensuring ListAll still
returns the successfully read records.

In `@internal/persis/filedistributed/stores_test.go`:
- Around line 156-208: The test is flaky because ClaimNext(ctx,
exec.DispatchTaskClaim{...}) has no queue filter and may claim either enqueued
task; fix by making the claim deterministic: add a queue filter to the
DispatchTaskClaim (e.g. set QueueName: "queue-a" or the appropriate queue filter
field on the exec.DispatchTaskClaim passed into store.ClaimNext) so ClaimNext
will deterministically select the task from "queue-a" and the subsequent
CountOutstandingByQueue / DeleteClaim / HasOutstandingAttempt assertions remain
valid.

In `@internal/service/coordinator/handler_test.go`:
- Around line 511-513: The test calls dispatchStore.CountOutstandingByQueue with
an empty string which is queue-scoped; change the call to use the actual test
queue name (e.g., "test-queue" or the variable used to create tasks) so it
asserts outstanding tasks for that queue specifically. Update the invocation of
CountOutstandingByQueue(...) and the subsequent require/assert to check the
count for "test-queue" (referencing CountOutstandingByQueue and the test
function in handler_test.go) instead of "".

In `@internal/service/coordinator/handler.go`:
- Around line 1127-1137: The current early-return treats any identical Status
enum as a noop and skips post-write side effects; change the logic in the
ReportStatus handling around h.remoteStatusDecision so that identical terminal
statuses are only considered noop when the full incoming payload (including
node/chat fields and any metadata used by persistChatMessages, lease/index sync
and Write) is unchanged, or alternatively perform the required post-write side
effects even on the noop path; update the block that currently returns on noop
(after calling h.remoteStatusDecision and h.logRejectedRemoteStatusUpdate) to
compare the full payload (not just latest.Status == incoming.Status) or invoke
the same Write/persistChatMessages and lease/index sync routines when status is
terminal but other payload pieces differ, ensuring follow-up terminal reports
with extra node/chat data still persist and trigger cleanup retries.
- Around line 1784-1789: The current early return when
h.activeDistributedRunStore != nil prevents the orphaned-run detector from
running; change the logic in the handler so that after calling
detectIndexedDistributedStatuses(ctx, now) you still invoke
detectOrphanedDistributedStatuses(ctx, now) as a fallback (i.e., remove the
return or call detectOrphanedDistributedStatuses unconditionally after
detectIndexedDistributedStatuses) so runs that lost their lease or whose index
rows failed to materialize (see syncActiveDistributedRunFromStatus) are still
detected; update the block that references h.activeDistributedRunStore,
detectIndexedDistributedStatuses, and detectOrphanedDistributedStatuses
accordingly.
- Around line 592-629: markPreparedAttemptDispatchFailed currently marks
prepared.attempt as Failed but leaves the opened handle cached in
h.openAttempts, which can leak resources; update
markPreparedAttemptDispatchFailed to always evict and close the prepared attempt
handle (prepared.attempt) from the coordinator's cache (h.openAttempts) whenever
prepared != nil && prepared.newlyCreated && prepared.attempt != nil and you are
exiting the function (both on early returns and after successfully writing the
failed status), by removing the entry keyed by task.AttemptKey from
h.openAttempts and calling the appropriate close/release method on
prepared.attempt; ensure you use the same mutex/lock that protects
h.openAttempts (e.g., h.openAttemptsMu) when mutating the map so there are no
races.

---

Nitpick comments:
In `@internal/persis/filedistributed/dispatch_task_store.go`:
- Around line 174-225: Both CountOutstandingByQueue and HasOutstandingAttempt
hold s.mu and call recycleExpiredClaims then scanOutstandingLocked which
JSON-decodes every file under pending/ and claims/, causing full-store
serialization; change the implementation to avoid long-held lock by either (a)
create a short-locked snapshot: while holding s.mu only call
recycleExpiredClaims and collect the list of filenames/metadata (not full
decode) or copy necessary in-memory indexes, then release s.mu and perform the
expensive JSON-decode/scan on that snapshot to compute counts/flags; or (b)
maintain and update lightweight in-memory indexes (per-queue outstanding counts
and attemptKey map) inside mutating methods (e.g., ClaimNext, Add/Remove
helpers) so CountOutstandingByQueue and HasOutstandingAttempt can read those
indexes under s.mu cheaply; update references to scanOutstandingLocked,
recycleExpiredClaims, CountOutstandingByQueue, HasOutstandingAttempt and
ClaimNext accordingly so scans are no longer performed while holding the main
mutex.

In `@internal/runtime/manager.go`:
- Around line 234-242: The code unconditionally calls attempt.ReadDAG even when
its result is only used for local running status checks; move the ReadDAG call
so it only executes when st.Status == core.Running &&
isLocalWorkerID(st.WorkerID) and dagRun.ID == st.DAGRunID, so change the block
around dagRun.ID == st.DAGRunID to first check st.Status and isLocalWorkerID
before invoking attempt.ReadDAG, then pass the obtained dag into
m.resolveRunningStatus; this avoids unnecessary filesystem I/O and prevents
unrelated DAG-read errors from occurring on non-local or non-running
saved-status lookups.