fix(executors): truthful executor rows, startup reconciliation, resume-safe cleanup (#1597) (#1608)

Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
point-source
2026-07-08 15:37:02 -07:00
committed by GitHub
parent bd22a1318c
commit e1cf99f821
37 changed files with 1855 additions and 99 deletions
@@ -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 {
@@ -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()
@@ -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
}
@@ -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
}
@@ -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)
}
@@ -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
}
@@ -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.
@@ -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),
@@ -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.
@@ -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)
@@ -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")
}
}
@@ -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)
@@ -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(),
@@ -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,
@@ -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.
@@ -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
@@ -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
@@ -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))
}
}
@@ -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)
}
}
+35 -16
View File
@@ -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)
}
}
@@ -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
}
@@ -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)
}
}
@@ -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
}
+30 -22
View File
@@ -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.
@@ -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)
}
@@ -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)
}
})
}
}
@@ -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.
@@ -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
@@ -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 '',
@@ -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 {
@@ -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)
}
}
@@ -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))
+26
View File
@@ -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<void> {
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`);
}
+1 -29
View File
@@ -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<void> {
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
// ---------------------------------------------------------------------------
@@ -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<SessionPage> {
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 });
});
});
@@ -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
+1 -1
View File
@@ -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 |