From e1cf99f821d91f24ffd8d36c1ec75caa03495bb5 Mon Sep 17 00:00:00 2001 From: point-source <47544779+point-source@users.noreply.github.com> Date: Wed, 8 Jul 2026 15:37:02 -0700 Subject: [PATCH] fix(executors): truthful executor rows, startup reconciliation, resume-safe cleanup (#1597) (#1608) Co-authored-by: Carlos Florencio --- .../runtime/agentctl/launcher/launcher.go | 12 + .../agent/runtime/lifecycle/executor_ssh.go | 6 + .../agent/runtime/lifecycle/liveness.go | 49 ++++ .../agent/runtime/lifecycle/liveness_test.go | 99 +++++++ .../agent/runtime/lifecycle/liveness_unix.go | 31 ++ .../runtime/lifecycle/liveness_windows.go | 24 ++ .../agent/runtime/lifecycle/manager.go | 16 + .../runtime/lifecycle/manager_interaction.go | 8 + .../agent/runtime/lifecycle/persistence.go | 136 ++++++++- .../runtime/lifecycle/persistence_test.go | 203 +++++++++++++ .../runtime/lifecycle/resume_safety_test.go | 101 +++++++ apps/backend/internal/backendapp/adapters.go | 10 + apps/backend/internal/backendapp/agentctl.go | 3 + apps/backend/internal/backendapp/agents.go | 4 + apps/backend/internal/common/config/config.go | 6 + .../orchestrator/event_handlers_test.go | 19 +- .../executor/executor_resume_test.go | 20 +- .../orchestrator/reconcile_liveness.go | 64 ++++ .../orchestrator/reconcile_restart_test.go | 229 +++++++++++++++ apps/backend/internal/orchestrator/service.go | 51 +++- .../internal/orchestrator/task_operations.go | 14 + .../orchestrator/task_operations_test.go | 42 ++- .../task/handlers/process_handlers_test.go | 3 + apps/backend/internal/task/models/models.go | 52 ++-- .../internal/task/models/resume_safety.go | 71 +++++ .../task/models/resume_safety_test.go | 56 ++++ .../internal/task/repository/interface.go | 6 + .../task/repository/sqlite/base_migrations.go | 5 + .../task/repository/sqlite/base_schema.go | 1 + .../task/repository/sqlite/executor.go | 44 ++- .../sqlite/executor_localpid_test.go | 275 ++++++++++++++++++ .../repository/sqlite/postgres_schema_test.go | 58 ++++ apps/web/e2e/helpers/type-while-busy.ts | 26 ++ apps/web/e2e/tests/chat/message-queue.spec.ts | 30 +- .../session/pause-resume-recovery.spec.ts | 139 +++++++++ ...-runtime-cleanup-uses-executors-running.md | 39 ++- docs/decisions/INDEX.md | 2 +- 37 files changed, 1855 insertions(+), 99 deletions(-) create mode 100644 apps/backend/internal/agent/runtime/lifecycle/liveness.go create mode 100644 apps/backend/internal/agent/runtime/lifecycle/liveness_test.go create mode 100644 apps/backend/internal/agent/runtime/lifecycle/liveness_unix.go create mode 100644 apps/backend/internal/agent/runtime/lifecycle/liveness_windows.go create mode 100644 apps/backend/internal/agent/runtime/lifecycle/resume_safety_test.go create mode 100644 apps/backend/internal/orchestrator/reconcile_liveness.go create mode 100644 apps/backend/internal/orchestrator/reconcile_restart_test.go create mode 100644 apps/backend/internal/task/models/resume_safety.go create mode 100644 apps/backend/internal/task/models/resume_safety_test.go create mode 100644 apps/backend/internal/task/repository/sqlite/executor_localpid_test.go create mode 100644 apps/web/e2e/helpers/type-while-busy.ts create mode 100644 apps/web/e2e/tests/session/pause-resume-recovery.spec.ts diff --git a/apps/backend/internal/agent/runtime/agentctl/launcher/launcher.go b/apps/backend/internal/agent/runtime/agentctl/launcher/launcher.go index 0262627de1..ad514cad75 100644 --- a/apps/backend/internal/agent/runtime/agentctl/launcher/launcher.go +++ b/apps/backend/internal/agent/runtime/agentctl/launcher/launcher.go @@ -96,6 +96,18 @@ func (l *Launcher) Port() int { return l.port } +// Pid returns the OS process id of the running agentctl control-server, or 0 if +// it has not started yet. This is the host-local liveness handle recorded in +// executors_running.local_pid for local/standalone rows (#1597 truthful executor rows). +func (l *Launcher) Pid() int { + l.mu.Lock() + defer l.mu.Unlock() + if l.cmd == nil || l.cmd.Process == nil { + return 0 + } + return l.cmd.Process.Pid +} + // AuthToken returns the auth token retrieved via handshake. // Used by the backend to authenticate requests to agentctl. func (l *Launcher) AuthToken() string { diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_ssh.go b/apps/backend/internal/agent/runtime/lifecycle/executor_ssh.go index 9c2d6f49ec..0d746c4c1c 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/executor_ssh.go +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_ssh.go @@ -465,6 +465,12 @@ func (r *SSHExecutor) ResumeRemoteInstance(ctx context.Context, req *ExecutorCre return fmt.Errorf("ssh resume: connect: %w", err) } + // SSH liveness is judged on the REMOTE host: MetadataKeySSHRemoteAgentctlPID + // (mirrored into executors_running.pid) is a pid on the remote box, so it is + // probed with a remote `kill -0` over this SSH client — never with a local + // os.FindProcess. The runtime-aware row predicate (RowProcessLiveness in + // liveness.go) deliberately returns Unknown for SSH rows so nothing applies a + // host-local check here (#1597 runtime-aware liveness). pid, _ := strconv.Atoi(pidStr) if !isRemoteAgentctlAlive(ctx, client, pid) { _ = client.Close() diff --git a/apps/backend/internal/agent/runtime/lifecycle/liveness.go b/apps/backend/internal/agent/runtime/lifecycle/liveness.go new file mode 100644 index 0000000000..34f1df01de --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/liveness.go @@ -0,0 +1,49 @@ +package lifecycle + +import ( + "github.com/kandev/kandev/internal/agentruntime" + "github.com/kandev/kandev/internal/task/models" +) + +// RowProcessLiveness reports the liveness of the OS process backing a row, +// branching on the row's runtime so that a host-local process check is NEVER +// applied to a remote row (#1597: executors_running.pid is the agentctl PID +// on the REMOTE host for SSH rows — never overload it with a local pid). +// +// It returns models.ProcessLiveness — a low-dependency type the orchestrator and +// task service can consume without importing runtime/lifecycle — so cleanup and +// reconciliation paths can classify a row through a small adapter method while +// the actual host-local probe stays in this platform-split package. +// +// - local/standalone: judged by local_pid on this host. +// - SSH/remote: Unknown here. The agent process lives on another host and its +// liveness is owned by the SSH executor's remote `kill -0` path; the SSH +// `pid` column is a REMOTE pid and must never be probed with a local +// os.FindProcess. +// - docker / anything else: Unknown for now (out of scope for this batch). +func RowProcessLiveness(row *models.ExecutorRunning) models.ProcessLiveness { + if row == nil { + return models.ProcessLivenessUnknown + } + if !isLocalRuntime(row.Runtime) { + // Never run a local-process existence check against a remote row. + return models.ProcessLivenessUnknown + } + if row.LocalPID <= 0 { + // A local row without a populated handle can't be judged — don't call it + // dead (that would risk pruning a row before its handle lands). + return models.ProcessLivenessUnknown + } + if isLocalPIDAlive(row.LocalPID) { + return models.ProcessLivenessAlive + } + return models.ProcessLivenessDead +} + +// isLocalRuntime reports whether a row's runtime runs its agent process on THIS +// host, so a local os.FindProcess check is meaningful. Standalone is the only +// local runtime; SSH/docker/remote-docker/sprites run the process elsewhere. +// An empty/unknown runtime is treated as non-local (Unknown), never probed. +func isLocalRuntime(runtime agentruntime.Runtime) bool { + return runtime == agentruntime.RuntimeStandalone +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/liveness_test.go b/apps/backend/internal/agent/runtime/lifecycle/liveness_test.go new file mode 100644 index 0000000000..de03fd952e --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/liveness_test.go @@ -0,0 +1,99 @@ +package lifecycle + +import ( + "os" + "os/exec" + "testing" + + "github.com/kandev/kandev/internal/agentruntime" + "github.com/kandev/kandev/internal/task/models" +) + +// TestRowProcessLivenessLocalAliveVsDead proves a live local row is +// distinguishable from a dead one from the row alone (#1597). +func TestRowProcessLivenessLocalAliveVsDead(t *testing.T) { + // This process is, by definition, alive. + alive := &models.ExecutorRunning{ + SessionID: "s-alive", + Runtime: agentruntime.RuntimeStandalone, + LocalPID: os.Getpid(), + } + if got := RowProcessLiveness(alive); got != models.ProcessLivenessAlive { + t.Fatalf("live standalone row: got %v, want models.ProcessLivenessAlive", got) + } + + dead := &models.ExecutorRunning{ + SessionID: "s-dead", + Runtime: agentruntime.RuntimeStandalone, + LocalPID: spawnAndReapPID(t), + } + if got := RowProcessLiveness(dead); got != models.ProcessLivenessDead { + t.Fatalf("dead standalone row: got %v, want models.ProcessLivenessDead", got) + } +} + +// TestRowProcessLivenessNeverProbesSSH is the no-regression guard: an SSH row's +// pid is a REMOTE pid, so the local predicate must never judge it — even when +// that pid value happens to match a live LOCAL process. It returns Unknown and +// defers to the SSH executor's remote kill -0 path (#1597 runtime-aware liveness). +func TestRowProcessLivenessNeverProbesSSH(t *testing.T) { + sshRow := &models.ExecutorRunning{ + SessionID: "s-ssh", + Runtime: agentruntime.RuntimeSSH, + // Deliberately set PID to THIS live local process. A local check would + // wrongly report Alive; a runtime-aware one must report Unknown. + PID: os.Getpid(), + LocalPID: 0, + } + if got := RowProcessLiveness(sshRow); got != models.ProcessLivenessUnknown { + t.Fatalf("ssh row must not be judged by a local check: got %v, want models.ProcessLivenessUnknown", got) + } +} + +// TestRowProcessLivenessUnknownCases covers rows that cannot be judged locally: +// docker/remote runtimes, an unpopulated local handle, and a nil row. +func TestRowProcessLivenessUnknownCases(t *testing.T) { + cases := []struct { + name string + row *models.ExecutorRunning + }{ + {"nil", nil}, + {"docker", &models.ExecutorRunning{Runtime: agentruntime.RuntimeDocker, LocalPID: os.Getpid()}}, + {"empty-runtime", &models.ExecutorRunning{Runtime: "", LocalPID: os.Getpid()}}, + {"local-no-handle", &models.ExecutorRunning{Runtime: agentruntime.RuntimeStandalone, LocalPID: 0}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := RowProcessLiveness(tc.row); got != models.ProcessLivenessUnknown { + t.Fatalf("got %v, want models.ProcessLivenessUnknown", got) + } + }) + } +} + +func TestIsLocalPIDAlive(t *testing.T) { + if !isLocalPIDAlive(os.Getpid()) { + t.Fatal("current process should be reported alive") + } + if isLocalPIDAlive(0) || isLocalPIDAlive(-1) { + t.Fatal("non-positive pids must be reported dead") + } + if isLocalPIDAlive(spawnAndReapPID(t)) { + t.Fatal("a reaped pid should be reported dead") + } +} + +// spawnAndReapPID starts a trivial child, waits for it to fully exit and be +// reaped, and returns its (now-dead) pid. Reaping avoids leaving a zombie that a +// signal-0 probe would still see as alive. +func spawnAndReapPID(t *testing.T) int { + t.Helper() + cmd := exec.Command("true") + if err := cmd.Start(); err != nil { + // Fall back to a very high pid unlikely to be in use. + return 0x7FFFFFF0 + } + pid := cmd.Process.Pid + _ = cmd.Wait() + return pid +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/liveness_unix.go b/apps/backend/internal/agent/runtime/lifecycle/liveness_unix.go new file mode 100644 index 0000000000..e849a826f2 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/liveness_unix.go @@ -0,0 +1,31 @@ +//go:build !windows + +package lifecycle + +import ( + "errors" + "os" + "syscall" +) + +// isLocalPIDAlive reports whether a process with the given pid currently exists +// on this host. It sends signal 0, which performs existence/permission checking +// without delivering a signal: a nil error means the process is alive, ESRCH +// means it is gone, and EPERM means it exists but is owned by another user +// (still alive). Only ever called for local/standalone rows — never for a remote +// SSH pid (see RowProcessLiveness). +func isLocalPIDAlive(pid int) bool { + if pid <= 0 { + return false + } + proc, err := os.FindProcess(pid) + if err != nil { + return false + } + err = proc.Signal(syscall.Signal(0)) + if err == nil { + return true + } + // The process exists but we lack permission to signal it — still alive. + return errors.Is(err, syscall.EPERM) +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/liveness_windows.go b/apps/backend/internal/agent/runtime/lifecycle/liveness_windows.go new file mode 100644 index 0000000000..1e01e22add --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/liveness_windows.go @@ -0,0 +1,24 @@ +//go:build windows + +package lifecycle + +import "os" + +// isLocalPIDAlive reports whether a process with the given pid currently exists +// on this host. On Windows os.FindProcess opens the process handle (via +// OpenProcess). Only ever called for local/standalone rows — never for a remote +// SSH pid (see RowProcessLiveness). +func isLocalPIDAlive(pid int) bool { + if pid <= 0 { + return false + } + proc, err := os.FindProcess(pid) + if err != nil { + return false + } + _ = proc.Release() + // Windows does not guarantee OpenProcess fails for a recently exited process + // whose handle is still open elsewhere. This may briefly report alive after + // exit, which is acceptable because Windows is not a server target. + return true +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager.go b/apps/backend/internal/agent/runtime/lifecycle/manager.go index a2816b99eb..26332dcc9e 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager.go @@ -122,6 +122,22 @@ type Manager struct { // returns CodeNpxCacheCorrupted. NewManager wires routingerr.RemediateNpxCache; // tests override it to avoid touching the real filesystem. remediateNpxCache func(path string, log *zap.Logger) error + + // standaloneHostPID is the OS process id of the standalone agentctl + // control-server this backend spawned on the local host. It is the + // host-local liveness handle recorded in executors_running.local_pid for + // local/standalone rows (see persistence.go / #1597 truthful executor rows). + // 0 when unset (tests, or before the launcher wires it). Never used for + // SSH/remote rows — their process lives on another host. + standaloneHostPID atomic.Int64 +} + +// SetStandaloneHostPID records the local agentctl control-server PID so +// local/standalone executor rows can carry a real host-local liveness handle. +// Wired during DI from the agentctl launcher (see backendapp). Safe to leave +// unset in tests that don't exercise the persistence path. +func (m *Manager) SetStandaloneHostPID(pid int) { + m.standaloneHostPID.Store(int64(pid)) } // NewManager creates a new lifecycle manager. diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go index e8cb3b53f8..fabb4a73d0 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go @@ -1192,6 +1192,14 @@ func (m *Manager) MarkCompleted(executionID string, exitCode int, errorMessage s // End session trace span execution.EndSessionSpan() + // Persist the terminal status to executors_running. Unlike the Ready/Running + // transitions (which flow through updateStatusAndPersist), MarkCompleted is + // the process-exit/crash boundary and historically skipped persistence — so a + // row kept claiming a `running`/`starting` process after it had exited + // (#1597). Re-stamping here leaves the row truthful (terminal + // status + fresh last_seen_at) the moment the process is gone. + m.persistExecutorRunning(context.Background(), execution) + m.logger.Info("execution completed", zap.String("execution_id", executionID), zap.Int("exit_code", exitCode), diff --git a/apps/backend/internal/agent/runtime/lifecycle/persistence.go b/apps/backend/internal/agent/runtime/lifecycle/persistence.go index 3359fa3d7e..14437edba9 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/persistence.go +++ b/apps/backend/internal/agent/runtime/lifecycle/persistence.go @@ -9,6 +9,7 @@ import ( "go.uber.org/zap" + "github.com/kandev/kandev/internal/agentruntime" "github.com/kandev/kandev/internal/task/models" v1 "github.com/kandev/kandev/pkg/api/v1" ) @@ -34,6 +35,13 @@ type ExecutorRunningWriter interface { // DeleteExecutorRunningBySessionID removes the row when an execution is // torn down. Idempotent: no-op if no row exists. DeleteExecutorRunningBySessionID(ctx context.Context, sessionID string) error + + // RepairExecutorRunningDead repairs a row in place (status=stopped, local_pid + // cleared, last_seen re-stamped) while preserving resume_token/worktree. Used + // instead of deletion when a stale-cleanup would otherwise destroy a resumable + // row (#1597 resume-safety invariant). Idempotent-friendly: returns + // ErrExecutorRunningNotFound if no row exists. + RepairExecutorRunningDead(ctx context.Context, sessionID string) error } // SetExecutorRunningWriter wires the writer used to persist row state in @@ -128,6 +136,23 @@ func agentctlPortFromExecution(execution *AgentExecution, agentctlURL string) in return port } +// resolveLocalPID returns the host-local liveness handle for an execution's +// executors_running row. For local/standalone runtimes this is the standalone +// agentctl control-server PID Kandev spawned on this host; for every other +// runtime it is 0. SSH/remote processes live on another host (tracked via the +// remote-host PID column) and docker processes live in a container, so this +// never returns a pid for a non-local runtime — a local-process liveness check +// can therefore never run against a remote row (#1597 runtime-aware liveness). +func (m *Manager) resolveLocalPID(execution *AgentExecution) int { + if execution == nil { + return 0 + } + if execution.RuntimeName == agentruntime.RuntimeStandalone { + return int(m.standaloneHostPID.Load()) + } + return 0 +} + func agentctlPIDFromExecution(execution *AgentExecution) int { if execution == nil { return 0 @@ -157,6 +182,20 @@ func metadataInt(metadata map[string]interface{}, key string) int { } } +// isTerminalExecutorRunningStatus reports whether a row status (as produced by +// executorRunningStatusFromExecution) marks the execution as finished — the +// states in which the row must not carry a live local liveness handle. +func isTerminalExecutorRunningStatus(status string) bool { + switch status { + case models.ExecutorRunningStatusFailed, + models.ExecutorRunningStatusStopped, + models.ExecutorRunningStatusComplete: + return true + default: + return false + } +} + func executorRunningStatusFromExecution(execution *AgentExecution) string { if execution == nil { return models.ExecutorRunningStatusStarting @@ -209,19 +248,51 @@ func (m *Manager) persistExecutorRunning(ctx context.Context, execution *AgentEx // owned columns. A prior row exists when: (a) backend is restarting and a // recovered execution is being persisted, or (b) a session is being re-launched // after a fresh-fallback cleanup that did NOT delete the row. + // + // The upsert overwrites resume_token / last_message_uuid with excluded values, + // so those are safe to write only when we know their current values — i.e. when + // the prior read succeeded or positively confirmed no row exists. If the read + // itself FAILS (transient DB error, not "not found"), we must NOT proceed: a + // blind upsert would blank a live resume_token, costing the session its resume + // ability (#1597 resume-safety invariant). Skipping is fail-safe — the row keeps + // its current columns and the next transition (or reconciliation) re-persists. var prior *models.ExecutorRunning if reader, ok := m.runningWriter.(executorRunningReader); ok { - if existing, err := reader.GetExecutorRunningBySessionID(ctx, execution.SessionID); err == nil { + existing, err := reader.GetExecutorRunningBySessionID(ctx, execution.SessionID) + switch { + case err == nil: prior = existing - } else if !errors.Is(err, models.ErrExecutorRunningNotFound) { - m.logger.Warn("failed to read prior executors_running row; orchestrator-owned columns may be cleared", + case errors.Is(err, models.ErrExecutorRunningNotFound): + // No prior row — first insert; nothing to carry forward. + default: + m.logger.Warn("skipping executors_running upsert: prior-row read failed, refusing to risk clobbering resume_token", zap.String("execution_id", execution.ID), zap.String("session_id", execution.SessionID), zap.Error(err)) + return } } running := buildRunningFromExecution(execution, prior) + // Attach the host-local liveness handle for local/standalone rows. Kept out + // of buildRunningFromExecution (a pure mapper) because the PID lives on the + // manager, wired from the agentctl launcher at DI. resolveLocalPID returns 0 + // for non-local runtimes, so a local-process check can never target a remote + // (SSH/docker) row. A terminal row carries no handle: for standalone the + // resolved PID is the shared agentctl control server, which outlives the + // session, and a completed/failed/stopped row must not claim a live process + // (#1597 truthful executor rows) — matching RepairExecutorRunningDead. + if !isTerminalExecutorRunningStatus(running.Status) { + running.LocalPID = m.resolveLocalPID(execution) + } + // last_seen_at reflects an actual liveness observation: every hooked + // transition re-stamps it. buildRunningFromExecution already stamps when a + // live endpoint is known; also stamp when we only have a local handle so a + // local row is never left with a NULL last_seen_at once populated. + if running.LastSeenAt == nil && running.LocalPID > 0 { + now := time.Now().UTC() + running.LastSeenAt = &now + } if err := m.runningWriter.UpsertExecutorRunning(ctx, running); err != nil { m.logger.Error("failed to persist executors_running row in lockstep with store", zap.String("execution_id", execution.ID), @@ -230,16 +301,65 @@ func (m *Manager) persistExecutorRunning(ctx context.Context, execution *AgentEx } } -// deleteExecutorRunning removes the persistence row when an execution is torn -// down (CleanupStaleExecutionBySessionID). Called after executionStore.Remove so -// the in-memory and persistent state are gone in the same operation. +// deleteExecutorRunning tears down the persistence row when an execution is +// cleaned up (CleanupStaleExecutionBySessionID). Called after executionStore.Remove +// so the in-memory and persistent state are gone in the same operation. // -// Best-effort: a failure here is logged but doesn't propagate. A subsequent -// Launch for the same session will UPSERT and overwrite anyway. +// Resume-safety invariant (#1597 resume-safety invariant): a row that still holds +// a resume_token is REPAIRED in place (status=stopped, local_pid cleared) rather +// than deleted, so a session stays resumable even if a subsequent relaunch fails. +// On the happy path the relaunch UPSERTs a fresh row over the repaired one, so +// repairing costs nothing; on the failure path it preserves the only handle to a +// resumable conversation. Rows with no resume_token are deleted as before. +// +// Deliberate deviation from models.RowMustBePreserved, which also preserves +// tokenless rows backing a non-terminal session: this path gates on the token +// alone because the token IS the resumable agent state. A row without one means +// the agent never established (or never reported) an ACP session — there is no +// agent-side context a preserved row could resume, and Kandev's own chat +// history lives in the task tables, untouched by this delete. Preserving a +// tokenless row here would keep only incidental metadata that the relaunch +// upsert rebuilds anyway, at the cost of wiring session-state reads into the +// lifecycle tier. Orchestrator-side reconciliation, which already knows session +// state, applies the full invariant via pruneOrRepairExecutorRow. +// +// Best-effort: a failure here is logged but doesn't propagate. func (m *Manager) deleteExecutorRunning(ctx context.Context, sessionID string) { if m.runningWriter == nil { return } + + // Inspect the row first so we never delete one we couldn't read (fail-safe: + // an unreadable row might hold a resume_token). + if reader, ok := m.runningWriter.(executorRunningReader); ok { + existing, err := reader.GetExecutorRunningBySessionID(ctx, sessionID) + switch { + case err == nil && existing != nil && existing.ResumeToken != "": + if repairErr := m.runningWriter.RepairExecutorRunningDead(ctx, sessionID); repairErr != nil && + !errors.Is(repairErr, models.ErrExecutorRunningNotFound) { + m.logger.Warn("failed to repair resumable executors_running row on cleanup; leaving row intact", + zap.String("session_id", sessionID), + zap.Error(repairErr)) + } else { + m.logger.Info("repaired resumable executors_running row instead of deleting (resume-safety invariant)", + zap.String("session_id", sessionID)) + } + return + case errors.Is(err, models.ErrExecutorRunningNotFound): + m.logger.Debug("delete executors_running on cleanup: row not found", + zap.String("session_id", sessionID)) + return + case err != nil: + m.logger.Warn("skipping executors_running delete: prior-row read failed, refusing to risk deleting a resumable row", + zap.String("session_id", sessionID), + zap.Error(err)) + return + } + } else { + m.logger.Warn("delete executors_running on cleanup: writer does not support reading; resume-safety check skipped", + zap.String("session_id", sessionID)) + } + if err := m.runningWriter.DeleteExecutorRunningBySessionID(ctx, sessionID); err != nil { // "not found" is expected for sessions that were never launched; everything // else is a real I/O failure (write timeout, locked DB) and should surface. diff --git a/apps/backend/internal/agent/runtime/lifecycle/persistence_test.go b/apps/backend/internal/agent/runtime/lifecycle/persistence_test.go index a1ffd1c328..161cd5b97b 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/persistence_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/persistence_test.go @@ -2,6 +2,7 @@ package lifecycle import ( "context" + "errors" "testing" agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl" @@ -23,6 +24,10 @@ func (w *captureExecutorRunningWriter) DeleteExecutorRunningBySessionID(_ contex return nil } +func (w *captureExecutorRunningWriter) RepairExecutorRunningDead(_ context.Context, _ string) error { + return nil +} + func TestBuildRunningFromExecutionPersistsLiveAgentctlEndpoint(t *testing.T) { log := newNopLogger(t) client := agentctl.NewClient("127.0.0.1", 45678, log) @@ -117,6 +122,204 @@ func TestMarkReadyPersistsReadyExecutorRunningStatus(t *testing.T) { } } +// TestLocalStandaloneRowCarriesLocalPID is the characterization test for the +// #1597 symptom "every local row reports pid=0 with no local liveness handle". +// A standalone execution persisted through the manager must carry the host-local +// agentctl PID in executors_running.local_pid, an observed last_seen_at, and a +// live endpoint — so a dead row is distinguishable from a live one. +func TestLocalStandaloneRowCarriesLocalPID(t *testing.T) { + log := newNopLogger(t) + client := agentctl.NewClient("127.0.0.1", 45678, log) + writer := &captureExecutorRunningWriter{} + mgr := newTestManager(t) + mgr.SetExecutorRunningWriter(writer) + mgr.SetStandaloneHostPID(918273) + + if err := mgr.executionStore.Add(&AgentExecution{ + ID: "exec-local", + TaskID: "task-1", + SessionID: "session-1", + RuntimeName: agentruntime.RuntimeStandalone, + Status: v1.AgentStatusRunning, + agentctl: client, + standalonePort: 45678, + }); err != nil { + t.Fatalf("Add execution: %v", err) + } + + if err := mgr.MarkBootReady("exec-local"); err != nil { + t.Fatalf("MarkBootReady: %v", err) + } + + if writer.running == nil { + t.Fatal("expected boot-ready to persist executors_running row") + } + if writer.running.Status != models.ExecutorRunningStatusReady { + t.Fatalf("persisted status = %q, want ready (not stuck starting/prepared)", writer.running.Status) + } + if writer.running.LocalPID != 918273 { + t.Fatalf("persisted local_pid = %d, want the host agentctl pid 918273", writer.running.LocalPID) + } + if writer.running.AgentctlPort != 45678 { + t.Fatalf("persisted agentctl_port = %d, want 45678", writer.running.AgentctlPort) + } + if writer.running.LastSeenAt == nil { + t.Fatal("persisted last_seen_at = nil, want an actual liveness observation") + } +} + +// TestSSHRowNeverCarriesLocalPID guards the #1597 pid-semantics constraint: an SSH row's process +// lives on a remote host, so the local liveness handle must stay 0 for it — the +// remote pid keeps living in the PID column, never local_pid. +func TestSSHRowNeverCarriesLocalPID(t *testing.T) { + log := newNopLogger(t) + client := agentctl.NewClient("127.0.0.1", 43001, log) + writer := &captureExecutorRunningWriter{} + mgr := newTestManager(t) + mgr.SetExecutorRunningWriter(writer) + mgr.SetStandaloneHostPID(918273) // a local pid exists, but must not leak onto an SSH row + + if err := mgr.executionStore.Add(&AgentExecution{ + ID: "exec-ssh", + TaskID: "task-1", + SessionID: "session-1", + RuntimeName: agentruntime.RuntimeSSH, + Status: v1.AgentStatusRunning, + agentctl: client, + Metadata: map[string]interface{}{ + MetadataKeySSHLocalForwardPort: "43001", + MetadataKeySSHRemoteAgentctlPID: "9321", + }, + }); err != nil { + t.Fatalf("Add execution: %v", err) + } + + if err := mgr.MarkReady("exec-ssh"); err != nil { + t.Fatalf("MarkReady: %v", err) + } + + if writer.running == nil { + t.Fatal("expected MarkReady to persist executors_running row") + } + if writer.running.LocalPID != 0 { + t.Fatalf("ssh row local_pid = %d, want 0 (remote process, not local)", writer.running.LocalPID) + } + if writer.running.PID != 9321 { + t.Fatalf("ssh row pid = %d, want remote agentctl pid 9321", writer.running.PID) + } +} + +// TestMarkCompletedPersistsTerminalRow is the characterization test for the +// core population gap: MarkCompleted (the process-exit/crash boundary) used to +// skip persistence, leaving the row claiming a `running` process after it exited. +// The terminal status must land in executors_running. +func TestMarkCompletedPersistsTerminalRow(t *testing.T) { + log := newNopLogger(t) + client := agentctl.NewClient("127.0.0.1", 45678, log) + writer := &captureExecutorRunningWriter{} + mgr := newTestManager(t) + mgr.SetExecutorRunningWriter(writer) + mgr.SetStandaloneHostPID(918273) + + if err := mgr.executionStore.Add(&AgentExecution{ + ID: "exec-done", + TaskID: "task-1", + SessionID: "session-1", + RuntimeName: agentruntime.RuntimeStandalone, + Status: v1.AgentStatusRunning, + agentctl: client, + standalonePort: 45678, + }); err != nil { + t.Fatalf("Add execution: %v", err) + } + + if err := mgr.MarkCompleted("exec-done", 0, ""); err != nil { + t.Fatalf("MarkCompleted: %v", err) + } + + if writer.running == nil { + t.Fatal("expected MarkCompleted to persist executors_running row (it historically did not)") + } + if writer.running.Status != models.ExecutorRunningStatusComplete { + t.Fatalf("persisted status = %q, want completed (row must not keep claiming a running process)", writer.running.Status) + } + if writer.running.LastSeenAt == nil { + t.Fatal("persisted last_seen_at = nil, want a fresh observation at the exit boundary") + } + // A terminal row must not carry a live local liveness handle: the resolved + // standalone PID is the shared agentctl control server, which outlives this + // session, and a completed row claiming a live process would read Alive from + // RowProcessLiveness — the exact lie #1597 truthful executor rows removes. + // Matches RepairExecutorRunningDead, which clears local_pid on repair. + if writer.running.LocalPID != 0 { + t.Fatalf("terminal row local_pid = %d, want 0 (must not claim the live shared agentctl process)", writer.running.LocalPID) + } +} + +// readErrExecutorRunningWriter implements executorRunningReader and returns a +// transient (non-NotFound) error from the prior-row read, recording whether +// UpsertExecutorRunning was subsequently called. +type readErrExecutorRunningWriter struct { + readErr error + upserted bool + lastUpsert *models.ExecutorRunning +} + +func (w *readErrExecutorRunningWriter) GetExecutorRunningBySessionID(_ context.Context, _ string) (*models.ExecutorRunning, error) { + return nil, w.readErr +} + +func (w *readErrExecutorRunningWriter) UpsertExecutorRunning(_ context.Context, running *models.ExecutorRunning) error { + w.upserted = true + w.lastUpsert = running + return nil +} + +func (w *readErrExecutorRunningWriter) DeleteExecutorRunningBySessionID(_ context.Context, _ string) error { + return nil +} + +func (w *readErrExecutorRunningWriter) RepairExecutorRunningDead(_ context.Context, _ string) error { + return nil +} + +// TestPersistSkipsUpsertWhenPriorReadFails guards #1597 resume-safety invariant: +// a transient prior-row read failure must NOT proceed to a blind upsert, because +// that would blank a live resume_token (the upsert overwrites it with the empty +// excluded value). MarkCompleted newly persists at the exit boundary, where a +// completed session may still hold a resume_token — so the fail-safe matters most +// there. A "not found" read, by contrast, is a real first-insert and must proceed. +func TestPersistSkipsUpsertWhenPriorReadFails(t *testing.T) { + mgr := newTestManager(t) + + transient := &readErrExecutorRunningWriter{readErr: errors.New("database is locked")} + mgr.SetExecutorRunningWriter(transient) + mgr.persistExecutorRunning(context.Background(), &AgentExecution{ + ID: "exec-1", + TaskID: "task-1", + SessionID: "session-1", + RuntimeName: agentruntime.RuntimeStandalone, + Status: v1.AgentStatusCompleted, + }) + if transient.upserted { + t.Fatal("prior-read failure must NOT clobber the row with a blind upsert") + } + + // A positive "not found" is a first insert and must still write. + notFound := &readErrExecutorRunningWriter{readErr: models.ErrExecutorRunningNotFound} + mgr.SetExecutorRunningWriter(notFound) + mgr.persistExecutorRunning(context.Background(), &AgentExecution{ + ID: "exec-2", + TaskID: "task-1", + SessionID: "session-2", + RuntimeName: agentruntime.RuntimeStandalone, + Status: v1.AgentStatusReady, + }) + if !notFound.upserted { + t.Fatal("a not-found prior read is a first insert and must proceed with the upsert") + } +} + func TestUpdateStatusPersistsRunningExecutorRunningStatus(t *testing.T) { log := newNopLogger(t) client := agentctl.NewClient("127.0.0.1", 45678, log) diff --git a/apps/backend/internal/agent/runtime/lifecycle/resume_safety_test.go b/apps/backend/internal/agent/runtime/lifecycle/resume_safety_test.go new file mode 100644 index 0000000000..3acdd9fa41 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/resume_safety_test.go @@ -0,0 +1,101 @@ +package lifecycle + +import ( + "context" + "testing" + + "github.com/kandev/kandev/internal/agentruntime" + "github.com/kandev/kandev/internal/task/models" +) + +// invariantWriter is a fake ExecutorRunningWriter + executorRunningReader that +// records whether a cleanup DELETED vs REPAIRED the row, so a test can pin the +// resume-safety deletion invariant (#1597 resume-safety invariant). +type invariantWriter struct { + prior *models.ExecutorRunning + deleted bool + repaired bool +} + +func (w *invariantWriter) UpsertExecutorRunning(_ context.Context, running *models.ExecutorRunning) error { + w.prior = running + return nil +} + +func (w *invariantWriter) GetExecutorRunningBySessionID(_ context.Context, _ string) (*models.ExecutorRunning, error) { + if w.prior == nil { + return nil, models.ErrExecutorRunningNotFound + } + return w.prior, nil +} + +func (w *invariantWriter) DeleteExecutorRunningBySessionID(_ context.Context, _ string) error { + w.deleted = true + w.prior = nil + return nil +} + +func (w *invariantWriter) RepairExecutorRunningDead(_ context.Context, _ string) error { + w.repaired = true + if w.prior != nil { + w.prior.Status = models.ExecutorRunningStatusStopped + w.prior.LocalPID = 0 + } + return nil +} + +// TestDeleteExecutorRunningRepairsResumableRow is the red-before-green +// characterization test for the on-demand stale-cleanup path +// (CleanupStaleExecutionBySessionID → deleteExecutorRunning). A row that still +// holds a resume_token must be repaired in place, never deleted — otherwise a +// stale-cleanup (or a resume whose relaunch fails) permanently costs the operator +// a resumable conversation (#1597 resume-safety invariant). +func TestDeleteExecutorRunningRepairsResumableRow(t *testing.T) { + log := newNopLogger(t) + writer := &invariantWriter{prior: &models.ExecutorRunning{ + SessionID: "session-1", + Runtime: agentruntime.RuntimeStandalone, + Status: models.ExecutorRunningStatusRunning, + ResumeToken: "resume-abc", + LocalPID: 4242, + }} + m := &Manager{logger: log, runningWriter: writer} + + m.deleteExecutorRunning(context.Background(), "session-1") + + if writer.deleted { + t.Fatal("deleteExecutorRunning deleted a row holding a resume_token; the resume-safety invariant requires repair-in-place") + } + if !writer.repaired { + t.Fatal("deleteExecutorRunning did not repair the resumable row") + } + if writer.prior == nil || writer.prior.ResumeToken != "resume-abc" { + t.Fatal("repair must preserve the resume_token") + } + if writer.prior.Status != models.ExecutorRunningStatusStopped || writer.prior.LocalPID != 0 { + t.Fatalf("repair must mark the row stopped and clear the local liveness handle; got status=%q local_pid=%d", + writer.prior.Status, writer.prior.LocalPID) + } +} + +// TestDeleteExecutorRunningPrunesNonResumableRow confirms the invariant does not +// over-preserve: a row with no resume_token is still deleted, so genuinely dead +// rows don't accumulate. +func TestDeleteExecutorRunningPrunesNonResumableRow(t *testing.T) { + log := newNopLogger(t) + writer := &invariantWriter{prior: &models.ExecutorRunning{ + SessionID: "session-2", + Runtime: agentruntime.RuntimeStandalone, + Status: models.ExecutorRunningStatusStopped, + }} + m := &Manager{logger: log, runningWriter: writer} + + m.deleteExecutorRunning(context.Background(), "session-2") + + if writer.repaired { + t.Fatal("a row with no resume_token should be deleted, not repaired") + } + if !writer.deleted { + t.Fatal("a row with no resume_token should be deleted") + } +} diff --git a/apps/backend/internal/backendapp/adapters.go b/apps/backend/internal/backendapp/adapters.go index 5ef3c0527c..a44e7b4bf6 100644 --- a/apps/backend/internal/backendapp/adapters.go +++ b/apps/backend/internal/backendapp/adapters.go @@ -276,6 +276,16 @@ func (a *lifecycleAdapter) StopAgentWithReason(ctx context.Context, agentInstanc return a.mgr.StopAgentWithReason(ctx, agentInstanceID, reason, force) } +// RowLiveness classifies the liveness of the OS process backing an +// executors_running row using the runtime-aware host-local probe. It is the +// orchestrator's window into the platform-split liveness check (kept in the +// lifecycle package) used by startup reconciliation +// (#1597 runtime-aware liveness). A local check never runs against a +// remote/SSH row — such rows return Unknown. +func (a *lifecycleAdapter) RowLiveness(row *models.ExecutorRunning) models.ProcessLiveness { + return lifecycle.RowProcessLiveness(row) +} + // GetAgentStatus returns the status of an agent execution func (a *lifecycleAdapter) GetAgentStatus(ctx context.Context, agentInstanceID string) (*v1.AgentExecution, error) { execution, found := a.mgr.GetExecution(agentInstanceID) diff --git a/apps/backend/internal/backendapp/agentctl.go b/apps/backend/internal/backendapp/agentctl.go index 7a3860fb00..2ae83c5065 100644 --- a/apps/backend/internal/backendapp/agentctl.go +++ b/apps/backend/internal/backendapp/agentctl.go @@ -39,6 +39,9 @@ func provideAgentctlLauncher(ctx context.Context, cfg *config.Config, log *logge } // Store the per-launch auth token so downstream clients can authenticate cfg.Agent.StandaloneAuthToken = l.AuthToken() + // Store the agentctl control-server PID so local/standalone executor rows can + // carry a real host-local liveness handle (executors_running.local_pid). + cfg.Agent.StandalonePID = l.Pid() return &agentctlLauncherResult{ cleanup: cleanup, binaryPath: l.BinaryPath(), diff --git a/apps/backend/internal/backendapp/agents.go b/apps/backend/internal/backendapp/agents.go index d02427385c..2e640c2db9 100644 --- a/apps/backend/internal/backendapp/agents.go +++ b/apps/backend/internal/backendapp/agents.go @@ -119,6 +119,10 @@ func provideLifecycleManager( preparerRegistry.Register(models.ExecutorTypeSSH, lifecycle.NewSSHPreparer(log)) lifecycleMgr.SetPreparerRegistry(preparerRegistry) lifecycleMgr.SetSecretStore(secretStore) + // Record the standalone agentctl control-server PID (populated by + // provideAgentctlLauncher, which runs before this) so local/standalone + // executor rows carry a real host-local liveness handle. + lifecycleMgr.SetStandaloneHostPID(cfg.Agent.StandalonePID) // Wire the agent_profiles reader so the launch-prep skill deploy hook // (ADR 0005 Wave A) can resolve full profile rows including the office // enrichment fields. Without a wired SkillDeployer this is a no-op, diff --git a/apps/backend/internal/common/config/config.go b/apps/backend/internal/common/config/config.go index 9028d3dce7..75f17c43c1 100644 --- a/apps/backend/internal/common/config/config.go +++ b/apps/backend/internal/common/config/config.go @@ -240,6 +240,12 @@ type AgentConfig struct { // StandaloneAuthToken is the per-launch auth token retrieved via handshake. // Set at runtime after agentctl starts; not persisted in config files. StandaloneAuthToken string `mapstructure:"-"` + + // StandalonePID is the OS process id of the standalone agentctl control-server + // this backend spawned. Set at runtime after agentctl starts (from the + // launcher); not persisted in config files. Used as the host-local liveness + // handle recorded in executors_running.local_pid for local/standalone rows. + StandalonePID int `mapstructure:"-"` } // ReadTimeoutDuration returns the read timeout as a time.Duration. diff --git a/apps/backend/internal/orchestrator/event_handlers_test.go b/apps/backend/internal/orchestrator/event_handlers_test.go index 7a5a6aa426..54566a6be8 100644 --- a/apps/backend/internal/orchestrator/event_handlers_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_test.go @@ -153,8 +153,13 @@ type mockAgentManager struct { // isAgentRunningFn, when non-nil, overrides isAgentRunning for // IsAgentRunningForSession. Lets tests model state changes mid-sequence // (e.g. stream disconnect between PromptAgent call and queue write). - isAgentRunningFn func(context.Context, string) bool - isAgentReadyFn func(context.Context, string) bool + isAgentRunningFn func(context.Context, string) bool + isAgentReadyFn func(context.Context, string) bool + // rowLivenessFn, when non-nil, makes the mock satisfy the orchestrator's + // optional rowLivenessProber so reconciliation tests can drive runtime-aware + // liveness per row. Nil → the mock is not a prober and reconciliation treats + // every row as Unknown. + rowLivenessFn func(*models.ExecutorRunning) models.ProcessLiveness resolveProfileErr error restartProcessCalls []string // tracks execution IDs passed to RestartAgentProcess restartProcessErr error @@ -324,6 +329,16 @@ func (m *mockAgentManager) IsAgentReadyForPrompt(ctx context.Context, sessionID } return m.IsAgentRunningForSession(ctx, sessionID) } + +// RowLiveness makes the mock satisfy the orchestrator's optional +// rowLivenessProber. It delegates to rowLivenessFn when set, else reports Unknown +// so tests that don't care about liveness see the safe default. +func (m *mockAgentManager) RowLiveness(row *models.ExecutorRunning) models.ProcessLiveness { + if m.rowLivenessFn != nil { + return m.rowLivenessFn(row) + } + return models.ProcessLivenessUnknown +} func (m *mockAgentManager) ResolveAgentProfile(_ context.Context, _ string) (*executor.AgentProfileInfo, error) { if m.resolveProfileErr != nil { return nil, m.resolveProfileErr diff --git a/apps/backend/internal/orchestrator/executor/executor_resume_test.go b/apps/backend/internal/orchestrator/executor/executor_resume_test.go index 9c89a5539d..f54c6bcf10 100644 --- a/apps/backend/internal/orchestrator/executor/executor_resume_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_resume_test.go @@ -113,20 +113,27 @@ func TestResumeSession_LiveAgentReturnsAlreadyRunning(t *testing.T) { } } -// TestResumeSession_StaleExecutionCleansUpAndRetries ensures the pre-existing -// stale-cleanup path still works when LaunchAgent reports "already has an agent -// running" but the lifecycle manager confirms no live agent exists. +// TestResumeSession_StaleExecutionCleansUpAndRetries is the "row looks live but +// the process is gone" half of the corrected pause→resume contract +// (#1597 pause→resume recovery): the session sits at WAITING_FOR_INPUT with a +// resumable executors_running row, but its agent process is dead. LaunchAgent +// reports "already has an agent running" (stale in-memory execution), the +// runtime-aware liveness probe confirms no live agent, so resume cleans the +// stale execution and relaunches — using the row's resume_token rather than +// wedging on ErrExecutionAlreadyRunning against a process that no longer exists. func TestResumeSession_StaleExecutionCleansUpAndRetries(t *testing.T) { repo := newMockRepository() setupLiveResumeTestFixture(repo) var launchCalls int + var retryResumeToken string agentMgr := &mockAgentManager{ launchAgentFunc: func(_ context.Context, req *LaunchAgentRequest) (*LaunchAgentResponse, error) { launchCalls++ if launchCalls == 1 { return nil, fmt.Errorf("%w: session %q (execution: %s)", lifecycle.ErrAgentAlreadyRunning, req.SessionID, "exec-stale") } + retryResumeToken = req.ACPSessionID return &LaunchAgentResponse{ AgentExecutionID: "exec-new", Status: v1.AgentStatusStarting, @@ -154,6 +161,13 @@ func TestResumeSession_StaleExecutionCleansUpAndRetries(t *testing.T) { if agentMgr.launchAgentCallCount != 2 { t.Errorf("expected LaunchAgent called twice, got %d", agentMgr.launchAgentCallCount) } + // The relaunch must resume the same conversation: the retry carries the + // executors_running row's resume_token (setupLiveResumeTestFixture seeds + // "token-abc"), so the operator's context is preserved rather than starting + // a fresh session. + if retryResumeToken != "token-abc" { + t.Errorf("expected relaunch to reuse resume_token %q, got %q", "token-abc", retryResumeToken) + } } // TestResumeSession_FailedStateForceCleansUpStaleState covers the FAILED-task diff --git a/apps/backend/internal/orchestrator/reconcile_liveness.go b/apps/backend/internal/orchestrator/reconcile_liveness.go new file mode 100644 index 0000000000..aca749b350 --- /dev/null +++ b/apps/backend/internal/orchestrator/reconcile_liveness.go @@ -0,0 +1,64 @@ +package orchestrator + +import ( + "context" + "errors" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/task/models" +) + +// rowLivenessProber is the optional capability the orchestrator uses to classify +// an executors_running row's backing-process liveness in a runtime-aware way. It +// is satisfied by the lifecycle adapter (backendapp.lifecycleAdapter). +// +// Kept as a narrow optional interface — type-asserted off s.agentManager like +// executor.ExecutorTypeCapabilities — so the large AgentManagerClient interface +// and its many test mocks don't have to grow a method just for reconciliation. +type rowLivenessProber interface { + RowLiveness(row *models.ExecutorRunning) models.ProcessLiveness +} + +// rowLiveness classifies a row's backing-process liveness, returning Unknown when +// no prober is wired (unit tests, degraded startup) so a caller never mistakes +// "can't probe" for "dead". The probe is runtime-aware: a local process check +// never runs against a remote/SSH row (#1597 runtime-aware liveness). +func (s *Service) rowLiveness(row *models.ExecutorRunning) models.ProcessLiveness { + prober, ok := s.agentManager.(rowLivenessProber) + if !ok || prober == nil { + return models.ProcessLivenessUnknown + } + return prober.RowLiveness(row) +} + +// pruneOrRepairExecutorRow enforces the resume-safety deletion invariant +// (#1597 resume-safety invariant) at a reconciliation cleanup site: a row backing +// a resumable session, or holding a resume_token, is repaired in place (never +// deleted); only a row that is neither is pruned. +func (s *Service) pruneOrRepairExecutorRow(ctx context.Context, running *models.ExecutorRunning, sessionState models.TaskSessionState) { + sessionID := running.SessionID + if models.RowMustBePreserved(running, sessionState) { + s.repairDeadRowLiveness(ctx, running) + return + } + if err := s.repo.DeleteExecutorRunningBySessionID(ctx, sessionID); err != nil && + !errors.Is(err, models.ErrExecutorRunningNotFound) { + s.logger.Warn("failed to prune terminal executor row during reconciliation", + zap.String("session_id", sessionID), + zap.Error(err)) + } +} + +// repairDeadRowLiveness repairs a preserved row so it no longer claims a live +// process — status=stopped, local_pid cleared, resume_token/worktree preserved — +// satisfying #1597's "never leave a row claiming a dead process" expected +// behavior. Best-effort; a missing row is not an error. +func (s *Service) repairDeadRowLiveness(ctx context.Context, running *models.ExecutorRunning) { + if err := s.repo.RepairExecutorRunningDead(ctx, running.SessionID); err != nil && + !errors.Is(err, models.ErrExecutorRunningNotFound) { + s.logger.Warn("failed to repair executor row liveness during reconciliation", + zap.String("session_id", running.SessionID), + zap.Error(err)) + } +} diff --git a/apps/backend/internal/orchestrator/reconcile_restart_test.go b/apps/backend/internal/orchestrator/reconcile_restart_test.go new file mode 100644 index 0000000000..2bfe7911c1 --- /dev/null +++ b/apps/backend/internal/orchestrator/reconcile_restart_test.go @@ -0,0 +1,229 @@ +package orchestrator + +import ( + "context" + "testing" + "time" + + "github.com/kandev/kandev/internal/agentruntime" + "github.com/kandev/kandev/internal/task/models" +) + +// TestReconcileSessionsOnStartupMakesRowsTrue is the restart-reconciliation +// integration test for #1597 startup reconciliation (the backlog stops +// growing, rows are made true, resumability is never lost). After a restart, +// reconciliation verifies each executors_running +// row against reality using runtime-aware liveness and: +// - prunes a terminal row whose process is dead and that holds no resume_token +// (the stale-row backlog trends toward zero); +// - repairs a resumable row whose process is dead in place — status=stopped, +// local_pid cleared, resume_token preserved — so it never keeps claiming a +// live process, yet stays resumable; +// - leaves a row whose process is still alive with its live handle intact; +// - never applies the local liveness check to an SSH (remote) row. +func TestReconcileSessionsOnStartupMakesRowsTrue(t *testing.T) { + repo := setupTestRepo(t) + ctx := context.Background() + now := time.Now().UTC() + + seedTaskAndSession(t, repo, "taskA", "sA", models.TaskSessionStateCompleted) // terminal, no token, dead → prune + seedTaskAndSession(t, repo, "taskB", "sB", models.TaskSessionStateWaitingForInput) // resumable + token, dead → repair + seedTaskAndSession(t, repo, "taskC", "sC", models.TaskSessionStateRunning) // alive → preserve live handle + seedTaskAndSession(t, repo, "taskD", "sD", models.TaskSessionStateWaitingForInput) // SSH remote → Unknown, untouched + + upsert := func(er *models.ExecutorRunning) { + er.CreatedAt, er.UpdatedAt = now, now + if err := repo.UpsertExecutorRunning(ctx, er); err != nil { + t.Fatalf("upsert %s: %v", er.SessionID, err) + } + } + upsert(&models.ExecutorRunning{ID: "sA", SessionID: "sA", TaskID: "taskA", Runtime: agentruntime.RuntimeStandalone, Status: models.ExecutorRunningStatusRunning, LocalPID: 111}) + upsert(&models.ExecutorRunning{ID: "sB", SessionID: "sB", TaskID: "taskB", Runtime: agentruntime.RuntimeStandalone, Status: models.ExecutorRunningStatusRunning, ResumeToken: "tokB", Resumable: true, LocalPID: 222}) + upsert(&models.ExecutorRunning{ID: "sC", SessionID: "sC", TaskID: "taskC", Runtime: agentruntime.RuntimeStandalone, Status: models.ExecutorRunningStatusRunning, LocalPID: 333}) + upsert(&models.ExecutorRunning{ID: "sD", SessionID: "sD", TaskID: "taskD", Runtime: agentruntime.RuntimeSSH, Status: models.ExecutorRunningStatusRunning, ResumeToken: "tokD", PID: 444}) + + agentMgr := &mockAgentManager{ + rowLivenessFn: func(r *models.ExecutorRunning) models.ProcessLiveness { + switch r.SessionID { + case "sA", "sB": + return models.ProcessLivenessDead + case "sC": + return models.ProcessLivenessAlive + default: // sD (SSH) — a local check must never judge a remote row + return models.ProcessLivenessUnknown + } + }, + } + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + svc.reconcileSessionsOnStartup(ctx) + + // sA: terminal + no resume_token + dead → pruned. + if _, err := repo.GetExecutorRunningBySessionID(ctx, "sA"); err == nil { + t.Error("sA: terminal dead row with no resume_token should be pruned") + } + + // sB: resumable + token + dead → preserved AND repaired. + b, err := repo.GetExecutorRunningBySessionID(ctx, "sB") + if err != nil { + t.Fatalf("sB should be preserved: %v", err) + } + if b.ResumeToken != "tokB" { + t.Errorf("sB resume_token lost during repair: %q", b.ResumeToken) + } + if b.Status != models.ExecutorRunningStatusStopped || b.LocalPID != 0 { + t.Errorf("sB should be repaired to stopped with cleared local_pid; got status=%q local_pid=%d", b.Status, b.LocalPID) + } + + // sC: alive → live local handle preserved (not repaired away). + c, err := repo.GetExecutorRunningBySessionID(ctx, "sC") + if err != nil { + t.Fatalf("sC should be preserved: %v", err) + } + if c.LocalPID != 333 { + t.Errorf("sC live local handle must be preserved; got local_pid=%d", c.LocalPID) + } + + // sD: SSH → local liveness Unknown; the local reconcile must not touch it. + d, err := repo.GetExecutorRunningBySessionID(ctx, "sD") + if err != nil { + t.Fatalf("sD should be preserved: %v", err) + } + if d.PID != 444 || d.ResumeToken != "tokD" { + t.Errorf("sD SSH row must be untouched by the local reconcile; got pid=%d token=%q", d.PID, d.ResumeToken) + } +} + +// TestReconcileSessionsOnStartup_IdleSessionDeadRowRepaired covers the office +// IDLE path: an office turn writes IDLE and tears down, so a crash/restart in +// that window leaves a row claiming status=running with a dead local_pid. +// Startup reconciliation must repair the row (stopped, local_pid cleared, +// resume_token preserved) WITHOUT flipping the session out of IDLE — the IDLE +// state is the office "between turns" shape and must be preserved (#1597). +func TestReconcileSessionsOnStartup_IdleSessionDeadRowRepaired(t *testing.T) { + repo := setupTestRepo(t) + ctx := context.Background() + now := time.Now().UTC() + + seedTaskAndSession(t, repo, "taskI", "sI", models.TaskSessionStateIdle) + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "sI", SessionID: "sI", TaskID: "taskI", Runtime: agentruntime.RuntimeStandalone, + Status: models.ExecutorRunningStatusRunning, ResumeToken: "tokI", Resumable: true, + LocalPID: 4343, CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("upsert sI: %v", err) + } + + agentMgr := &mockAgentManager{ + rowLivenessFn: func(*models.ExecutorRunning) models.ProcessLiveness { + return models.ProcessLivenessDead + }, + } + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + + svc.reconcileSessionsOnStartup(ctx) + + // Session state stays IDLE (not flipped to WAITING_FOR_INPUT). + session, err := repo.GetTaskSession(ctx, "sI") + if err != nil { + t.Fatalf("GetTaskSession(sI): %v", err) + } + if session.State != models.TaskSessionStateIdle { + t.Errorf("IDLE session state must be preserved; got %q", session.State) + } + // But the row is repaired so it no longer claims a live process. + row, err := repo.GetExecutorRunningBySessionID(ctx, "sI") + if err != nil { + t.Fatalf("idle resumable row must be preserved: %v", err) + } + if row.Status != models.ExecutorRunningStatusStopped || row.LocalPID != 0 { + t.Errorf("dead idle row must be repaired to stopped with cleared local_pid; got status=%q local_pid=%d", row.Status, row.LocalPID) + } + if row.ResumeToken != "tokI" { + t.Errorf("resume_token must survive the idle repair; got %q", row.ResumeToken) + } +} + +// TestReconcileSessionsOnStartup_MissingSessionStopsAgentAndDeletesRow covers +// the orphan-row branch of startup reconciliation: a row whose task_session is +// gone entirely (deleted task/worktree) routes to handleMissingSessionOnStartup, +// which stops the still-registered runtime handle (forced) and, once that +// succeeds, deletes the now-meaningless executors_running row — so orphan rows +// don't survive restarts and inflate the backlog (#1597). +func TestReconcileSessionsOnStartup_MissingSessionStopsAgentAndDeletesRow(t *testing.T) { + repo := setupTestRepo(t) + ctx := context.Background() + now := time.Now().UTC() + + // No backing task_session row for "sO" — GetTaskSession returns not-found. + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "sO", + SessionID: "sO", + TaskID: "taskO", + Runtime: agentruntime.RuntimeStandalone, + Status: models.ExecutorRunningStatusRunning, + AgentExecutionID: "execO", + LocalPID: 4242, + CreatedAt: now, + UpdatedAt: now, + }); err != nil { + t.Fatalf("upsert sO: %v", err) + } + + agentMgr := &mockAgentManager{} + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + + svc.reconcileSessionsOnStartup(ctx) + + if len(agentMgr.stopAgentWithReasonArgs) != 1 || agentMgr.stopAgentWithReasonArgs[0].ExecutionID != "execO" { + t.Fatalf("orphan row must stop its runtime handle via StopAgentWithReason; got %+v", agentMgr.stopAgentWithReasonArgs) + } + if !agentMgr.stopAgentWithReasonArgs[0].Force { + t.Errorf("missing-session stop must be forced") + } + if _, err := repo.GetExecutorRunningBySessionID(ctx, "sO"); err == nil { + t.Error("orphan row must be deleted once the runtime stop succeeds") + } +} + +// TestReconcileSessionsOnStartup_CreatedSessionRowPrunedUnlessResumable covers +// the never-started-session cleanup site, which routes through the resume-safety +// invariant rather than deleting unconditionally: a Created session's row with +// no resume_token is pruned (nothing to lose), while the rare Created row that +// already carries a resume_token is repaired in place — the token is the only +// handle to the agent-side conversation and must survive +// (#1597 resume-safety invariant). +func TestReconcileSessionsOnStartup_CreatedSessionRowPrunedUnlessResumable(t *testing.T) { + repo := setupTestRepo(t) + ctx := context.Background() + now := time.Now().UTC() + + seedTaskAndSession(t, repo, "taskE", "sE", models.TaskSessionStateCreated) // no token → prune + seedTaskAndSession(t, repo, "taskF", "sF", models.TaskSessionStateCreated) // token → repair + + upsert := func(er *models.ExecutorRunning) { + er.CreatedAt, er.UpdatedAt = now, now + if err := repo.UpsertExecutorRunning(ctx, er); err != nil { + t.Fatalf("upsert %s: %v", er.SessionID, err) + } + } + upsert(&models.ExecutorRunning{ID: "sE", SessionID: "sE", TaskID: "taskE", Runtime: agentruntime.RuntimeStandalone, Status: models.ExecutorRunningStatusStarting, LocalPID: 555}) + upsert(&models.ExecutorRunning{ID: "sF", SessionID: "sF", TaskID: "taskF", Runtime: agentruntime.RuntimeStandalone, Status: models.ExecutorRunningStatusStarting, ResumeToken: "tokF", Resumable: true, LocalPID: 666}) + + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{}) + svc.reconcileSessionsOnStartup(ctx) + + if _, err := repo.GetExecutorRunningBySessionID(ctx, "sE"); err == nil { + t.Error("sE: never-started row with no resume_token should be pruned") + } + + f, err := repo.GetExecutorRunningBySessionID(ctx, "sF") + if err != nil { + t.Fatalf("sF: Created row holding a resume_token must be preserved: %v", err) + } + if f.ResumeToken != "tokF" { + t.Errorf("sF resume_token lost: %q", f.ResumeToken) + } + if f.Status != models.ExecutorRunningStatusStopped || f.LocalPID != 0 { + t.Errorf("sF should be repaired to stopped with cleared local_pid; got status=%q local_pid=%d", f.Status, f.LocalPID) + } +} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 0f77fc99dd..2ce36cf010 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -170,6 +170,11 @@ type sessionExecutorStore interface { HasExecutorRunningRow(ctx context.Context, sessionID string) (bool, error) UpdateResumeToken(ctx context.Context, sessionID, expectedExecID, resumeToken, lastMessageUUID string) error UpdateExecutorRunningStatus(ctx context.Context, sessionID, status string) error + // RepairExecutorRunningDead repairs a row in place (status=stopped, local_pid + // cleared, last_seen re-stamped) while preserving resume_token/worktree — used + // by reconciliation to honor the resume-safety invariant instead of deleting a + // resumable row. + RepairExecutorRunningDead(ctx context.Context, sessionID string) error // Executor GetExecutor(ctx context.Context, id string) (*models.Executor, error) // Task @@ -1067,11 +1072,10 @@ func (s *Service) reconcileOneSessionOnStartup(ctx context.Context, running *mod if previousState == models.TaskSessionStateCreated { s.logger.Info("session was never started; cleaning up", zap.String("session_id", sessionID)) - if err := s.repo.DeleteExecutorRunningBySessionID(ctx, sessionID); err != nil { - s.logger.Warn("failed to remove executor record for unstarted session", - zap.String("session_id", sessionID), - zap.Error(err)) - } + // A never-started session has no conversation to lose, so its row is + // prunable — but route through the invariant so the rare case of a + // Created row that still carries a resume_token is repaired, not deleted. + s.pruneOrRepairExecutorRow(ctx, running, previousState) return } @@ -1083,6 +1087,15 @@ func (s *Service) reconcileOneSessionOnStartup(ctx context.Context, running *mod // the chat UI render as "working" because the office session shape // uses IDLE specifically to avoid that. if previousState == models.TaskSessionStateIdle { + // Keep the IDLE session state (no flip to WAITING_FOR_INPUT), but still + // make the ROW true: an office turn writes IDLE and then tears down, so a + // crash/restart in that window leaves a row claiming status=running with a + // dead local_pid. If the local process is confirmed dead, repair the row + // in place — resume_token/worktree are preserved (RowMustBePreserved + // treats IDLE as resumable). Remote rows report Unknown and are untouched. + if s.rowLiveness(running) == models.ProcessLivenessDead { + s.repairDeadRowLiveness(ctx, running) + } s.logger.Info("session reconciled for lazy recovery (idle, no state change)", zap.String("session_id", sessionID), zap.String("task_id", running.TaskID), @@ -1126,7 +1139,15 @@ func (s *Service) reconcileOneSessionOnStartup(ctx context.Context, running *mod } // PRESERVE the ExecutorRunning record — it holds the resume token and worktree info - // needed for lazy recovery when the user opens the session. + // needed for lazy recovery when the user opens the session. But make the row + // TRUE: after a restart the spawned process is normally gone, so if the local + // liveness handle is confirmed dead, repair the row (status=stopped, local_pid + // cleared) so it never keeps claiming a live process (#1597 expected behavior). + // resume_token/worktree are preserved by the repair. Remote rows report Unknown + // and are left to their own runtime's status poll. + if s.rowLiveness(running) == models.ProcessLivenessDead { + s.repairDeadRowLiveness(ctx, running) + } s.logger.Info("session reconciled for lazy recovery", zap.String("session_id", sessionID), @@ -1199,11 +1220,11 @@ func (s *Service) handleTerminalSessionOnStartup(ctx context.Context, session *m return true } } - if err := s.repo.DeleteExecutorRunningBySessionID(ctx, sessionID); err != nil { - s.logger.Warn("failed to remove executor record for terminal session", - zap.String("session_id", sessionID), - zap.Error(err)) - } + // Resume-safety invariant: prune the row only if it is not still resumable + // (no resume_token). A terminal session that kept a resume_token — e.g. an + // office run that COMPLETED a turn but stays resumable — is repaired in + // place instead of deleted. + s.pruneOrRepairExecutorRow(ctx, running, previousState) return true case models.TaskSessionStateFailed: s.handleFailedSessionOnStartup(ctx, session, running) @@ -1248,11 +1269,9 @@ func (s *Service) handleFailedSessionOnStartup(ctx context.Context, session *mod return } } - if err := s.repo.DeleteExecutorRunningBySessionID(ctx, sessionID); err != nil { - s.logger.Warn("failed to remove executor record for failed session", - zap.String("session_id", sessionID), - zap.Error(err)) - } + // Prune only subject to the resume-safety invariant (a lingering + // resume_token is repaired in place rather than deleted). + s.pruneOrRepairExecutorRow(ctx, running, models.TaskSessionStateFailed) } } diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 9204121e28..f12bed200d 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -2614,6 +2614,20 @@ func (s *Service) CancelAgent(ctx context.Context, sessionID string) error { // concurrent agent.complete event having already closed the turn. s.completeTurnForSession(ctx, sessionID) + // Deliver any message the operator queued while the turn was running + // (#1597 pause→resume recovery). On a normal cancel the agent emits + // complete(cancelled) → handleAgentReady → on_turn_complete, which drains + // the queue. But an escalated cancel (agent hung) or a dead-process cancel + // fires no agent.ready event, so nothing would drain it — leaving the + // operator's queued message stranded until a second manual send. Draining + // here after the turn is completed covers those paths. It is idempotent with + // the event-driven drain: drainQueuedMessageForPromptableSession pops via + // TakeQueued atomically, so a normal cancel that also fires handleAgentReady + // cannot double-deliver. + if session != nil { + s.drainQueuedMessageForPromptableSession(ctx, sessionID) + } + s.logger.Debug("agent turn cancelled", zap.String("session_id", sessionID)) return nil } diff --git a/apps/backend/internal/orchestrator/task_operations_test.go b/apps/backend/internal/orchestrator/task_operations_test.go index 6b09e666af..2c2b14f88b 100644 --- a/apps/backend/internal/orchestrator/task_operations_test.go +++ b/apps/backend/internal/orchestrator/task_operations_test.go @@ -597,10 +597,30 @@ func TestCancelAgent_TaskStateReconcile(t *testing.T) { } } -func TestCancelAgent_LeavesQueuedMessageForManualDrain(t *testing.T) { +// TestCancelAgent_DeliversQueuedMessageOnResume pins the corrected pause→resume +// contract (#1597 pause→resume recovery): a message queued while the agent was +// running is DELIVERED once cancel settles the session to WAITING_FOR_INPUT, not +// stranded on the queue for indefinite manual drain. +// +// This replaces the former TestCancelAgent_LeavesQueuedMessageForManualDrain, +// which codified the wedge as expected. On a normal cancel the agent's +// complete(cancelled) event fires handleAgentReady → on_turn_complete → drain, +// but on an escalated / dead-process cancel no agent.ready event ever fires, so +// nothing drained the queue and the operator's queued message was lost until a +// second manual send. Cancel now drains directly after reconciling the session. +// +// Proof of delivery is the queue emptying: drainQueuedMessageForPromptableSession +// pops the message synchronously (TakeQueued) before dispatching it, so a +// Count==0 immediately after CancelAgent returns is deterministic. The actual +// PromptTask dispatch runs in a goroutine (state may already be RUNNING by the +// time we re-read), mirroring TestHandleAgentBootReady_DrainsOrphanedQueuedMessage. +func TestCancelAgent_DeliversQueuedMessageOnResume(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) - agentMgr := &mockAgentManager{isAgentRunning: true} + // repoForExecutionLookup + isAgentRunning let the executeQueuedMessage + // goroutine's PromptTask → ensureSessionRunning → GetExecutionBySession land + // on a working path instead of nil-derefing s.executor under -race. + agentMgr := &mockAgentManager{isAgentRunning: true, repoForExecutionLookup: repo} svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr) svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{}) @@ -616,23 +636,21 @@ func TestCancelAgent_LeavesQueuedMessageForManualDrain(t *testing.T) { t.Fatalf("cancel agent: %v", err) } + // Cancel settles the session so a new prompt can land. The drain goroutine + // may already have moved it back to RUNNING; either state proves the wedge + // (stuck RUNNING with a full queue) is gone. updated, err := repo.GetTaskSession(ctx, "session1") if err != nil { t.Fatalf("get updated session: %v", err) } - if updated.State != models.TaskSessionStateWaitingForInput { - t.Fatalf("expected session state %q, got %q", models.TaskSessionStateWaitingForInput, updated.State) + if updated.State != models.TaskSessionStateWaitingForInput && + updated.State != models.TaskSessionStateRunning { + t.Fatalf("expected session state WAITING_FOR_INPUT or RUNNING after cancel, got %q", updated.State) } status := svc.messageQueue.GetStatus(ctx, "session1") - if status.Count != 1 { - t.Fatalf("expected cancel to leave queued message for manual drain, count=%d entries=%+v", status.Count, status.Entries) - } - if got := status.Entries[0].Content; got != "queued after cancel" { - t.Fatalf("queued prompt = %q, want %q", got, "queued after cancel") - } - if len(agentMgr.capturedPrompts) != 0 { - t.Fatalf("expected cancel not to prompt queued message, got %d prompts", len(agentMgr.capturedPrompts)) + if status.Count != 0 { + t.Fatalf("expected cancel to deliver the queued message on resume, count=%d entries=%+v", status.Count, status.Entries) } } diff --git a/apps/backend/internal/task/handlers/process_handlers_test.go b/apps/backend/internal/task/handlers/process_handlers_test.go index 043e4d2404..4b5b5476d6 100644 --- a/apps/backend/internal/task/handlers/process_handlers_test.go +++ b/apps/backend/internal/task/handlers/process_handlers_test.go @@ -405,6 +405,9 @@ func (m *mockRepository) UpdateResumeToken(ctx context.Context, sessionID, expec func (m *mockRepository) UpdateExecutorRunningStatus(ctx context.Context, sessionID, status string) error { return nil } +func (m *mockRepository) RepairExecutorRunningDead(ctx context.Context, sessionID string) error { + return nil +} func (m *mockRepository) CreateEnvironment(ctx context.Context, environment *models.Environment) error { return nil } diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index 1b6070343e..e7b44596f8 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -890,28 +890,36 @@ type Executor struct { // ExecutorRunning tracks an active executor instance for a session. type ExecutorRunning struct { - ID string `json:"id"` - SessionID string `json:"session_id"` - TaskID string `json:"task_id"` - ExecutorID string `json:"executor_id"` - Runtime agentruntime.Runtime `json:"runtime,omitempty"` - Status string `json:"status"` - Resumable bool `json:"resumable"` - ResumeToken string `json:"resume_token,omitempty"` - LastMessageUUID string `json:"last_message_uuid,omitempty"` - AgentExecutionID string `json:"agent_execution_id,omitempty"` - ContainerID string `json:"container_id,omitempty"` - AgentctlURL string `json:"agentctl_url,omitempty"` - AgentctlPort int `json:"agentctl_port,omitempty"` - PID int `json:"pid,omitempty"` - WorktreeID string `json:"worktree_id,omitempty"` - WorktreePath string `json:"worktree_path,omitempty"` - WorktreeBranch string `json:"worktree_branch,omitempty"` - ErrorMessage string `json:"error_message,omitempty"` - Metadata map[string]interface{} `json:"metadata,omitempty"` - LastSeenAt *time.Time `json:"last_seen_at,omitempty"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + ID string `json:"id"` + SessionID string `json:"session_id"` + TaskID string `json:"task_id"` + ExecutorID string `json:"executor_id"` + Runtime agentruntime.Runtime `json:"runtime,omitempty"` + Status string `json:"status"` + Resumable bool `json:"resumable"` + ResumeToken string `json:"resume_token,omitempty"` + LastMessageUUID string `json:"last_message_uuid,omitempty"` + AgentExecutionID string `json:"agent_execution_id,omitempty"` + ContainerID string `json:"container_id,omitempty"` + AgentctlURL string `json:"agentctl_url,omitempty"` + AgentctlPort int `json:"agentctl_port,omitempty"` + // PID is SSH-only: the agentctl PID on the *remote* host, used by the SSH + // executor's remote-pid stop path. It is 0 for local/standalone rows. + PID int `json:"pid,omitempty"` + // LocalPID is a host-local liveness handle: the PID of the standalone + // agentctl control-server process Kandev spawned on this host. Populated for + // local/standalone runtimes so a dead row is distinguishable from a live one + // without external context. 0 for SSH/remote rows (their process lives on + // another host — see PID). Never probe LocalPID for SSH rows. + LocalPID int `json:"local_pid,omitempty"` + WorktreeID string `json:"worktree_id,omitempty"` + WorktreePath string `json:"worktree_path,omitempty"` + WorktreeBranch string `json:"worktree_branch,omitempty"` + ErrorMessage string `json:"error_message,omitempty"` + Metadata map[string]interface{} `json:"metadata,omitempty"` + LastSeenAt *time.Time `json:"last_seen_at,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` } // ProfileEnvVar represents an environment variable for an executor profile. diff --git a/apps/backend/internal/task/models/resume_safety.go b/apps/backend/internal/task/models/resume_safety.go new file mode 100644 index 0000000000..734fe8c34f --- /dev/null +++ b/apps/backend/internal/task/models/resume_safety.go @@ -0,0 +1,71 @@ +package models + +// ProcessLiveness classifies whether the OS process backing an executors_running +// row is still alive, judged in a runtime-aware way. It is a low-dependency +// mirror of the runtime/lifecycle package's liveness probe result, so the +// orchestrator and task service can reason about a row's liveness without +// importing runtime/lifecycle (#1597 startup reconciliation, +// #1597 runtime-aware liveness). +type ProcessLiveness int + +const ( + // ProcessLivenessUnknown means the row cannot be judged by a host-local + // process check — a remote (SSH) runtime whose process lives on another host, + // a containerized/docker runtime, or a local row whose liveness handle is not + // yet populated. Callers MUST NOT treat Unknown as dead. + ProcessLivenessUnknown ProcessLiveness = iota + // ProcessLivenessAlive means the row's local process handle refers to a + // process that currently exists on this host. + ProcessLivenessAlive + // ProcessLivenessDead means the row's local process handle refers to a + // process that no longer exists on this host — the row is stale. Pruning it + // is safe only subject to the resume-safety invariant (RowMustBePreserved), + // not from liveness alone. + ProcessLivenessDead +) + +// IsResumableSessionState reports whether a session is in an open, resumable +// state whose executors_running row must be preserved even without a +// resume_token — the agent may still be reachable or the conversation continued +// on the next turn. The resumable states are Starting, Running, WaitingForInput, +// and Idle (office between-turns). +// +// Explicitly NOT resumable: Created (a never-started placeholder with no +// conversation to lose) and the terminal states (Completed/Failed/Cancelled). +// Those may be pruned when the row also carries no resume_token. +func IsResumableSessionState(state TaskSessionState) bool { + switch state { + case TaskSessionStateStarting, TaskSessionStateRunning, + TaskSessionStateWaitingForInput, TaskSessionStateIdle: + return true + default: + return false + } +} + +// RowMustBePreserved is the resume-safety deletion invariant: a row holding a +// resume_token, or backing a session in an open/resumable state, is repaired in +// place and never deleted. Only a row that is neither (a finished or +// never-started session with no resume_token) may be pruned. +// +// This is the single source of truth for "is it safe to delete this +// executors_running row" across every cleanup path (cancel cleanup, startup +// reconciliation, on-demand stale cleanup, task teardown). The guarantee is +// expressed as a deletion invariant rather than by duplicating resume_token into +// a second table, which would reintroduce the divergence risk this effort +// removes (#1597: resume_token must not be duplicated into a second table). +// +// One documented deviation: the lifecycle tier's stale-execution cleanup +// (lifecycle.Manager deleteExecutorRunning) gates on the resume_token alone, +// without the session-state half — a tokenless row has no agent-side session to +// resume, so preserving it there buys nothing. See the rationale on that +// function before "fixing" the divergence. +func RowMustBePreserved(running *ExecutorRunning, sessionState TaskSessionState) bool { + if running == nil { + return false + } + if running.ResumeToken != "" { + return true + } + return IsResumableSessionState(sessionState) +} diff --git a/apps/backend/internal/task/models/resume_safety_test.go b/apps/backend/internal/task/models/resume_safety_test.go new file mode 100644 index 0000000000..30960d71ce --- /dev/null +++ b/apps/backend/internal/task/models/resume_safety_test.go @@ -0,0 +1,56 @@ +package models + +import "testing" + +func TestIsResumableSessionState(t *testing.T) { + resumable := []TaskSessionState{ + TaskSessionStateStarting, TaskSessionStateRunning, + TaskSessionStateWaitingForInput, TaskSessionStateIdle, + } + for _, s := range resumable { + if !IsResumableSessionState(s) { + t.Errorf("IsResumableSessionState(%q) = false, want true", s) + } + } + notResumable := []TaskSessionState{ + TaskSessionStateCreated, TaskSessionStateCompleted, + TaskSessionStateFailed, TaskSessionStateCancelled, + } + for _, s := range notResumable { + if IsResumableSessionState(s) { + t.Errorf("IsResumableSessionState(%q) = true, want false", s) + } + } +} + +func TestRowMustBePreserved(t *testing.T) { + tests := []struct { + name string + running *ExecutorRunning + state TaskSessionState + want bool + }{ + {"nil row is never preserved", nil, TaskSessionStateRunning, false}, + {"resume_token on terminal session is preserved", + &ExecutorRunning{ResumeToken: "tok"}, TaskSessionStateCompleted, true}, + {"running session without token is preserved", + &ExecutorRunning{}, TaskSessionStateRunning, true}, + {"waiting-for-input session without token is preserved", + &ExecutorRunning{}, TaskSessionStateWaitingForInput, true}, + {"idle office session without token is preserved", + &ExecutorRunning{}, TaskSessionStateIdle, true}, + {"never-started created session without token is prunable", + &ExecutorRunning{}, TaskSessionStateCreated, false}, + {"terminal session without token is prunable", + &ExecutorRunning{}, TaskSessionStateCancelled, false}, + {"terminal session with token is still preserved", + &ExecutorRunning{ResumeToken: "tok"}, TaskSessionStateFailed, true}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := RowMustBePreserved(tc.running, tc.state); got != tc.want { + t.Errorf("RowMustBePreserved = %v, want %v", got, tc.want) + } + }) + } +} diff --git a/apps/backend/internal/task/repository/interface.go b/apps/backend/internal/task/repository/interface.go index fe47bc3a69..53f176967d 100644 --- a/apps/backend/internal/task/repository/interface.go +++ b/apps/backend/internal/task/repository/interface.go @@ -251,6 +251,12 @@ type ExecutorRepository interface { // launch) so the row doesn't sit on the misleading default "starting" forever. // Returns models.ErrExecutorRunningNotFound if no row exists for the session. UpdateExecutorRunningStatus(ctx context.Context, sessionID, status string) error + // RepairExecutorRunningDead repairs a row in place to reflect a dead backing + // process (status=stopped, local_pid cleared, last_seen re-stamped) while + // preserving resume_token/worktree/endpoint. Used by cleanup paths to honor + // the resume-safety invariant instead of deleting a resumable row. + // Returns models.ErrExecutorRunningNotFound if no row exists for the session. + RepairExecutorRunningDead(ctx context.Context, sessionID string) error } // EnvironmentRepository handles environment CRUD. diff --git a/apps/backend/internal/task/repository/sqlite/base_migrations.go b/apps/backend/internal/task/repository/sqlite/base_migrations.go index 437fa99481..f8ef51b657 100644 --- a/apps/backend/internal/task/repository/sqlite/base_migrations.go +++ b/apps/backend/internal/task/repository/sqlite/base_migrations.go @@ -46,6 +46,11 @@ func (r *Repository) migrateSessionsAddCostColumns() { func (r *Repository) runMigrations() error { r.migrate.Apply("executors_running.last_message_uuid", `ALTER TABLE executors_running ADD COLUMN last_message_uuid TEXT DEFAULT ''`) r.migrate.Apply("executors_running.metadata", `ALTER TABLE executors_running ADD COLUMN metadata TEXT DEFAULT '{}'`) + // local_pid holds a host-local liveness handle (the standalone agentctl + // control-server PID Kandev spawns) for local/standalone rows. It is kept + // deliberately separate from the SSH-only `pid` column, which holds an + // agentctl PID on the *remote* host. See ADR 0025 / issue #1597. + r.migrate.Apply("executors_running.local_pid", `ALTER TABLE executors_running ADD COLUMN local_pid INTEGER DEFAULT 0`) r.migrate.Apply("tasks.is_ephemeral", `ALTER TABLE tasks ADD COLUMN is_ephemeral INTEGER NOT NULL DEFAULT 0`) r.migrate.Apply("task_repositories.checkout_branch", `ALTER TABLE task_repositories ADD COLUMN checkout_branch TEXT DEFAULT ''`) // Multi-branch support: drop the old UNIQUE(task_id, repository_id) and diff --git a/apps/backend/internal/task/repository/sqlite/base_schema.go b/apps/backend/internal/task/repository/sqlite/base_schema.go index 363d8ec296..41c72ac147 100644 --- a/apps/backend/internal/task/repository/sqlite/base_schema.go +++ b/apps/backend/internal/task/repository/sqlite/base_schema.go @@ -176,6 +176,7 @@ const infraSchemaDDL = ` agentctl_url TEXT DEFAULT '', agentctl_port INTEGER DEFAULT 0, pid INTEGER DEFAULT 0, + local_pid INTEGER DEFAULT 0, worktree_id TEXT DEFAULT '', worktree_path TEXT DEFAULT '', worktree_branch TEXT DEFAULT '', diff --git a/apps/backend/internal/task/repository/sqlite/executor.go b/apps/backend/internal/task/repository/sqlite/executor.go index 755c53d80f..9e0cd0a3b7 100644 --- a/apps/backend/internal/task/repository/sqlite/executor.go +++ b/apps/backend/internal/task/repository/sqlite/executor.go @@ -164,10 +164,10 @@ func (r *Repository) UpsertExecutorRunning(ctx context.Context, running *models. _, err := r.db.ExecContext(ctx, r.db.Rebind(` INSERT INTO executors_running ( id, session_id, task_id, executor_id, runtime, status, resumable, resume_token, - last_message_uuid, agent_execution_id, container_id, agentctl_url, agentctl_port, pid, + last_message_uuid, agent_execution_id, container_id, agentctl_url, agentctl_port, pid, local_pid, worktree_id, worktree_path, worktree_branch, last_seen_at, error_message, metadata, created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(session_id) DO UPDATE SET id = excluded.id, task_id = excluded.task_id, @@ -182,6 +182,7 @@ func (r *Repository) UpsertExecutorRunning(ctx context.Context, running *models. agentctl_url = excluded.agentctl_url, agentctl_port = excluded.agentctl_port, pid = excluded.pid, + local_pid = excluded.local_pid, worktree_id = excluded.worktree_id, worktree_path = excluded.worktree_path, worktree_branch = excluded.worktree_branch, @@ -204,6 +205,7 @@ func (r *Repository) UpsertExecutorRunning(ctx context.Context, running *models. running.AgentctlURL, running.AgentctlPort, running.PID, + running.LocalPID, running.WorktreeID, running.WorktreePath, running.WorktreeBranch, @@ -219,7 +221,7 @@ func (r *Repository) UpsertExecutorRunning(ctx context.Context, running *models. func (r *Repository) ListExecutorsRunning(ctx context.Context) ([]*models.ExecutorRunning, error) { rows, err := r.ro.QueryContext(ctx, ` SELECT id, session_id, task_id, executor_id, runtime, status, resumable, resume_token, - last_message_uuid, agent_execution_id, container_id, agentctl_url, agentctl_port, pid, + last_message_uuid, agent_execution_id, container_id, agentctl_url, agentctl_port, pid, local_pid, worktree_id, worktree_path, worktree_branch, last_seen_at, error_message, metadata, created_at, updated_at FROM executors_running @@ -239,7 +241,7 @@ func (r *Repository) ListExecutorsRunningByTaskID(ctx context.Context, taskID st } rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(` SELECT id, session_id, task_id, executor_id, runtime, status, resumable, resume_token, - last_message_uuid, agent_execution_id, container_id, agentctl_url, agentctl_port, pid, + last_message_uuid, agent_execution_id, container_id, agentctl_url, agentctl_port, pid, local_pid, worktree_id, worktree_path, worktree_branch, last_seen_at, error_message, metadata, created_at, updated_at FROM executors_running @@ -265,7 +267,7 @@ func (r *Repository) GetExecutorRunningBySessionID(ctx context.Context, sessionI err := r.ro.QueryRowContext(ctx, r.ro.Rebind(` SELECT id, session_id, task_id, executor_id, runtime, status, resumable, resume_token, - last_message_uuid, agent_execution_id, container_id, agentctl_url, agentctl_port, pid, + last_message_uuid, agent_execution_id, container_id, agentctl_url, agentctl_port, pid, local_pid, worktree_id, worktree_path, worktree_branch, last_seen_at, error_message, metadata, created_at, updated_at FROM executors_running @@ -285,6 +287,7 @@ func (r *Repository) GetExecutorRunningBySessionID(ctx context.Context, sessionI &running.AgentctlURL, &running.AgentctlPort, &running.PID, + &running.LocalPID, &running.WorktreeID, &running.WorktreePath, &running.WorktreeBranch, @@ -336,6 +339,7 @@ func scanExecutorRunningRows(rows *sql.Rows) ([]*models.ExecutorRunning, error) &running.AgentctlURL, &running.AgentctlPort, &running.PID, + &running.LocalPID, &running.WorktreeID, &running.WorktreePath, &running.WorktreeBranch, @@ -448,6 +452,36 @@ func (r *Repository) UpdateResumeToken(ctx context.Context, sessionID, expectedE return nil } +// RepairExecutorRunningDead repairs a row in place to reflect that its backing +// process is gone, WITHOUT deleting it. The resume-safety invariant requires a +// resumable / non-terminal row be repaired rather than pruned +// (#1597 resume-safety invariant): flip status to "stopped", clear the local +// liveness handle (local_pid = 0) so the row no longer claims a live process, +// and re-stamp last_seen_at as a fresh liveness observation. resume_token, +// worktree, and endpoint columns are intentionally preserved so the session +// stays resumable. +// +// Returns ErrExecutorRunningNotFound when no row exists for the session. +func (r *Repository) RepairExecutorRunningDead(ctx context.Context, sessionID string) error { + if sessionID == "" { + return fmt.Errorf("session_id is required") + } + now := time.Now().UTC() + result, err := r.db.ExecContext(ctx, r.db.Rebind(` + UPDATE executors_running + SET status = ?, local_pid = 0, last_seen_at = ?, updated_at = ? + WHERE session_id = ? + `), models.ExecutorRunningStatusStopped, now, now, sessionID) + if err != nil { + return err + } + rows, _ := result.RowsAffected() + if rows == 0 { + return fmt.Errorf("%w for session: %s", models.ErrExecutorRunningNotFound, sessionID) + } + return nil +} + // UpdateExecutorRunningStatus narrowly updates the status column. // Returns ErrExecutorRunningNotFound when no row exists for the session. func (r *Repository) UpdateExecutorRunningStatus(ctx context.Context, sessionID, status string) error { diff --git a/apps/backend/internal/task/repository/sqlite/executor_localpid_test.go b/apps/backend/internal/task/repository/sqlite/executor_localpid_test.go new file mode 100644 index 0000000000..b3a512c186 --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/executor_localpid_test.go @@ -0,0 +1,275 @@ +package sqlite + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/kandev/kandev/internal/agentruntime" + "github.com/kandev/kandev/internal/task/models" +) + +// TestExecutorRunningLocalPIDRoundTrips proves the local_pid column is an +// independent, fully-plumbed column: it is written by UpsertExecutorRunning and +// read back by GetExecutorRunningBySessionID / ListExecutorsRunning without +// aliasing the SSH-only pid column. This is the schema half of the #1597 +// executor-row-desync fix (#1597 truthful executor rows): a local/standalone row +// must be able to carry a real host-local liveness handle distinct from the +// remote-host SSH pid. +func TestExecutorRunningLocalPIDRoundTrips(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + lastSeen := time.Now().UTC().Truncate(time.Second) + + seedExecutorRunningCleanupTask(t, repo, "task-1") + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "session-local", + TaskID: "task-1", + State: models.TaskSessionStateRunning, + }); err != nil { + t.Fatalf("CreateTaskSession(session-local): %v", err) + } + + // A local/standalone row: local_pid is the host-local agentctl PID, pid is + // 0 (SSH-only, no remote host). last_seen_at reflects a real observation. + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "session-local", + SessionID: "session-local", + TaskID: "task-1", + ExecutorID: "exec-local", + Runtime: agentruntime.RuntimeStandalone, + Status: models.ExecutorRunningStatusReady, + Resumable: true, + AgentctlURL: "http://127.0.0.1:8765", + AgentctlPort: 8765, + PID: 0, + LocalPID: 424242, + LastSeenAt: &lastSeen, + }); err != nil { + t.Fatalf("UpsertExecutorRunning(session-local): %v", err) + } + + got, err := repo.GetExecutorRunningBySessionID(ctx, "session-local") + if err != nil { + t.Fatalf("GetExecutorRunningBySessionID: %v", err) + } + if got.LocalPID != 424242 { + t.Fatalf("local_pid not persisted/read: got %d want 424242", got.LocalPID) + } + if got.PID != 0 { + t.Fatalf("pid must stay independent of local_pid: got %d want 0", got.PID) + } + if got.Status != models.ExecutorRunningStatusReady { + t.Fatalf("status not persisted: got %q want %q", got.Status, models.ExecutorRunningStatusReady) + } + if got.LastSeenAt == nil || !got.LastSeenAt.Equal(lastSeen) { + t.Fatalf("last_seen_at not persisted: got %v want %v", got.LastSeenAt, lastSeen) + } + + // The upsert path (ON CONFLICT) must update local_pid too, not just insert it. + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "session-local", + SessionID: "session-local", + TaskID: "task-1", + ExecutorID: "exec-local", + Runtime: agentruntime.RuntimeStandalone, + Status: models.ExecutorRunningStatusReady, + LocalPID: 515151, + }); err != nil { + t.Fatalf("UpsertExecutorRunning(update local_pid): %v", err) + } + got, err = repo.GetExecutorRunningBySessionID(ctx, "session-local") + if err != nil { + t.Fatalf("GetExecutorRunningBySessionID(after update): %v", err) + } + if got.LocalPID != 515151 { + t.Fatalf("local_pid not updated on conflict: got %d want 515151", got.LocalPID) + } + + // list path must scan the column too. + rows, err := repo.ListExecutorsRunning(ctx) + if err != nil { + t.Fatalf("ListExecutorsRunning: %v", err) + } + var found bool + for _, row := range rows { + if row.SessionID == "session-local" { + found = true + if row.LocalPID != 515151 { + t.Fatalf("list did not scan local_pid: got %d want 515151", row.LocalPID) + } + } + } + if !found { + t.Fatalf("session-local row missing from ListExecutorsRunning") + } +} + +// TestExecutorRunningLocalPIDSeparateFromSSHPID guards the #1597 pid-semantics rule +// that a local liveness handle must never overload the SSH pid column: an SSH row +// carries pid (remote host) with local_pid==0, and a local row carries local_pid +// with pid==0, and neither leaks into the other. +func TestExecutorRunningLocalPIDSeparateFromSSHPID(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + + seedExecutorRunningCleanupTask(t, repo, "task-1") + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ID: "session-ssh", TaskID: "task-1", State: models.TaskSessionStateRunning}); err != nil { + t.Fatalf("CreateTaskSession(session-ssh): %v", err) + } + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "session-ssh", + SessionID: "session-ssh", + TaskID: "task-1", + ExecutorID: "exec-ssh", + Runtime: agentruntime.RuntimeSSH, + Status: models.ExecutorRunningStatusRunning, + PID: 9999, // remote-host agentctl pid + LocalPID: 0, + }); err != nil { + t.Fatalf("UpsertExecutorRunning(session-ssh): %v", err) + } + + got, err := repo.GetExecutorRunningBySessionID(ctx, "session-ssh") + if err != nil { + t.Fatalf("GetExecutorRunningBySessionID(session-ssh): %v", err) + } + if got.PID != 9999 { + t.Fatalf("ssh pid clobbered: got %d want 9999", got.PID) + } + if got.LocalPID != 0 { + t.Fatalf("ssh row must not carry a local_pid: got %d want 0", got.LocalPID) + } +} + +// TestRepairExecutorRunningDead proves the repair-in-place primitive of the +// resume-safety invariant (#1597 resume-safety invariant): a row whose process is +// gone is marked stopped with its local liveness handle cleared, WITHOUT losing +// the resume_token / worktree that keep the session resumable. +func TestRepairExecutorRunningDead(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + + seedExecutorRunningCleanupTask(t, repo, "task-1") + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ID: "session-r", TaskID: "task-1", State: models.TaskSessionStateWaitingForInput}); err != nil { + t.Fatalf("CreateTaskSession(session-r): %v", err) + } + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "session-r", + SessionID: "session-r", + TaskID: "task-1", + ExecutorID: "exec-local", + Runtime: agentruntime.RuntimeStandalone, + Status: models.ExecutorRunningStatusRunning, + Resumable: true, + ResumeToken: "acp-resume-xyz", + WorktreePath: "/tmp/wt/session-r", + WorktreeBranch: "feature/x", + LocalPID: 777777, + }); err != nil { + t.Fatalf("UpsertExecutorRunning(session-r): %v", err) + } + + if err := repo.RepairExecutorRunningDead(ctx, "session-r"); err != nil { + t.Fatalf("RepairExecutorRunningDead: %v", err) + } + + got, err := repo.GetExecutorRunningBySessionID(ctx, "session-r") + if err != nil { + t.Fatalf("GetExecutorRunningBySessionID(session-r): %v", err) + } + if got.Status != models.ExecutorRunningStatusStopped { + t.Fatalf("repair status: got %q want stopped", got.Status) + } + if got.LocalPID != 0 { + t.Fatalf("repair must clear the local liveness handle: got local_pid=%d", got.LocalPID) + } + if got.ResumeToken != "acp-resume-xyz" { + t.Fatalf("repair must preserve resume_token: got %q", got.ResumeToken) + } + if got.WorktreePath != "/tmp/wt/session-r" || got.WorktreeBranch != "feature/x" { + t.Fatalf("repair must preserve worktree columns: got path=%q branch=%q", got.WorktreePath, got.WorktreeBranch) + } + if got.LastSeenAt == nil { + t.Fatal("repair must stamp a fresh last_seen_at observation") + } + + // Missing row → ErrExecutorRunningNotFound (idempotent-friendly for callers). + if err := repo.RepairExecutorRunningDead(ctx, "no-such-session"); !errors.Is(err, models.ErrExecutorRunningNotFound) { + t.Fatalf("repair on missing row: got %v want ErrExecutorRunningNotFound", err) + } +} + +// TestExecutorRunningLocalPIDMigrationOnLegacyDB is the same-database replay test +// mandated by ADR 0027 for a schema change: it proves the `local_pid` ADD COLUMN +// migration upgrades a DB that PREDATES the column, adding it with the default +// while leaving existing rows intact and readable. This is the real production +// upgrade path — the operator's existing executors_running rows (all pid=0) gain +// local_pid=0 on first boot of the new binary, not a fresh install. +func TestExecutorRunningLocalPIDMigrationOnLegacyDB(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + + // Seed a row on the current (migrated) schema. + seedExecutorRunningCleanupTask(t, repo, "task-legacy") + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "session-legacy", TaskID: "task-legacy", State: models.TaskSessionStateWaitingForInput, + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "session-legacy", + SessionID: "session-legacy", + TaskID: "task-legacy", + ExecutorID: "exec-legacy", + Runtime: agentruntime.RuntimeStandalone, + Status: models.ExecutorRunningStatusStarting, + Resumable: true, + ResumeToken: "legacy-resume-token", + LocalPID: 321, + }); err != nil { + t.Fatalf("UpsertExecutorRunning: %v", err) + } + + // Rewind to a pre-local_pid schema: drop the column so the row now looks like + // one written by an older binary that never had it. + if _, err := repo.db.Exec(`ALTER TABLE executors_running DROP COLUMN local_pid`); err != nil { + t.Fatalf("simulate legacy schema (drop local_pid): %v", err) + } + + // Re-run migrations, exactly as a boot of the new binary would. The + // idempotent ADD COLUMN must re-add local_pid without erroring. + if err := repo.runMigrations(); err != nil { + t.Fatalf("runMigrations on legacy DB: %v", err) + } + + // The pre-existing row survives, is readable, and its resume_token is intact — + // the migration must not drop rows. local_pid comes back as the column default + // (0) for a row that predated the column. + got, err := repo.GetExecutorRunningBySessionID(ctx, "session-legacy") + if err != nil { + t.Fatalf("legacy row must survive the migration: %v", err) + } + if got.LocalPID != 0 { + t.Errorf("legacy row local_pid = %d, want 0 (column default after ADD COLUMN)", got.LocalPID) + } + if got.ResumeToken != "legacy-resume-token" { + t.Errorf("resume_token lost across migration: got %q", got.ResumeToken) + } + if got.Status != models.ExecutorRunningStatusStarting { + t.Errorf("status lost across migration: got %q", got.Status) + } + + // And the re-added column is fully usable: a repair writes/reads local_pid=0. + if err := repo.RepairExecutorRunningDead(ctx, "session-legacy"); err != nil { + t.Fatalf("repair after migration: %v", err) + } + repaired, err := repo.GetExecutorRunningBySessionID(ctx, "session-legacy") + if err != nil { + t.Fatalf("read after repair: %v", err) + } + if repaired.Status != models.ExecutorRunningStatusStopped || repaired.LocalPID != 0 { + t.Errorf("re-added local_pid column must be writable; got status=%q local_pid=%d", repaired.Status, repaired.LocalPID) + } +} diff --git a/apps/backend/internal/task/repository/sqlite/postgres_schema_test.go b/apps/backend/internal/task/repository/sqlite/postgres_schema_test.go index 5f431d10dc..b6961e1d0e 100644 --- a/apps/backend/internal/task/repository/sqlite/postgres_schema_test.go +++ b/apps/backend/internal/task/repository/sqlite/postgres_schema_test.go @@ -5,10 +5,68 @@ import ( "testing" "time" + "github.com/kandev/kandev/internal/agentruntime" "github.com/kandev/kandev/internal/task/models" "github.com/kandev/kandev/internal/testutil" ) +// TestPostgresExecutorRunningLocalPIDMigration is the Postgres counterpart to +// TestExecutorRunningLocalPIDMigrationOnLegacyDB (SQLite): local_pid is on the +// shared migration path, so ADR 0027 asks for env-gated Postgres replay coverage +// too. It rewinds to a pre-local_pid schema, re-runs migrations, and asserts the +// ADD COLUMN re-adds the column with its default while an existing row survives. +// Skips unless KANDEV_TEST_POSTGRES_DSN is set. +func TestPostgresExecutorRunningLocalPIDMigration(t *testing.T) { + db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + repo, err := NewWithDB(db, db, nil) + if err != nil { + t.Fatalf("init postgres schema: %v", err) + } + ctx := context.Background() + + now := time.Now().UTC() + if _, err := db.Exec(db.Rebind(` + INSERT INTO tasks (id, workspace_id, title, created_at, updated_at) + VALUES (?, ?, ?, ?, ?) + `), "task-pg", "ws-task-pg", "Task pg", now, now); err != nil { + t.Fatalf("seed task: %v", err) + } + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "session-pg", TaskID: "task-pg", State: models.TaskSessionStateWaitingForInput, + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "session-pg", SessionID: "session-pg", TaskID: "task-pg", ExecutorID: "exec-pg", + Runtime: agentruntime.RuntimeStandalone, Status: models.ExecutorRunningStatusStarting, + Resumable: true, ResumeToken: "pg-resume-token", LocalPID: 321, + }); err != nil { + t.Fatalf("UpsertExecutorRunning: %v", err) + } + + // Rewind to a pre-local_pid schema, then re-migrate as a new-binary boot would. + if _, err := db.Exec(`ALTER TABLE executors_running DROP COLUMN local_pid`); err != nil { + t.Fatalf("simulate legacy schema (drop local_pid): %v", err) + } + if err := repo.runMigrations(); err != nil { + t.Fatalf("runMigrations on legacy postgres DB: %v", err) + } + + got, err := repo.GetExecutorRunningBySessionID(ctx, "session-pg") + if err != nil { + t.Fatalf("legacy row must survive the migration: %v", err) + } + if got.LocalPID != 0 { + t.Errorf("legacy row local_pid = %d, want 0 (column default after ADD COLUMN)", got.LocalPID) + } + if got.ResumeToken != "pg-resume-token" { + t.Errorf("resume_token lost across migration: got %q", got.ResumeToken) + } + if got.Status != models.ExecutorRunningStatusStarting { + t.Errorf("status lost across migration: got %q", got.Status) + } +} + func TestPostgresSchemaReinitializes(t *testing.T) { db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) diff --git a/apps/web/e2e/helpers/type-while-busy.ts b/apps/web/e2e/helpers/type-while-busy.ts new file mode 100644 index 0000000000..312c0c10bd --- /dev/null +++ b/apps/web/e2e/helpers/type-while-busy.ts @@ -0,0 +1,26 @@ +import { type Locator, type Page } from "@playwright/test"; + +/** + * Type text into the TipTap editor while the agent is busy. + * fill() silently fails on TipTap when the busy placeholder is shown, + * so we retry clicking and typing until text appears in the editor. + */ +export async function typeWhileBusy(page: Page, editor: Locator, text: string): Promise { + const modifier = process.platform === "darwin" ? "Meta" : "Control"; + await editor.scrollIntoViewIfNeeded(); + for (let attempt = 0; attempt < 3; attempt++) { + const box = await editor.boundingBox(); + if (!box) throw new Error("Editor bounding box not found"); + await page.mouse.click(box.x + 20, box.y + box.height / 2); + await page.waitForTimeout(200); + await page.keyboard.type(text); + await page.waitForTimeout(100); + const content = await editor.textContent(); + if (content?.includes(text)) return; + // Text wasn't entered; select all and clear for retry + await page.keyboard.press(`${modifier}+a`); + await page.keyboard.press("Backspace"); + await page.waitForTimeout(200); + } + throw new Error(`Failed to type "${text}" into editor after 3 attempts`); +} diff --git a/apps/web/e2e/tests/chat/message-queue.spec.ts b/apps/web/e2e/tests/chat/message-queue.spec.ts index dc0a8b42a1..0c929e06f6 100644 --- a/apps/web/e2e/tests/chat/message-queue.spec.ts +++ b/apps/web/e2e/tests/chat/message-queue.spec.ts @@ -2,37 +2,9 @@ import { type Locator, type Page } from "@playwright/test"; import { test, expect } from "../../fixtures/test-base"; import type { SeedData } from "../../fixtures/test-base"; import type { ApiClient } from "../../helpers/api-client"; +import { typeWhileBusy } from "../../helpers/type-while-busy"; import { SessionPage } from "../../pages/session-page"; -// --------------------------------------------------------------------------- -// Shared helpers -// --------------------------------------------------------------------------- - -/** - * Type text into the TipTap editor while the agent is busy. - * fill() silently fails on TipTap when the busy placeholder is shown, - * so we retry clicking and typing until text appears in the editor. - */ -async function typeWhileBusy(page: Page, editor: Locator, text: string): Promise { - const modifier = process.platform === "darwin" ? "Meta" : "Control"; - await editor.scrollIntoViewIfNeeded(); - for (let attempt = 0; attempt < 3; attempt++) { - const box = await editor.boundingBox(); - if (!box) throw new Error("Editor bounding box not found"); - await page.mouse.click(box.x + 20, box.y + box.height / 2); - await page.waitForTimeout(200); - await page.keyboard.type(text); - await page.waitForTimeout(100); - const content = await editor.textContent(); - if (content?.includes(text)) return; - // Text wasn't entered; select all and clear for retry - await page.keyboard.press(`${modifier}+a`); - await page.keyboard.press("Backspace"); - await page.waitForTimeout(200); - } - throw new Error(`Failed to type "${text}" into editor after 3 attempts`); -} - // --------------------------------------------------------------------------- // Quick Chat queue tests // --------------------------------------------------------------------------- diff --git a/apps/web/e2e/tests/session/pause-resume-recovery.spec.ts b/apps/web/e2e/tests/session/pause-resume-recovery.spec.ts new file mode 100644 index 0000000000..76bbc24af1 --- /dev/null +++ b/apps/web/e2e/tests/session/pause-resume-recovery.spec.ts @@ -0,0 +1,139 @@ +import { type Page } from "@playwright/test"; + +import { test, expect } from "../../fixtures/test-base"; +import type { SeedData } from "../../fixtures/test-base"; +import type { ApiClient } from "../../helpers/api-client"; +import { typeWhileBusy } from "../../helpers/type-while-busy"; +import { SessionPage } from "../../pages/session-page"; + +// --------------------------------------------------------------------------- +// Pause → resume recovery (#1597 pause→resume recovery) +// +// The operator's headline pain: pause a running agent turn and the session +// wedges — the next message is dropped or the composer stays stuck "running", +// and the only recovery is to restart the whole headless service. These tests +// pin the corrected behavior end-to-end against the mock agent: +// +// 1. Pausing a running turn returns the SAME session to an input-ready state; +// a newly typed message resumes it with prior context intact — no wedged +// "still running" composer, no service restart. +// 2. A message queued while the turn was running is DELIVERED once the pause +// settles the session, rather than being stranded on the queue. +// --------------------------------------------------------------------------- + +/** Seed an ACP task, open its session, and wait for the initial turn to idle. */ +async function seedTaskAndWaitForIdle( + testPage: Page, + apiClient: ApiClient, + seedData: SeedData, + title: string, +): Promise { + const task = await apiClient.createTaskWithAgent( + seedData.workspaceId, + title, + seedData.agentProfileId, + { + description: "/e2e:simple-message", + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + repository_ids: [seedData.repositoryId], + }, + ); + + await testPage.goto(`/t/${task.id}`); + const session = new SessionPage(testPage); + await session.waitForLoad(); + await session.waitForChatIdle({ timeout: 30_000 }); + return session; +} + +test.describe("Pause → resume recovery", () => { + // Cancel/resume timing can be sensitive under CI load. + test.describe.configure({ retries: 1 }); + + test("pausing a running turn lets a newly typed message resume the same session", async ({ + testPage, + apiClient, + seedData, + }) => { + test.setTimeout(90_000); + + const session = await seedTaskAndWaitForIdle( + testPage, + apiClient, + seedData, + "Pause then resume", + ); + + // The initial turn produced one response — this is the context that must + // survive the pause and prove the same session (not a fresh one) resumed. + await expect( + session.chat.getByText("simple mock response", { exact: false }).nth(0), + ).toBeVisible({ timeout: 30_000 }); + const sessionUrl = testPage.url(); + + // Start a long-running turn so we have time to pause it mid-flight. + await session.sendMessage("/slow 8s"); + await expect(session.agentStatus()).toBeVisible({ timeout: 15_000 }); + await expect(session.cancelAgentButton()).toBeVisible({ timeout: 15_000 }); + + // Pause: the operator stops the running turn. + await session.cancelAgentButton().click(); + + // The session must settle back to input-ready — NOT wedge on "still + // running". This is the regression that forced a service restart. + await expect(session.agentStatus()).not.toBeVisible({ timeout: 30_000 }); + await expect(session.idleInput()).toBeVisible({ timeout: 30_000 }); + await expect(session.cancelAgentButton()).not.toBeVisible({ timeout: 15_000 }); + + // Still the same session — no navigation / service bounce. + expect(testPage.url()).toBe(sessionUrl); + + // Sending a new message resumes the SAME session with its context intact. + await session.sendMessage("/e2e:simple-message"); + await session.expectChatResponseVisible("simple mock response", 1, { timeout: 30_000 }); + + // Prior conversation is still present — context was preserved across the pause. + await expect( + session.chat.getByText("simple mock response", { exact: false }).nth(0), + ).toBeVisible({ timeout: 15_000 }); + }); + + test("a message queued during a running turn is delivered when the turn is paused", async ({ + testPage, + apiClient, + seedData, + }) => { + test.setTimeout(90_000); + + const session = await seedTaskAndWaitForIdle( + testPage, + apiClient, + seedData, + "Pause drains queued message", + ); + + // Keep the agent busy, then queue a follow-up while it is running. + await session.sendMessage("/slow 8s"); + await expect(session.agentStatus()).toBeVisible({ timeout: 15_000 }); + await expect(session.cancelAgentButton()).toBeVisible({ timeout: 15_000 }); + + const editor = testPage.locator(".tiptap.ProseMirror").first(); + await typeWhileBusy(testPage, editor, "/e2e:simple-message"); + await testPage.getByTestId("submit-message-button").click(); + + // The queued message is parked while the turn runs. + await expect(testPage.getByTestId("queue-chip")).toBeVisible({ timeout: 10_000 }); + + // Pause the running turn. Cancel must DELIVER the queued message rather than + // strand it: on an escalated / dead-process cancel no agent.ready fires, so + // CancelAgent drains the queue directly. + await session.cancelAgentButton().click(); + + // The queued message drains (chip clears) and its response arrives — proof + // the paused session accepted the queued input instead of losing it. + await expect(testPage.getByTestId("queue-chip")).not.toBeVisible({ timeout: 30_000 }); + await session.expectChatResponseVisible("simple mock response", 1, { timeout: 30_000 }); + await expect(session.idleInput()).toBeVisible({ timeout: 30_000 }); + }); +}); diff --git a/docs/decisions/0025-runtime-cleanup-uses-executors-running.md b/docs/decisions/0025-runtime-cleanup-uses-executors-running.md index 652334cae1..1c3fd4084d 100644 --- a/docs/decisions/0025-runtime-cleanup-uses-executors-running.md +++ b/docs/decisions/0025-runtime-cleanup-uses-executors-running.md @@ -1,9 +1,46 @@ # 0025: Runtime Cleanup Uses `executors_running` -**Status:** accepted +**Status:** accepted (amended 2026-07-06 — see "Update") **Date:** 2026-06-22 **Area:** backend +## Update (2026-07-06, #1597 executor-row-desync) + +This decision **stays**: `executors_running` remains the authoritative durable +runtime inventory. It was made *trustworthy* rather than reverted. Three +clarifications now hold: + +- **Events are the primary producer; startup reconciliation heals what events + cannot.** Every lifecycle transition writes the row (launch, boot-ready, + turn-complete, cancel, process-exit/crash, stop), populating a host-local + liveness handle (`executors_running.local_pid`) for local/standalone rows. A + startup pass repairs rows whose process is confirmed dead and prunes only + terminal ones — a backend restart is exactly the moment events could not + have fired. A periodic in-run polling pass was prototyped and deliberately + not merged: it defended against failure modes (OOM kill, dropped event) + that have not been observed, and polling never updates a row the moment an + event could. +- **`local_pid` is a NEW column, deliberately not the existing `pid`.** + `executors_running.pid` holds the agentctl PID *on the remote host* for SSH + rows; overloading it with a host-local pid would silently change that + column's meaning and invite local process checks against remote rows. The + two handles stay separate, and liveness is runtime-aware + (`lifecycle.RowProcessLiveness`): a host-local process check never runs + against a remote/SSH or containerized row — those report Unknown, never + Dead. +- **One ironclad deletion invariant governs every reconciliation cleanup + path.** A row backing a resumable session, or holding a `resume_token`, is + **repaired in place** (status=stopped, `local_pid` cleared, + `resume_token`/worktree preserved) — never deleted; only a + finished/never-started row with no `resume_token` may be pruned. + Reconciliation repairs rather than deletes-and-relaunches because + `resume_token` is single-sourced in this table — an erroneous delete costs + the operator the only handle to a resumable conversation. Duplicating the + token into a second table was rejected: two writers of the same fact is the + divergence pattern this decision exists to eliminate. The rule is + `models.RowMustBePreserved`; see #1597 for the measured evidence and + expected behavior this implements. + ## Context Archiving and deleting tasks can remove task/worktree records while ACP agent diff --git a/docs/decisions/INDEX.md b/docs/decisions/INDEX.md index 92cbd1a51d..d486646bf5 100644 --- a/docs/decisions/INDEX.md +++ b/docs/decisions/INDEX.md @@ -30,7 +30,7 @@ Read individual ADRs for full context. Create new ones via `/record decision` or | 0022 | [Embedded Vite assets](0022-embedded-vite-assets.md) | accepted | backend, frontend, cli | 2026-06-17 | | 0023 | [Active workspace cookie for boot state](0023-active-workspace-cookie.md) | accepted | backend, frontend | 2026-06-18 | | 0024 | [Go-fronted Vite dev mode](0024-go-fronted-vite-dev-mode.md) | accepted | backend, frontend, cli | 2026-06-18 | -| 0025 | [Runtime cleanup uses `executors_running`](0025-runtime-cleanup-uses-executors-running.md) | accepted | backend | 2026-06-22 | +| 0025 | [Runtime cleanup uses `executors_running`](0025-runtime-cleanup-uses-executors-running.md) | accepted (amended 2026-07-06) | backend | 2026-06-22 | | 0026 | [Tauri desktop shell over native runtime](0026-tauri-desktop-shell.md) | accepted | frontend, backend, cli, infra | 2026-06-23 | | 0027 | [Replayable schema migrations across SQLite and Postgres](0027-replayable-schema-migrations.md) | accepted | backend | 2026-06-24 | | 0028 | [Backend-owned task-create last-used preferences](0028-task-create-last-used-source-of-truth.md) | accepted | backend, frontend | 2026-06-29 |