fix(backend): harden office-disabled cron, title limits, and dead-runtime cleanup (#2206)
This commit is contained in:
@@ -22,8 +22,19 @@ func TestReconcileWorkspaceSources_RejectsMissingFolderTarget(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// canonicalTempDir resolves symlinks in t.TempDir() so tests hand production
|
||||
// code a canonical owned control root (macOS /var -> /private/var); no-op on Linux.
|
||||
func canonicalTempDir(t *testing.T) string {
|
||||
t.Helper()
|
||||
d, err := filepath.EvalSymlinks(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatalf("EvalSymlinks: %v", err)
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
func TestReconcileWorkspaceRepositories_RecreatesMissingOwnedLink(t *testing.T) {
|
||||
root, source := t.TempDir(), t.TempDir()
|
||||
root, source := canonicalTempDir(t), t.TempDir()
|
||||
writeMarker(t, source)
|
||||
if err := reconcileWorkspaceRepositories(root, []WorkspaceRepositorySpec{{RepoName: "api", RepositoryPath: source}}, nil); err != nil {
|
||||
t.Fatalf("reconcileWorkspaceRepositories: %v", err)
|
||||
@@ -77,7 +88,7 @@ func TestReconcileWorkspaceRepositories_SkipsRepositoryThatIsWorkspaceRoot(t *te
|
||||
// contract though — the user has to learn the entry is there — so the warning
|
||||
// and its structured entry path are asserted too.
|
||||
func TestReconcileWorkspaceRepositories_PreservesAndReportsPreExistingSelfLink(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
root := canonicalTempDir(t)
|
||||
marker := writeMarker(t, root)
|
||||
if _, err := worktree.CreateOwnedDirectoryLink(root, "api", root); err != nil {
|
||||
t.Fatalf("seed self link: %v", err)
|
||||
@@ -127,7 +138,7 @@ func TestReconcileWorkspaceRepositories_DoesNotWarnWithoutSelfLink(t *testing.T)
|
||||
|
||||
// The guard is per spec, not a blanket skip: siblings still need their links.
|
||||
func TestReconcileWorkspaceRepositories_LinksSiblingWhenPrimaryIsWorkspaceRoot(t *testing.T) {
|
||||
root, sibling := t.TempDir(), t.TempDir()
|
||||
root, sibling := canonicalTempDir(t), t.TempDir()
|
||||
writeMarker(t, sibling)
|
||||
|
||||
err := reconcileWorkspaceRepositories(root, []WorkspaceRepositorySpec{
|
||||
@@ -231,7 +242,7 @@ func TestReconcileWorkspaceRepositories_RejectsTraversalRepoName(t *testing.T) {
|
||||
// where the first repository is a real sibling — so the guard must not degrade
|
||||
// into skipping index 0.
|
||||
func TestReconcileWorkspaceRepositories_LinksPrimaryWhenRootIsTaskDirectory(t *testing.T) {
|
||||
root := filepath.Join(t.TempDir(), "tasks", "task-1")
|
||||
root := filepath.Join(canonicalTempDir(t), "tasks", "task-1")
|
||||
primary := t.TempDir()
|
||||
writeMarker(t, primary)
|
||||
|
||||
|
||||
@@ -359,7 +359,7 @@ func TestWorkspaceRemoveMaterializedRepository_RejectsSymlinkedGitMetadata(t *te
|
||||
|
||||
func TestWorkspaceRemoveMaterializedRepository_RestoresSwappedReplacement(t *testing.T) {
|
||||
origin := createMaterializeOrigin(t)
|
||||
workDir := t.TempDir()
|
||||
workDir := canonicalTempDir(t)
|
||||
destination := filepath.Join(workDir, "second-repo")
|
||||
if _, err := materializeRepository(context.Background(), origin, destination, "main", ""); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -393,7 +393,7 @@ func TestWorkspaceRemoveMaterializedRepository_RestoresSwappedReplacement(t *tes
|
||||
|
||||
func TestWorkspaceRemoveMaterializedRepository_RejectsPostQuarantineReplacement(t *testing.T) {
|
||||
origin := createMaterializeOrigin(t)
|
||||
workDir := t.TempDir()
|
||||
workDir := canonicalTempDir(t)
|
||||
destination := filepath.Join(workDir, "second-repo")
|
||||
if _, err := materializeRepository(context.Background(), origin, destination, "main", ""); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -441,6 +441,16 @@ func TestWorkspaceRemoveMaterializedRepository_RejectsTraversal(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// canonicalTempDir resolves symlinks so expected paths match the canonical roots production computes (no-op on Linux).
|
||||
func canonicalTempDir(t *testing.T) string {
|
||||
t.Helper()
|
||||
d, err := filepath.EvalSymlinks(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatalf("EvalSymlinks: %v", err)
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
func newMaterializeTestServer(t *testing.T, workDir string) *Server {
|
||||
t.Helper()
|
||||
log := newTestLogger()
|
||||
|
||||
@@ -106,7 +106,11 @@ func TestRunGitOutput_PollingVariantReleasesThrottleOnTimeout(t *testing.T) {
|
||||
|
||||
func TestWorkspaceGitAdmissionWaitDoesNotConsumeTimeout(t *testing.T) {
|
||||
prev := gitCommandTimeout
|
||||
gitCommandTimeout = 100 * time.Millisecond
|
||||
// The execution budget must comfortably exceed a cold shell-shim start
|
||||
// (slow on macOS), or the freshly-admitted command is SIGKILLed before it
|
||||
// can print. The hold below is still longer than this budget, so the test
|
||||
// still proves the queued command gets a fresh budget after admission.
|
||||
gitCommandTimeout = 2 * time.Second
|
||||
t.Cleanup(func() { gitCommandTimeout = prev })
|
||||
|
||||
restoreCap := subproc.Git().SetCapForTest(1)
|
||||
@@ -132,8 +136,8 @@ func TestWorkspaceGitAdmissionWaitDoesNotConsumeTimeout(t *testing.T) {
|
||||
}()
|
||||
waitForAnyGitWaiter(t, 1)
|
||||
// Hold the slot longer than the execution timeout. The queued command
|
||||
// should still receive a fresh 100ms execution budget after release.
|
||||
timer := time.NewTimer(150 * time.Millisecond)
|
||||
// should still receive a fresh execution budget after release.
|
||||
timer := time.NewTimer(2500 * time.Millisecond)
|
||||
<-timer.C
|
||||
hold()
|
||||
|
||||
@@ -142,7 +146,7 @@ func TestWorkspaceGitAdmissionWaitDoesNotConsumeTimeout(t *testing.T) {
|
||||
if result.err != nil || string(result.out) != "ok" {
|
||||
t.Fatalf("result = (%q, %v), want fast command after queued admission", result.out, result.err)
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("queued git command did not complete")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
|
||||
"github.com/kandev/kandev/internal/agent/agents"
|
||||
"github.com/kandev/kandev/internal/agent/registry"
|
||||
runtimeapi "github.com/kandev/kandev/internal/agent/runtime"
|
||||
"github.com/kandev/kandev/internal/agent/runtime/agentctl"
|
||||
"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
|
||||
"github.com/kandev/kandev/internal/agentctl/types/streams"
|
||||
@@ -322,8 +323,25 @@ func (a *lifecycleAdapter) StopAgent(ctx context.Context, agentInstanceID string
|
||||
}
|
||||
|
||||
// StopAgentWithReason stops a running agent and propagates the stop reason to runtime teardown.
|
||||
// The lifecycle-tier not-found sentinel is normalized to the public runtime
|
||||
// sentinel at this seam so orchestrator consumers depend only on
|
||||
// runtimeapi.ErrNotFound and never import runtime/lifecycle.
|
||||
func (a *lifecycleAdapter) StopAgentWithReason(ctx context.Context, agentInstanceID string, reason string, force bool) error {
|
||||
return a.mgr.StopAgentWithReason(ctx, agentInstanceID, reason, force)
|
||||
return normalizeRuntimeStopError(a.mgr.StopAgentWithReason(ctx, agentInstanceID, reason, force))
|
||||
}
|
||||
|
||||
// normalizeRuntimeStopError maps the backend lifecycle not-found sentinel onto
|
||||
// the public runtime not-found sentinel while preserving the original error for
|
||||
// diagnostics. Unrelated errors (and nil) pass through unchanged so a transient
|
||||
// runtime failure is never mistaken for an absent runtime.
|
||||
func normalizeRuntimeStopError(err error) error {
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
if errors.Is(err, lifecycle.ErrExecutionNotFound) {
|
||||
return errors.Join(runtimeapi.ErrNotFound, err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// RowLiveness classifies the liveness of the OS process backing an
|
||||
|
||||
@@ -9,6 +9,8 @@ import (
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
runtimeapi "github.com/kandev/kandev/internal/agent/runtime"
|
||||
"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
githubsvc "github.com/kandev/kandev/internal/github"
|
||||
"github.com/kandev/kandev/internal/repoclone"
|
||||
@@ -223,7 +225,7 @@ func TestResolveForReviewRedetectsStoredMasterAfterClonePath(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatalf("CreateWorkspace: %v", err)
|
||||
}
|
||||
basePath := t.TempDir()
|
||||
basePath := canonicalTempDir(t)
|
||||
cloner := repoclone.NewCloner(
|
||||
repoclone.Config{BasePath: basePath}, repoclone.ProtocolHTTPS, "", newTestLogger(),
|
||||
)
|
||||
@@ -414,6 +416,30 @@ func TestWrapGitHubTaskIssueStoreError(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestNormalizeRuntimeStopError asserts the lifecycle adapter maps the backend
|
||||
// lifecycle not-found sentinel onto the public runtime not-found sentinel, so
|
||||
// orchestrator/reconciliation can depend only on runtimeapi.ErrNotFound and
|
||||
// never has to import runtime/lifecycle. Unrelated errors pass through unchanged.
|
||||
func TestNormalizeRuntimeStopError(t *testing.T) {
|
||||
if got := normalizeRuntimeStopError(nil); got != nil {
|
||||
t.Fatalf("normalizeRuntimeStopError(nil) = %v, want nil", got)
|
||||
}
|
||||
|
||||
wrapped := fmt.Errorf("stop agent: %w", lifecycle.ErrExecutionNotFound)
|
||||
got := normalizeRuntimeStopError(wrapped)
|
||||
if !errors.Is(got, runtimeapi.ErrNotFound) {
|
||||
t.Fatalf("lifecycle not-found must normalize to runtimeapi.ErrNotFound; got %v", got)
|
||||
}
|
||||
if !errors.Is(got, lifecycle.ErrExecutionNotFound) {
|
||||
t.Fatalf("original lifecycle sentinel must be preserved for diagnostics; got %v", got)
|
||||
}
|
||||
|
||||
other := errors.New("agentctl unreachable")
|
||||
if got := normalizeRuntimeStopError(other); got != other {
|
||||
t.Fatalf("unrelated error must pass through unchanged; got %v, want %v", got, other)
|
||||
}
|
||||
}
|
||||
|
||||
// jiraSecretAdapter Set/Exists branching is tested in
|
||||
// internal/integrations/secretadapter/secretadapter_test.go now that the
|
||||
// upsert helper lives there.
|
||||
|
||||
@@ -19,6 +19,16 @@ import (
|
||||
"github.com/kandev/kandev/internal/worktree"
|
||||
)
|
||||
|
||||
// canonicalTempDir resolves symlinks so macOS /var temp roots pass owned-root guards.
|
||||
func canonicalTempDir(t *testing.T) string {
|
||||
t.Helper()
|
||||
d, err := filepath.EvalSymlinks(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatalf("EvalSymlinks: %v", err)
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
// stubRescanner records calls so the materializer test can assert that the
|
||||
// agentctl rescan was triggered with the expected work_dir (task root) and
|
||||
// that the frontend-visible "worktree materialized" event also fired.
|
||||
@@ -198,7 +208,7 @@ func TestBranchMaterializer_SecondBranchKeepsTaskRootPromoted(t *testing.T) {
|
||||
func setupMaterializerScenario(t *testing.T) (repoPath, taskRoot, primaryPath string) {
|
||||
t.Helper()
|
||||
|
||||
tmp := t.TempDir()
|
||||
tmp := canonicalTempDir(t)
|
||||
bareDir := filepath.Join(tmp, "origin.git")
|
||||
runGit(t, tmp, "init", "--bare", "-b", "main", bareDir)
|
||||
|
||||
|
||||
@@ -36,7 +36,15 @@ func startCronScheduler(
|
||||
) *schedulercron.Loop {
|
||||
heartbeat := buildHeartbeatHandler(repos, dispatcher, log)
|
||||
budget := buildBudgetHandler(repos, dispatcher, log)
|
||||
routines := schedulercron.NewRoutinesHandler(routineSvc, nil, log)
|
||||
// Only wire a routines ticker when the office routines service exists.
|
||||
// Assigning a nil *RoutineService straight into the RoutineTicker
|
||||
// interface would produce a non-nil typed-nil interface and panic on
|
||||
// every tick when Office is disabled.
|
||||
var routineTicker schedulercron.RoutineTicker
|
||||
if routineSvc != nil {
|
||||
routineTicker = routineSvc
|
||||
}
|
||||
routines := schedulercron.NewRoutinesHandler(routineTicker, nil, log)
|
||||
loop := schedulercron.NewLoop(schedulercron.DefaultTickInterval, log,
|
||||
heartbeat, budget, routines, officeRecovery)
|
||||
loop.Start(ctx)
|
||||
|
||||
@@ -52,6 +52,21 @@ func freePort(t *testing.T) int {
|
||||
return port
|
||||
}
|
||||
|
||||
// requireLoopbackAlias skips the test when host cannot be bound. Linux treats
|
||||
// all of 127.0.0.0/8 as loopback, but macOS only configures 127.0.0.1 by
|
||||
// default, so multi-loopback binding tests are unrunnable there without an
|
||||
// explicit lo0 alias.
|
||||
func requireLoopbackAlias(t *testing.T, host string) {
|
||||
t.Helper()
|
||||
ln, err := net.Listen("tcp", net.JoinHostPort(host, "0"))
|
||||
if err != nil {
|
||||
t.Skipf("loopback alias %s not available on this host: %v", host, err)
|
||||
}
|
||||
if err := ln.Close(); err != nil {
|
||||
t.Fatalf("close loopback probe listener: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func noContentServer() *http.Server {
|
||||
return &http.Server{
|
||||
Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
@@ -109,6 +124,7 @@ func shutdown(t *testing.T, listeners *serverListeners, server *http.Server) {
|
||||
}
|
||||
|
||||
func TestStartHTTPServersMultipleLoopbackAddresses(t *testing.T) {
|
||||
requireLoopbackAlias(t, "127.0.0.2")
|
||||
log := testLogger(t)
|
||||
port := freePort(t)
|
||||
server := noContentServer()
|
||||
@@ -152,6 +168,7 @@ func TestStartHTTPServersAllFailIsFatal(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestStartHTTPServersPartialFailSelfHeals(t *testing.T) {
|
||||
requireLoopbackAlias(t, "127.0.0.2")
|
||||
defer goleak.VerifyNone(t)
|
||||
|
||||
// Shorten the retry cadence for the test and restore it after.
|
||||
|
||||
@@ -104,6 +104,9 @@ func provideOrchestrator(
|
||||
orchestratorSvc.SetTaskGitCredentialPolicyResolver(githubExecutorCredentialPolicyAdapter{service: githubSvc})
|
||||
}
|
||||
taskSvc.SetExecutionStopper(orchestratorSvc)
|
||||
// Runtime-aware liveness lets durable cleanup treat a not-found stop for a
|
||||
// confirmed-dead local runtime as already stopped instead of retrying forever.
|
||||
taskSvc.SetRowLivenessProber(agentManagerClient)
|
||||
taskSvc.SetContextWindowResetter(orchestratorSvc.ResetContextWindow)
|
||||
taskSvc.SetGitArchiveCapture(orchestratorSvc)
|
||||
orchestratorSvc.SetWorktreeManager(lifecycleMgr.WorktreeManager())
|
||||
|
||||
@@ -41,6 +41,53 @@ func TestGlobalSchedulingStartsOutsideOfficeFeatureGate(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestRoutinesHandlerNotWiredWithTypedNilService guards the Office-disabled
|
||||
// cron panic regression. startCronScheduler must gate the routines service
|
||||
// behind a nil check before assigning it into the RoutineTicker interface;
|
||||
// passing the raw *RoutineService parameter (routineSvc) straight into
|
||||
// NewRoutinesHandler produces a typed-nil interface that panics every tick
|
||||
// when Office is disabled.
|
||||
func TestRoutinesHandlerNotWiredWithTypedNilService(t *testing.T) {
|
||||
fset := token.NewFileSet()
|
||||
file, err := parser.ParseFile(fset, "cron.go", nil, 0)
|
||||
if err != nil {
|
||||
t.Fatalf("parse cron.go: %v", err)
|
||||
}
|
||||
|
||||
var startFn *ast.FuncDecl
|
||||
for _, decl := range file.Decls {
|
||||
fn, ok := decl.(*ast.FuncDecl)
|
||||
if ok && fn.Name.Name == "startCronScheduler" {
|
||||
startFn = fn
|
||||
break
|
||||
}
|
||||
}
|
||||
if startFn == nil {
|
||||
t.Fatal("startCronScheduler not found in cron.go")
|
||||
}
|
||||
|
||||
var passesRawService bool
|
||||
ast.Inspect(startFn, func(n ast.Node) bool {
|
||||
call, ok := n.(*ast.CallExpr)
|
||||
if !ok {
|
||||
return true
|
||||
}
|
||||
sel, ok := call.Fun.(*ast.SelectorExpr)
|
||||
if !ok || sel.Sel.Name != "NewRoutinesHandler" || len(call.Args) == 0 {
|
||||
return true
|
||||
}
|
||||
if ident, ok := call.Args[0].(*ast.Ident); ok && ident.Name == "routineSvc" {
|
||||
passesRawService = true
|
||||
}
|
||||
return true
|
||||
})
|
||||
|
||||
if passesRawService {
|
||||
t.Error("startCronScheduler passes the raw routineSvc pointer into NewRoutinesHandler; " +
|
||||
"gate it behind a nil check via a RoutineTicker interface variable to avoid the typed-nil panic")
|
||||
}
|
||||
}
|
||||
|
||||
func countNamedCalls(fn *ast.FuncDecl, name string) int {
|
||||
count := 0
|
||||
ast.Inspect(fn, func(n ast.Node) bool {
|
||||
|
||||
@@ -269,7 +269,7 @@ func TestWorkspaceSourceMaterializer_RemoteRejectsLocalOnlyAdditionalRepositoryB
|
||||
func TestWorkspaceSourceMaterializer_LocalFolderCreatesLiveTaskEntryAndRebindsSessions(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
repo := newMaterializerRepo(t)
|
||||
tasksBase := filepath.Join(t.TempDir(), "tasks")
|
||||
tasksBase := filepath.Join(canonicalTempDir(t), "tasks")
|
||||
mgr := newMaterializerWorktreeMgr(t, filepath.Join(tasksBase, "task-1"))
|
||||
source := filepath.Join(t.TempDir(), "notes")
|
||||
if err := os.MkdirAll(source, 0o755); err != nil {
|
||||
@@ -319,7 +319,7 @@ func TestWorkspaceSourceMaterializer_LocalFolderCreatesLiveTaskEntryAndRebindsSe
|
||||
func TestWorkspaceSourceMaterializer_LocalFolderUsesPersistedBatchSourceOnlyOnce(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
repo := newMaterializerRepo(t)
|
||||
tasksBase := filepath.Join(t.TempDir(), "tasks")
|
||||
tasksBase := filepath.Join(canonicalTempDir(t), "tasks")
|
||||
mgr := newMaterializerWorktreeMgr(t, filepath.Join(tasksBase, "task-1"))
|
||||
source := filepath.Join(t.TempDir(), "notes")
|
||||
if err := os.MkdirAll(source, 0o755); err != nil {
|
||||
@@ -356,7 +356,7 @@ func TestWorkspaceSourceMaterializer_LocalFolderUsesPersistedBatchSourceOnlyOnce
|
||||
func TestWorkspaceSourceMaterializer_LocalPromotionPreservesPrimaryRepository(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
repo := newMaterializerRepo(t)
|
||||
tasksBase := filepath.Join(t.TempDir(), "tasks")
|
||||
tasksBase := filepath.Join(canonicalTempDir(t), "tasks")
|
||||
mgr := newMaterializerWorktreeMgr(t, filepath.Join(tasksBase, "task-1"))
|
||||
primary := filepath.Join(t.TempDir(), "primary")
|
||||
folder := filepath.Join(t.TempDir(), "notes")
|
||||
@@ -402,10 +402,10 @@ func TestWorkspaceSourceMaterializer_LocalPromotionPreservesPrimaryRepository(t
|
||||
func TestWorkspaceSourceMaterializer_LocalClonesProviderRepositoryBeforeLinking(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
repo := newMaterializerRepo(t)
|
||||
tasksBase := filepath.Join(t.TempDir(), "tasks")
|
||||
tasksBase := filepath.Join(canonicalTempDir(t), "tasks")
|
||||
mgr := newMaterializerWorktreeMgr(t, filepath.Join(tasksBase, "task-1"))
|
||||
seedWorkspaceSourceTask(t, repo, t.TempDir())
|
||||
clonePath := filepath.Join(t.TempDir(), "cloned")
|
||||
clonePath := filepath.Join(canonicalTempDir(t), "cloned")
|
||||
if err := os.MkdirAll(clonePath, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -481,9 +481,9 @@ func TestWorkspaceSourceMaterializer_RollsBackLinkAndPathWhenAdoptionFails(t *te
|
||||
func TestWorkspaceSourceMaterializer_RebindFailureRestoresEarlierSessionsInReverseOrder(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
repo := newMaterializerRepo(t)
|
||||
tasksBase := filepath.Join(t.TempDir(), "tasks")
|
||||
tasksBase := filepath.Join(canonicalTempDir(t), "tasks")
|
||||
mgr := newMaterializerWorktreeMgr(t, filepath.Join(tasksBase, "task-1"))
|
||||
source := filepath.Join(t.TempDir(), "notes")
|
||||
source := filepath.Join(canonicalTempDir(t), "notes")
|
||||
if err := os.MkdirAll(source, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
@@ -460,7 +460,7 @@ func TestServiceNodeToolBinDirsResolvesFnmMultishellSymlinks(t *testing.T) {
|
||||
if runtime.GOOS == "windows" {
|
||||
t.Skip("service PATH rendering is POSIX-only")
|
||||
}
|
||||
tmp := t.TempDir()
|
||||
tmp := canonicalTempDir(t)
|
||||
stableBin := filepath.Join(tmp, "home", "alice", ".local", "share", "fnm", "node-versions", "v25.2.1", "installation", "bin")
|
||||
multishellBin := filepath.Join(tmp, "run", "user", "1000", "fnm_multishells", "12345_67890", "bin")
|
||||
for _, dir := range []string{stableBin, multishellBin} {
|
||||
@@ -522,3 +522,14 @@ func TestBackupUnmanagedServiceFileSkipsManagedFile(t *testing.T) {
|
||||
t.Fatalf("managed file should not create backup, stat err=%v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// canonicalTempDir returns t.TempDir() with symlinks resolved so tests agree
|
||||
// with production code that reports canonical paths (macOS /var -> /private/var).
|
||||
func canonicalTempDir(t *testing.T) string {
|
||||
t.Helper()
|
||||
resolved, err := filepath.EvalSymlinks(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return resolved
|
||||
}
|
||||
|
||||
@@ -63,7 +63,7 @@ func TestRunStartUsesSelfExecutableAndBackendCWD(t *testing.T) {
|
||||
launchManaged = oldLaunchManaged
|
||||
})
|
||||
|
||||
exe := filepath.Join(t.TempDir(), "bin", "kandev")
|
||||
exe := filepath.Join(canonicalTempDir(t), "bin", "kandev")
|
||||
executablePath = func() (string, error) {
|
||||
return exe, nil
|
||||
}
|
||||
@@ -73,7 +73,9 @@ func TestRunStartUsesSelfExecutableAndBackendCWD(t *testing.T) {
|
||||
got = cfg
|
||||
return 42
|
||||
}
|
||||
t.Setenv("KANDEV_HOME_DIR", t.TempDir())
|
||||
// ensureDataDir rejects a data dir with any symlinked ancestor, so the home
|
||||
// dir must be canonical (macOS temp dirs sit under the /var -> /private/var link).
|
||||
t.Setenv("KANDEV_HOME_DIR", canonicalTempDir(t))
|
||||
|
||||
code := runStart(context.Background(), Options{Command: CommandStart, BackendPort: 48123, Headless: true})
|
||||
if code != 42 {
|
||||
|
||||
@@ -146,6 +146,10 @@ func TestNewServiceRetainsSystemProviderForNonDesktopLaunch(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestSemanticOccurrencesUseEventSpecificCopyAndOccurrenceIdempotency(t *testing.T) {
|
||||
// Keep the auto-provisioned system provider out so the test only exercises
|
||||
// the single configured local provider; SystemProvider.Available() is true
|
||||
// on desktop platforms (e.g. macOS) and would add an extra delivery.
|
||||
t.Setenv(desktopNativeNotificationsEnv, "true")
|
||||
log, err := logger.NewFromZap(zap.NewNop())
|
||||
if err != nil {
|
||||
t.Fatalf("create logger: %v", err)
|
||||
@@ -302,6 +306,10 @@ func TestUpdateAvailabilityRoutesProviderScopedOccurrenceWithReleasePayload(t *t
|
||||
}
|
||||
|
||||
func TestNoEligibleLocalUpdateSubscriberReleasesOccurrenceClaimForReplay(t *testing.T) {
|
||||
// Keep the auto-provisioned system provider out so the test only exercises
|
||||
// the single configured local provider; SystemProvider.Available() is true
|
||||
// on desktop platforms (e.g. macOS) and would add an extra delivery.
|
||||
t.Setenv(desktopNativeNotificationsEnv, "true")
|
||||
log, err := logger.NewFromZap(zap.NewNop())
|
||||
if err != nil {
|
||||
t.Fatalf("create logger: %v", err)
|
||||
|
||||
@@ -448,7 +448,7 @@ func buildReviewTaskRequest(evt *github.NewReviewPREvent, repositories []ReviewT
|
||||
WorkspaceID: evt.WorkspaceID,
|
||||
WorkflowID: evt.WorkflowID,
|
||||
WorkflowStepID: evt.WorkflowStepID,
|
||||
Title: fmt.Sprintf("PR #%d: %s", pr.Number, pr.Title),
|
||||
Title: service.TruncateTaskTitle(fmt.Sprintf("PR #%d: %s", pr.Number, pr.Title)),
|
||||
Description: interpolateReviewPrompt(evt.Prompt, pr),
|
||||
Repositories: repositories,
|
||||
Metadata: map[string]interface{}{
|
||||
@@ -474,6 +474,13 @@ func buildReviewTaskRequest(evt *github.NewReviewPREvent, repositories []ReviewT
|
||||
}
|
||||
}
|
||||
|
||||
// buildIssueTaskTitle builds the "Issue #<n>: <title>" task title for an issue
|
||||
// watcher, shortened to the task title limit so an over-long remote issue title
|
||||
// does not cause CreateIssueTask to reject the whole task.
|
||||
func buildIssueTaskTitle(number int, title string) string {
|
||||
return service.TruncateTaskTitle(fmt.Sprintf("Issue #%d: %s", number, title))
|
||||
}
|
||||
|
||||
// shouldAutoStartStep checks if the workflow step has the OnEnterAutoStartAgent action.
|
||||
func (s *Service) shouldAutoStartStep(ctx context.Context, stepID string) bool {
|
||||
if s.workflowStepGetter == nil || stepID == "" {
|
||||
@@ -1365,7 +1372,7 @@ func (s *Service) createIssueTask(ctx context.Context, evt *github.NewIssueEvent
|
||||
WorkspaceID: evt.WorkspaceID,
|
||||
WorkflowID: evt.WorkflowID,
|
||||
WorkflowStepID: evt.WorkflowStepID,
|
||||
Title: fmt.Sprintf("Issue #%d: %s", issue.Number, issue.Title),
|
||||
Title: buildIssueTaskTitle(issue.Number, issue.Title),
|
||||
Description: interpolateIssuePrompt(evt.Prompt, issue),
|
||||
Metadata: map[string]interface{}{
|
||||
issueWatchIDKey: evt.IssueWatchID,
|
||||
|
||||
@@ -2,18 +2,141 @@ package orchestrator
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
"unicode/utf8"
|
||||
|
||||
"github.com/kandev/kandev/internal/github"
|
||||
"github.com/kandev/kandev/internal/orchestrator/executor"
|
||||
"github.com/kandev/kandev/internal/orchestrator/watcher"
|
||||
taskmodels "github.com/kandev/kandev/internal/task/models"
|
||||
taskservice "github.com/kandev/kandev/internal/task/service"
|
||||
wfmodels "github.com/kandev/kandev/internal/workflow/models"
|
||||
v1 "github.com/kandev/kandev/pkg/api/v1"
|
||||
)
|
||||
|
||||
// TestBuildReviewTaskRequest_TruncatesTitle is the regression test for the
|
||||
// dropped-review-task bug: a PR whose "PR #<n>: <title>" string exceeds the
|
||||
// 60-rune task title limit must be shortened (prefix kept, ≤60 runes, trailing
|
||||
// ellipsis) so CreateReviewTask accepts it instead of rejecting the whole task.
|
||||
func TestBuildReviewTaskRequest_TruncatesTitle(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
prTitle string
|
||||
wantExact string // when non-empty, the full expected title
|
||||
wantTrunc bool
|
||||
}{
|
||||
{
|
||||
name: "short ascii unchanged",
|
||||
prTitle: "Fix login bug",
|
||||
wantExact: "PR #42: Fix login bug",
|
||||
},
|
||||
{
|
||||
name: "long ascii truncated",
|
||||
prTitle: strings.Repeat("a", 100),
|
||||
wantTrunc: true,
|
||||
},
|
||||
{
|
||||
name: "long multibyte not split",
|
||||
prTitle: strings.Repeat("é", 100),
|
||||
wantTrunc: true,
|
||||
},
|
||||
{
|
||||
// Prefix "PR #42: " is 8 runes, so a 52-rune PR title yields exactly
|
||||
// the 60-rune limit and must pass through untouched (boundary case).
|
||||
name: "exactly at limit unchanged",
|
||||
prTitle: strings.Repeat("a", taskservice.TaskTitleMaxLength-len("PR #42: ")),
|
||||
wantExact: "PR #42: " + strings.Repeat("a", taskservice.TaskTitleMaxLength-len("PR #42: ")),
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
evt := newReviewEvent()
|
||||
evt.PR.Title = tt.prTitle
|
||||
req := buildReviewTaskRequest(evt, nil, "acme/widget")
|
||||
|
||||
if got := utf8.RuneCountInString(req.Title); got > taskservice.TaskTitleMaxLength {
|
||||
t.Fatalf("title = %d runes, want ≤ %d", got, taskservice.TaskTitleMaxLength)
|
||||
}
|
||||
if !utf8.ValidString(req.Title) {
|
||||
t.Fatalf("title %q is not valid UTF-8; byte-based truncation split a rune", req.Title)
|
||||
}
|
||||
if err := taskservice.ValidateTaskTitle(req.Title); err != nil {
|
||||
t.Fatalf("ValidateTaskTitle: %v", err)
|
||||
}
|
||||
if !strings.HasPrefix(req.Title, "PR #42:") {
|
||||
t.Errorf("title %q lost the PR prefix", req.Title)
|
||||
}
|
||||
if tt.wantExact != "" && req.Title != tt.wantExact {
|
||||
t.Errorf("title = %q, want %q", req.Title, tt.wantExact)
|
||||
}
|
||||
if tt.wantTrunc && !strings.HasSuffix(req.Title, "…") {
|
||||
t.Errorf("truncated title %q should end with ellipsis", req.Title)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestBuildIssueTaskTitle_TruncatesTitle mirrors the review-title regression for
|
||||
// the GitHub issue watcher's "Issue #<n>: <title>" builder.
|
||||
func TestBuildIssueTaskTitle_TruncatesTitle(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
issueTitle string
|
||||
wantExact string
|
||||
wantTrunc bool
|
||||
}{
|
||||
{
|
||||
name: "short ascii unchanged",
|
||||
issueTitle: "Broken export",
|
||||
wantExact: "Issue #7: Broken export",
|
||||
},
|
||||
{
|
||||
name: "long ascii truncated",
|
||||
issueTitle: strings.Repeat("b", 100),
|
||||
wantTrunc: true,
|
||||
},
|
||||
{
|
||||
name: "long multibyte not split",
|
||||
issueTitle: strings.Repeat("ü", 100),
|
||||
wantTrunc: true,
|
||||
},
|
||||
{
|
||||
// Prefix "Issue #7: " is 10 runes, so a 50-rune issue title yields
|
||||
// exactly the 60-rune limit and must pass through untouched.
|
||||
name: "exactly at limit unchanged",
|
||||
issueTitle: strings.Repeat("b", taskservice.TaskTitleMaxLength-len("Issue #7: ")),
|
||||
wantExact: "Issue #7: " + strings.Repeat("b", taskservice.TaskTitleMaxLength-len("Issue #7: ")),
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
got := buildIssueTaskTitle(7, tt.issueTitle)
|
||||
|
||||
if n := utf8.RuneCountInString(got); n > taskservice.TaskTitleMaxLength {
|
||||
t.Fatalf("title = %d runes, want ≤ %d", n, taskservice.TaskTitleMaxLength)
|
||||
}
|
||||
if !utf8.ValidString(got) {
|
||||
t.Fatalf("title %q is not valid UTF-8; byte-based truncation split a rune", got)
|
||||
}
|
||||
if err := taskservice.ValidateTaskTitle(got); err != nil {
|
||||
t.Fatalf("ValidateTaskTitle: %v", err)
|
||||
}
|
||||
if !strings.HasPrefix(got, "Issue #7:") {
|
||||
t.Errorf("title %q lost the Issue prefix", got)
|
||||
}
|
||||
if tt.wantExact != "" && got != tt.wantExact {
|
||||
t.Errorf("title = %q, want %q", got, tt.wantExact)
|
||||
}
|
||||
if tt.wantTrunc && !strings.HasSuffix(got, "…") {
|
||||
t.Errorf("truncated title %q should end with ellipsis", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// countingReviewTaskCreator records how many times CreateReviewTask was called.
|
||||
type countingReviewTaskCreator struct {
|
||||
calls int
|
||||
|
||||
@@ -21,6 +21,14 @@ import (
|
||||
func (e *Executor) Stop(ctx context.Context, sessionID string, reason string, force bool) error {
|
||||
session, err := e.repo.GetTaskSession(ctx, sessionID)
|
||||
if err != nil {
|
||||
// A genuinely missing session row means the runtime this stop would
|
||||
// target is gone; normalize to the public not-found sentinel so durable
|
||||
// cleanup can treat a confirmed-dead local runtime as already stopped.
|
||||
// An unrelated store error is NOT reclassified — it stays a plain
|
||||
// ErrExecutionNotFound so a transient failure remains retryable.
|
||||
if errors.Is(err, models.ErrTaskSessionNotFound) {
|
||||
return fmt.Errorf("%w: %w: %w", ErrExecutionNotFound, runtimeapi.ErrNotFound, err)
|
||||
}
|
||||
return ErrExecutionNotFound
|
||||
}
|
||||
return e.stopWithSession(ctx, session, reason, force)
|
||||
|
||||
@@ -108,6 +108,46 @@ func TestStopSessionDetailed_ClassifiesExecutionLookup(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestStop_MissingSessionReportsRuntimeNotFound covers the boundary contract the
|
||||
// durable task cleanup depends on: when the task_session row is genuinely gone,
|
||||
// Stop must report the runtime as not-found (wrapping the public
|
||||
// runtimeapi.ErrNotFound sentinel) so cleanup can classify a confirmed-dead
|
||||
// local runtime as already-stopped. An unrelated store error must NOT be
|
||||
// collapsed into a not-found result.
|
||||
func TestStop_MissingSessionReportsRuntimeNotFound(t *testing.T) {
|
||||
t.Run("missing session row normalizes to runtime not-found", func(t *testing.T) {
|
||||
repo := newMockRepository()
|
||||
repo.getTaskSessionFunc = func(context.Context, string) (*models.TaskSession, error) {
|
||||
return nil, fmt.Errorf("agent session not found: %w", models.ErrTaskSessionNotFound)
|
||||
}
|
||||
exec := newTestExecutor(t, &mockAgentManager{}, repo)
|
||||
|
||||
err := exec.Stop(context.Background(), "sess-gone", "cleanup", true)
|
||||
if !errors.Is(err, ErrExecutionNotFound) {
|
||||
t.Fatalf("error = %v, want ErrExecutionNotFound", err)
|
||||
}
|
||||
if !errors.Is(err, runtimeapi.ErrNotFound) {
|
||||
t.Fatalf("missing session must normalize to runtimeapi.ErrNotFound; got %v", err)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("unrelated store error is not collapsed to not-found", func(t *testing.T) {
|
||||
repo := newMockRepository()
|
||||
repo.getTaskSessionFunc = func(context.Context, string) (*models.TaskSession, error) {
|
||||
return nil, errors.New("database is unavailable")
|
||||
}
|
||||
exec := newTestExecutor(t, &mockAgentManager{}, repo)
|
||||
|
||||
err := exec.Stop(context.Background(), "sess-x", "cleanup", true)
|
||||
if !errors.Is(err, ErrExecutionNotFound) {
|
||||
t.Fatalf("error = %v, want ErrExecutionNotFound", err)
|
||||
}
|
||||
if errors.Is(err, runtimeapi.ErrNotFound) {
|
||||
t.Fatalf("an unrelated store error must not be reclassified as runtime not-found; got %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestStopSessionDetailed_RejectsInvalidSession(t *testing.T) {
|
||||
exec := newTestExecutor(t, &mockAgentManager{}, newMockRepository())
|
||||
|
||||
|
||||
@@ -2,13 +2,20 @@ package orchestrator
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
runtimeapi "github.com/kandev/kandev/internal/agent/runtime"
|
||||
"github.com/kandev/kandev/internal/agentruntime"
|
||||
"github.com/kandev/kandev/internal/task/models"
|
||||
)
|
||||
|
||||
// errWorkTransient is a non-not-found stop error used to assert that a transient
|
||||
// stop failure never causes a confirmed-dead row to be pruned.
|
||||
var errWorkTransient = errors.New("runtime stop transient failure")
|
||||
|
||||
// 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,
|
||||
@@ -185,6 +192,145 @@ func TestReconcileSessionsOnStartup_MissingSessionStopsAgentAndDeletesRow(t *tes
|
||||
}
|
||||
}
|
||||
|
||||
// TestReconcileSessionsOnStartup_MissingSessionDeadRowPrunedOnNotFound is the
|
||||
// regression test for the stale-orphan-row bug: a missing-session row whose
|
||||
// runtime handle is already gone returns a not-found error from the stop, which
|
||||
// the old code treated as "stop failed" and preserved forever. A confirmed-dead
|
||||
// LOCAL row must instead treat the not-found stop as already-stopped and prune
|
||||
// the row under the resume-safety invariant (no resume_token → delete).
|
||||
func TestReconcileSessionsOnStartup_MissingSessionDeadRowPrunedOnNotFound(t *testing.T) {
|
||||
repo := setupTestRepo(t)
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC()
|
||||
|
||||
if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{
|
||||
ID: "sDead", SessionID: "sDead", TaskID: "taskDead", Runtime: agentruntime.RuntimeStandalone,
|
||||
Status: models.ExecutorRunningStatusRunning, AgentExecutionID: "execDead", LocalPID: 4242,
|
||||
CreatedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("upsert sDead: %v", err)
|
||||
}
|
||||
|
||||
agentMgr := &mockAgentManager{
|
||||
// The runtime seam (lifecycleAdapter) normalizes the lifecycle not-found
|
||||
// sentinel to runtimeapi.ErrNotFound before it reaches the orchestrator.
|
||||
stopAgentWithReasonErr: fmt.Errorf("stop agent: %w", runtimeapi.ErrNotFound),
|
||||
rowLivenessFn: func(*models.ExecutorRunning) models.ProcessLiveness {
|
||||
return models.ProcessLivenessDead
|
||||
},
|
||||
}
|
||||
svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
|
||||
svc.reconcileSessionsOnStartup(ctx)
|
||||
|
||||
if _, err := repo.GetExecutorRunningBySessionID(ctx, "sDead"); err == nil {
|
||||
t.Error("confirmed-dead local orphan row with a not-found stop must be pruned, not preserved")
|
||||
}
|
||||
}
|
||||
|
||||
// TestReconcileSessionsOnStartup_MissingSessionDeadRowWithTokenRepaired covers
|
||||
// the resume-safety half of the confirmed-dead not-found path: a dead local row
|
||||
// that still carries a resume_token is repaired in place (never deleted).
|
||||
func TestReconcileSessionsOnStartup_MissingSessionDeadRowWithTokenRepaired(t *testing.T) {
|
||||
repo := setupTestRepo(t)
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC()
|
||||
|
||||
if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{
|
||||
ID: "sTok", SessionID: "sTok", TaskID: "taskTok", Runtime: agentruntime.RuntimeStandalone,
|
||||
Status: models.ExecutorRunningStatusRunning, AgentExecutionID: "execTok",
|
||||
ResumeToken: "tokTok", Resumable: true, LocalPID: 5252,
|
||||
CreatedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("upsert sTok: %v", err)
|
||||
}
|
||||
|
||||
agentMgr := &mockAgentManager{
|
||||
stopAgentWithReasonErr: fmt.Errorf("stop agent: %w", runtimeapi.ErrNotFound),
|
||||
rowLivenessFn: func(*models.ExecutorRunning) models.ProcessLiveness {
|
||||
return models.ProcessLivenessDead
|
||||
},
|
||||
}
|
||||
svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
|
||||
svc.reconcileSessionsOnStartup(ctx)
|
||||
|
||||
row, err := repo.GetExecutorRunningBySessionID(ctx, "sTok")
|
||||
if err != nil {
|
||||
t.Fatalf("confirmed-dead row holding a resume_token must be preserved: %v", err)
|
||||
}
|
||||
if row.ResumeToken != "tokTok" {
|
||||
t.Errorf("resume_token must survive repair; got %q", row.ResumeToken)
|
||||
}
|
||||
if row.Status != models.ExecutorRunningStatusStopped || row.LocalPID != 0 {
|
||||
t.Errorf("dead row must be repaired to stopped with cleared local_pid; got status=%q local_pid=%d", row.Status, row.LocalPID)
|
||||
}
|
||||
}
|
||||
|
||||
// TestReconcileSessionsOnStartup_MissingSessionUnknownRowPreservedOnNotFound
|
||||
// guards the anti-blanket-ignore rule: a not-found stop for a row whose local
|
||||
// liveness is Unknown (remote/containerized/no local handle) must NOT prune the
|
||||
// row — only a confirmed-dead LOCAL row is reclassified.
|
||||
func TestReconcileSessionsOnStartup_MissingSessionUnknownRowPreservedOnNotFound(t *testing.T) {
|
||||
repo := setupTestRepo(t)
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC()
|
||||
|
||||
if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{
|
||||
ID: "sRemote", SessionID: "sRemote", TaskID: "taskRemote", Runtime: agentruntime.RuntimeSSH,
|
||||
Status: models.ExecutorRunningStatusRunning, AgentExecutionID: "execRemote",
|
||||
ResumeToken: "tokRemote", PID: 4444, CreatedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("upsert sRemote: %v", err)
|
||||
}
|
||||
|
||||
agentMgr := &mockAgentManager{
|
||||
stopAgentWithReasonErr: fmt.Errorf("stop agent: %w", runtimeapi.ErrNotFound),
|
||||
rowLivenessFn: func(*models.ExecutorRunning) models.ProcessLiveness {
|
||||
return models.ProcessLivenessUnknown
|
||||
},
|
||||
}
|
||||
svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
|
||||
svc.reconcileSessionsOnStartup(ctx)
|
||||
|
||||
row, err := repo.GetExecutorRunningBySessionID(ctx, "sRemote")
|
||||
if err != nil {
|
||||
t.Fatalf("unknown/remote row must be preserved on a host-local not-found: %v", err)
|
||||
}
|
||||
if row.ResumeToken != "tokRemote" || row.PID != 4444 {
|
||||
t.Errorf("unknown/remote row must be left intact; got token=%q pid=%d", row.ResumeToken, row.PID)
|
||||
}
|
||||
}
|
||||
|
||||
// TestReconcileSessionsOnStartup_MissingSessionNonNotFoundPreserved guards the
|
||||
// retryable-failure rule: a stop that fails with a non-not-found error must
|
||||
// preserve the row even when the row is confirmed dead — a transient stop
|
||||
// failure is never mistaken for an absent runtime.
|
||||
func TestReconcileSessionsOnStartup_MissingSessionNonNotFoundPreserved(t *testing.T) {
|
||||
repo := setupTestRepo(t)
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC()
|
||||
|
||||
if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{
|
||||
ID: "sErr", SessionID: "sErr", TaskID: "taskErr", Runtime: agentruntime.RuntimeStandalone,
|
||||
Status: models.ExecutorRunningStatusRunning, AgentExecutionID: "execErr", LocalPID: 6262,
|
||||
CreatedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("upsert sErr: %v", err)
|
||||
}
|
||||
|
||||
agentMgr := &mockAgentManager{
|
||||
stopAgentWithReasonErr: errWorkTransient, // not a not-found sentinel
|
||||
rowLivenessFn: func(*models.ExecutorRunning) models.ProcessLiveness {
|
||||
return models.ProcessLivenessDead
|
||||
},
|
||||
}
|
||||
svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
|
||||
svc.reconcileSessionsOnStartup(ctx)
|
||||
|
||||
if _, err := repo.GetExecutorRunningBySessionID(ctx, "sErr"); err != nil {
|
||||
t.Errorf("a non-not-found stop failure must preserve the row for retry: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
@@ -227,3 +373,179 @@ func TestReconcileSessionsOnStartup_CreatedSessionRowPrunedUnlessResumable(t *te
|
||||
t.Errorf("sF should be repaired to stopped with cleared local_pid; got status=%q local_pid=%d", f.Status, f.LocalPID)
|
||||
}
|
||||
}
|
||||
|
||||
// TestReconcileSessionsOnStartup_TerminalDeadRowPrunedOnNotFound is the
|
||||
// regression test for the "failed to stop terminal session runtime; preserving
|
||||
// executor record" warning: a COMPLETED/CANCELLED session whose confirmed-dead
|
||||
// LOCAL runtime returns a not-found stop must be treated as already-stopped and
|
||||
// its tokenless row pruned — not preserved forever. Mirrors the missing-session
|
||||
// classification so terminal reconciliation stops leaking stale rows.
|
||||
func TestReconcileSessionsOnStartup_TerminalDeadRowPrunedOnNotFound(t *testing.T) {
|
||||
for _, state := range []models.TaskSessionState{
|
||||
models.TaskSessionStateCompleted,
|
||||
models.TaskSessionStateCancelled,
|
||||
} {
|
||||
t.Run(string(state), func(t *testing.T) {
|
||||
repo := setupTestRepo(t)
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC()
|
||||
|
||||
seedTaskAndSession(t, repo, "taskT", "sT", state)
|
||||
if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{
|
||||
ID: "sT", SessionID: "sT", TaskID: "taskT", Runtime: agentruntime.RuntimeStandalone,
|
||||
Status: models.ExecutorRunningStatusRunning, AgentExecutionID: "execT", LocalPID: 7272,
|
||||
CreatedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("upsert sT: %v", err)
|
||||
}
|
||||
|
||||
agentMgr := &mockAgentManager{
|
||||
stopAgentWithReasonErr: fmt.Errorf("stop agent: %w", runtimeapi.ErrNotFound),
|
||||
rowLivenessFn: func(*models.ExecutorRunning) models.ProcessLiveness {
|
||||
return models.ProcessLivenessDead
|
||||
},
|
||||
}
|
||||
svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
|
||||
svc.reconcileSessionsOnStartup(ctx)
|
||||
|
||||
if _, err := repo.GetExecutorRunningBySessionID(ctx, "sT"); err == nil {
|
||||
t.Error("confirmed-dead terminal row with a not-found stop must be pruned, not preserved")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestReconcileSessionsOnStartup_TerminalDeadRowWithTokenRepaired covers the
|
||||
// resume-safety half of the terminal not-found path: a terminal row still
|
||||
// carrying a resume_token is repaired in place, never deleted.
|
||||
func TestReconcileSessionsOnStartup_TerminalDeadRowWithTokenRepaired(t *testing.T) {
|
||||
repo := setupTestRepo(t)
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC()
|
||||
|
||||
seedTaskAndSession(t, repo, "taskTT", "sTT", models.TaskSessionStateCompleted)
|
||||
if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{
|
||||
ID: "sTT", SessionID: "sTT", TaskID: "taskTT", Runtime: agentruntime.RuntimeStandalone,
|
||||
Status: models.ExecutorRunningStatusRunning, AgentExecutionID: "execTT",
|
||||
ResumeToken: "tokTT", Resumable: true, LocalPID: 8282,
|
||||
CreatedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("upsert sTT: %v", err)
|
||||
}
|
||||
|
||||
agentMgr := &mockAgentManager{
|
||||
stopAgentWithReasonErr: fmt.Errorf("stop agent: %w", runtimeapi.ErrNotFound),
|
||||
rowLivenessFn: func(*models.ExecutorRunning) models.ProcessLiveness {
|
||||
return models.ProcessLivenessDead
|
||||
},
|
||||
}
|
||||
svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
|
||||
svc.reconcileSessionsOnStartup(ctx)
|
||||
|
||||
row, err := repo.GetExecutorRunningBySessionID(ctx, "sTT")
|
||||
if err != nil {
|
||||
t.Fatalf("terminal row holding a resume_token must be preserved: %v", err)
|
||||
}
|
||||
if row.ResumeToken != "tokTT" {
|
||||
t.Errorf("resume_token must survive repair; got %q", row.ResumeToken)
|
||||
}
|
||||
if row.Status != models.ExecutorRunningStatusStopped || row.LocalPID != 0 {
|
||||
t.Errorf("dead terminal row must be repaired to stopped with cleared local_pid; got status=%q local_pid=%d", row.Status, row.LocalPID)
|
||||
}
|
||||
}
|
||||
|
||||
// TestReconcileSessionsOnStartup_TerminalUnknownRowPreservedOnNotFound guards the
|
||||
// anti-blanket-ignore rule for the terminal path: a not-found stop for a row
|
||||
// whose local liveness is Unknown (remote/no local handle) must NOT prune the
|
||||
// row — only a confirmed-dead LOCAL row is reclassified.
|
||||
func TestReconcileSessionsOnStartup_TerminalUnknownRowPreservedOnNotFound(t *testing.T) {
|
||||
repo := setupTestRepo(t)
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC()
|
||||
|
||||
seedTaskAndSession(t, repo, "taskTU", "sTU", models.TaskSessionStateCancelled)
|
||||
if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{
|
||||
ID: "sTU", SessionID: "sTU", TaskID: "taskTU", Runtime: agentruntime.RuntimeSSH,
|
||||
Status: models.ExecutorRunningStatusRunning, AgentExecutionID: "execTU",
|
||||
PID: 9191, CreatedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("upsert sTU: %v", err)
|
||||
}
|
||||
|
||||
agentMgr := &mockAgentManager{
|
||||
stopAgentWithReasonErr: fmt.Errorf("stop agent: %w", runtimeapi.ErrNotFound),
|
||||
rowLivenessFn: func(*models.ExecutorRunning) models.ProcessLiveness {
|
||||
return models.ProcessLivenessUnknown
|
||||
},
|
||||
}
|
||||
svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
|
||||
svc.reconcileSessionsOnStartup(ctx)
|
||||
|
||||
if _, err := repo.GetExecutorRunningBySessionID(ctx, "sTU"); err != nil {
|
||||
t.Fatalf("unknown/remote terminal row must be preserved on a host-local not-found: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestReconcileSessionsOnStartup_FailedNonResumableDeadRowPrunedOnNotFound
|
||||
// extends the classification to the non-resumable FAILED startup path: a failed
|
||||
// session with no resume handle whose dead-local runtime returns not-found must
|
||||
// have its row pruned rather than preserved with the "failed to stop failed
|
||||
// session runtime; preserving executor record" warning.
|
||||
func TestReconcileSessionsOnStartup_FailedNonResumableDeadRowPrunedOnNotFound(t *testing.T) {
|
||||
repo := setupTestRepo(t)
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC()
|
||||
|
||||
seedTaskAndSession(t, repo, "taskFail", "sFail", models.TaskSessionStateFailed)
|
||||
if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{
|
||||
ID: "sFail", SessionID: "sFail", TaskID: "taskFail", Runtime: agentruntime.RuntimeStandalone,
|
||||
Status: models.ExecutorRunningStatusRunning, AgentExecutionID: "execFail", LocalPID: 3131,
|
||||
CreatedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("upsert sFail: %v", err)
|
||||
}
|
||||
|
||||
agentMgr := &mockAgentManager{
|
||||
stopAgentWithReasonErr: fmt.Errorf("stop agent: %w", runtimeapi.ErrNotFound),
|
||||
rowLivenessFn: func(*models.ExecutorRunning) models.ProcessLiveness {
|
||||
return models.ProcessLivenessDead
|
||||
},
|
||||
}
|
||||
svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
|
||||
svc.reconcileSessionsOnStartup(ctx)
|
||||
|
||||
if _, err := repo.GetExecutorRunningBySessionID(ctx, "sFail"); err == nil {
|
||||
t.Error("confirmed-dead non-resumable failed row with a not-found stop must be pruned, not preserved")
|
||||
}
|
||||
}
|
||||
|
||||
// TestReconcileSessionsOnStartup_TerminalNonNotFoundPreserved guards the
|
||||
// retryable-failure rule for the terminal path: a stop that fails with a
|
||||
// non-not-found error must preserve the row even when the row is confirmed dead.
|
||||
func TestReconcileSessionsOnStartup_TerminalNonNotFoundPreserved(t *testing.T) {
|
||||
repo := setupTestRepo(t)
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC()
|
||||
|
||||
seedTaskAndSession(t, repo, "taskTE", "sTE", models.TaskSessionStateCompleted)
|
||||
if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{
|
||||
ID: "sTE", SessionID: "sTE", TaskID: "taskTE", Runtime: agentruntime.RuntimeStandalone,
|
||||
Status: models.ExecutorRunningStatusRunning, AgentExecutionID: "execTE", LocalPID: 4141,
|
||||
CreatedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("upsert sTE: %v", err)
|
||||
}
|
||||
|
||||
agentMgr := &mockAgentManager{
|
||||
stopAgentWithReasonErr: errWorkTransient, // not a not-found sentinel
|
||||
rowLivenessFn: func(*models.ExecutorRunning) models.ProcessLiveness {
|
||||
return models.ProcessLivenessDead
|
||||
},
|
||||
}
|
||||
svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
|
||||
svc.reconcileSessionsOnStartup(ctx)
|
||||
|
||||
if _, err := repo.GetExecutorRunningBySessionID(ctx, "sTE"); err != nil {
|
||||
t.Errorf("a non-not-found stop failure must preserve the terminal row for retry: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -22,6 +22,7 @@ import (
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
||||
runtimeapi "github.com/kandev/kandev/internal/agent/runtime"
|
||||
"github.com/kandev/kandev/internal/agent/runtime/routingerr"
|
||||
"github.com/kandev/kandev/internal/agentctl/types/streams"
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
@@ -1740,11 +1741,25 @@ func (s *Service) handleMissingSessionOnStartup(ctx context.Context, running *mo
|
||||
return
|
||||
}
|
||||
if err := s.agentManager.StopAgentWithReason(ctx, executionID, "startup missing session cleanup", true); err != nil {
|
||||
s.logger.Warn("failed to stop missing-session runtime; preserving executor record",
|
||||
// A not-found stop for a runtime whose row is a confirmed-dead LOCAL
|
||||
// process means the runtime is already gone — treat it as an idempotent
|
||||
// successful stop and prune/repair the row under the resume-safety
|
||||
// invariant, so the orphan row does not survive restarts forever. Any
|
||||
// other error (alive/unknown/remote row, or a non-not-found failure) is
|
||||
// preserved and left for a later attempt.
|
||||
if !stopReportsRuntimeAbsent(err) || s.rowLiveness(running) != models.ProcessLivenessDead {
|
||||
s.logger.Warn("failed to stop missing-session runtime; preserving executor record",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("task_id", running.TaskID),
|
||||
zap.String("execution_id", executionID),
|
||||
zap.Error(err))
|
||||
return
|
||||
}
|
||||
s.logger.Info("missing-session runtime already gone (confirmed dead); pruning executor record",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("task_id", running.TaskID),
|
||||
zap.String("execution_id", executionID),
|
||||
zap.Error(err))
|
||||
zap.String("execution_id", executionID))
|
||||
s.pruneOrRepairExecutorRow(ctx, running, models.TaskSessionStateCancelled)
|
||||
return
|
||||
}
|
||||
if err := s.repo.DeleteExecutorRunningBySessionID(ctx, sessionID); err != nil {
|
||||
@@ -1755,6 +1770,17 @@ func (s *Service) handleMissingSessionOnStartup(ctx context.Context, running *mo
|
||||
}
|
||||
}
|
||||
|
||||
// stopReportsRuntimeAbsent reports whether a stop error means the runtime is
|
||||
// already gone (a typed not-found sentinel), as opposed to a transient failure.
|
||||
// Only a typed sentinel counts — a generic store/lookup error must never be
|
||||
// reinterpreted as an absent runtime. The runtime seam (lifecycleAdapter)
|
||||
// normalizes the backend lifecycle not-found sentinel to runtimeapi.ErrNotFound,
|
||||
// so reconciliation depends only on the public runtime and executor sentinels.
|
||||
func stopReportsRuntimeAbsent(err error) bool {
|
||||
return errors.Is(err, runtimeapi.ErrNotFound) ||
|
||||
errors.Is(err, executor.ErrExecutionNotFound)
|
||||
}
|
||||
|
||||
func (s *Service) abandonOpenTurnsOnStartup(ctx context.Context, sessionID, reason string) {
|
||||
if s.turnService == nil {
|
||||
return
|
||||
@@ -1783,15 +1809,8 @@ func (s *Service) handleTerminalSessionOnStartup(ctx context.Context, session *m
|
||||
zap.String("task_id", session.TaskID),
|
||||
zap.String("state", string(previousState)))
|
||||
s.abandonOpenTurnsOnStartup(ctx, sessionID, "terminal session cleanup")
|
||||
executionID := strings.TrimSpace(running.AgentExecutionID)
|
||||
if executionID != "" && s.agentManager != nil {
|
||||
if err := s.agentManager.StopAgentWithReason(ctx, executionID, "startup terminal session cleanup", true); err != nil {
|
||||
s.logger.Warn("failed to stop terminal session runtime; preserving executor record",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("execution_id", executionID),
|
||||
zap.Error(err))
|
||||
return true
|
||||
}
|
||||
if !s.stopRuntimeForStartupCleanup(ctx, running, "startup terminal session cleanup") {
|
||||
return true
|
||||
}
|
||||
// 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
|
||||
@@ -1835,15 +1854,8 @@ func (s *Service) handleFailedSessionOnStartup(ctx context.Context, session *mod
|
||||
s.logger.Info("stopping failed session runtime before cleaning up executor record",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("task_id", session.TaskID))
|
||||
executionID := strings.TrimSpace(running.AgentExecutionID)
|
||||
if executionID != "" && s.agentManager != nil {
|
||||
if err := s.agentManager.StopAgentWithReason(ctx, executionID, "startup failed session cleanup", true); err != nil {
|
||||
s.logger.Warn("failed to stop failed session runtime; preserving executor record",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("execution_id", executionID),
|
||||
zap.Error(err))
|
||||
return
|
||||
}
|
||||
if !s.stopRuntimeForStartupCleanup(ctx, running, "startup failed session cleanup") {
|
||||
return
|
||||
}
|
||||
// Prune only subject to the resume-safety invariant (a lingering
|
||||
// resume_token is repaired in place rather than deleted).
|
||||
@@ -1851,6 +1863,40 @@ func (s *Service) handleFailedSessionOnStartup(ctx context.Context, session *mod
|
||||
}
|
||||
}
|
||||
|
||||
// stopRuntimeForStartupCleanup stops a session's still-registered runtime handle
|
||||
// during startup reconciliation and reports whether the caller may proceed to
|
||||
// prune/repair the executor row. It mirrors handleMissingSessionOnStartup's
|
||||
// classification so terminal and non-resumable-failed reconciliation stop
|
||||
// leaking stale rows: a not-found stop for a confirmed-dead LOCAL runtime means
|
||||
// the runtime is already gone and cleanup proceeds; any other error
|
||||
// (alive/unknown/remote row, or a non-not-found failure) preserves the row for a
|
||||
// later attempt. A row with no stoppable handle is already stoppable-complete.
|
||||
func (s *Service) stopRuntimeForStartupCleanup(ctx context.Context, running *models.ExecutorRunning, reason string) bool {
|
||||
executionID := strings.TrimSpace(running.AgentExecutionID)
|
||||
if executionID == "" || s.agentManager == nil {
|
||||
return true
|
||||
}
|
||||
err := s.agentManager.StopAgentWithReason(ctx, executionID, reason, true)
|
||||
if err == nil {
|
||||
return true
|
||||
}
|
||||
if !stopReportsRuntimeAbsent(err) || s.rowLiveness(running) != models.ProcessLivenessDead {
|
||||
s.logger.Warn("failed to stop session runtime during startup cleanup; preserving executor record",
|
||||
zap.String("session_id", running.SessionID),
|
||||
zap.String("task_id", running.TaskID),
|
||||
zap.String("execution_id", executionID),
|
||||
zap.String("reason", reason),
|
||||
zap.Error(err))
|
||||
return false
|
||||
}
|
||||
s.logger.Info("session runtime already gone (confirmed dead) during startup cleanup; proceeding to prune/repair",
|
||||
zap.String("session_id", running.SessionID),
|
||||
zap.String("task_id", running.TaskID),
|
||||
zap.String("execution_id", executionID),
|
||||
zap.String("reason", reason))
|
||||
return true
|
||||
}
|
||||
|
||||
func canResumeRunning(running *models.ExecutorRunning) bool {
|
||||
if running == nil || running.ResumeToken == "" {
|
||||
return false
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
"github.com/kandev/kandev/internal/task/service"
|
||||
workflowmodels "github.com/kandev/kandev/internal/workflow/models"
|
||||
)
|
||||
|
||||
@@ -349,6 +350,16 @@ func (c *WatcherDispatchCoordinator) Dispatch(ctx context.Context, src WatcherSo
|
||||
return
|
||||
}
|
||||
|
||||
// Bound the generated title once, here, so every watcher source (Linear,
|
||||
// Jira, GitLab, future) inherits the task-title limit without each
|
||||
// BuildTaskRequest re-implementing truncation. A remote issue title that
|
||||
// would otherwise exceed the limit is shortened rather than rejected by
|
||||
// backend title validation. GitHub's review/issue path builds its tasks
|
||||
// through a separate creator and truncates at its own builders.
|
||||
if req != nil {
|
||||
req.Title = service.TruncateTaskTitle(req.Title)
|
||||
}
|
||||
|
||||
// A bound repository deleted after the watch was configured would let
|
||||
// CreateTask insert a task row before repository association fails. Self-heal
|
||||
// and release the reservation instead of creating an orphan task.
|
||||
|
||||
@@ -3,12 +3,15 @@ package orchestrator
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
"unicode/utf8"
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
"github.com/kandev/kandev/internal/task/models"
|
||||
"github.com/kandev/kandev/internal/task/service"
|
||||
workflowmodels "github.com/kandev/kandev/internal/workflow/models"
|
||||
)
|
||||
|
||||
@@ -182,6 +185,68 @@ func TestCoordinator_Dispatch_HappyPath(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestCoordinator_Dispatch_TruncatesOverlongTitle pins the watcher-wide title
|
||||
// guarantee: any WatcherSource (Linear, Jira, GitLab, future) whose
|
||||
// BuildTaskRequest returns an overlong Title has that title bounded to the task
|
||||
// title limit by the coordinator before CreateIssueTask, so a long remote issue
|
||||
// title is never rejected by backend title validation.
|
||||
func TestCoordinator_Dispatch_TruncatesOverlongTitle(t *testing.T) {
|
||||
longTitle := "[ENG-1] " + strings.Repeat("x", 200)
|
||||
src := &fakeWatcherSource{
|
||||
name: "linear",
|
||||
reserveOK: true,
|
||||
buildReq: &IssueTaskRequest{
|
||||
WorkspaceID: "ws-1",
|
||||
WorkflowStepID: "step-1",
|
||||
Title: longTitle,
|
||||
Description: "body",
|
||||
},
|
||||
}
|
||||
tc := &fakeTaskCreator{}
|
||||
starter := &fakeTaskStarter{}
|
||||
c := newTestCoordinator(t, tc, false, starter)
|
||||
|
||||
c.Dispatch(context.Background(), src, "evt")
|
||||
|
||||
if tc.gotReq == nil {
|
||||
t.Fatal("expected task creator to receive the built request")
|
||||
}
|
||||
got := tc.gotReq.Title
|
||||
if runes := utf8.RuneCountInString(got); runes > service.TaskTitleMaxLength {
|
||||
t.Fatalf("title not truncated: %d runes (limit %d): %q", runes, service.TaskTitleMaxLength, got)
|
||||
}
|
||||
if !utf8.ValidString(got) {
|
||||
t.Fatalf("truncated title is not valid UTF-8: %q", got)
|
||||
}
|
||||
if !strings.HasPrefix(got, "[ENG-1] ") {
|
||||
t.Fatalf("truncation dropped the source prefix: %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCoordinator_Dispatch_KeepsShortTitleUnchanged confirms the truncation is a
|
||||
// no-op for a title already within the limit — the coordinator does not append an
|
||||
// ellipsis or otherwise rewrite a compliant title.
|
||||
func TestCoordinator_Dispatch_KeepsShortTitleUnchanged(t *testing.T) {
|
||||
src := &fakeWatcherSource{
|
||||
name: "jira",
|
||||
reserveOK: true,
|
||||
buildReq: &IssueTaskRequest{
|
||||
WorkspaceID: "ws-1",
|
||||
WorkflowStepID: "step-1",
|
||||
Title: "[PROJ-7] short enough",
|
||||
Description: "body",
|
||||
},
|
||||
}
|
||||
tc := &fakeTaskCreator{}
|
||||
c := newTestCoordinator(t, tc, false, &fakeTaskStarter{})
|
||||
|
||||
c.Dispatch(context.Background(), src, "evt")
|
||||
|
||||
if tc.gotReq == nil || tc.gotReq.Title != "[PROJ-7] short enough" {
|
||||
t.Fatalf("expected short title unchanged, got %+v", tc.gotReq)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCoordinator_Dispatch_ReserveLost(t *testing.T) {
|
||||
src := &fakeWatcherSource{name: "linear", reserveOK: false}
|
||||
tc := &fakeTaskCreator{}
|
||||
|
||||
@@ -50,8 +50,8 @@ func TestEnsureWorkspaceClonedWithBasicAuthKeepsCredentialScopedToGitChild(t *te
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
binDir := t.TempDir()
|
||||
capturePath := filepath.Join(t.TempDir(), "git-env")
|
||||
binDir := canonicalTempDir(t)
|
||||
capturePath := filepath.Join(canonicalTempDir(t), "git-env")
|
||||
fakeGit := "#!/bin/sh\nprintf '%s\\n%s' \"$GIT_CONFIG_KEY_1\" \"$GIT_CONFIG_VALUE_1\" > \"$CAPTURE_PATH\"\n" +
|
||||
"if [ \"$BLOCK_GIT\" = 1 ]; then exec sleep 10; fi\nexit 1\n"
|
||||
if err := os.WriteFile(filepath.Join(binDir, "git"), []byte(fakeGit), 0o755); err != nil {
|
||||
@@ -63,11 +63,14 @@ func TestEnsureWorkspaceClonedWithBasicAuthKeepsCredentialScopedToGitChild(t *te
|
||||
if tc.cancel {
|
||||
t.Setenv("BLOCK_GIT", "1")
|
||||
}
|
||||
cloner := NewCloner(Config{BasePath: t.TempDir()}, ProtocolHTTPS, "", logger.Default())
|
||||
cloner := NewCloner(Config{BasePath: canonicalTempDir(t)}, ProtocolHTTPS, "", logger.Default())
|
||||
ctx := context.Background()
|
||||
var cancel context.CancelFunc
|
||||
if tc.cancel {
|
||||
ctx, cancel = context.WithTimeout(ctx, 20*time.Millisecond)
|
||||
// Generous enough for a cold shell start (macOS is slow) so the
|
||||
// fake git reliably writes CAPTURE_PATH before entering its long
|
||||
// sleep; the deadline then cancels the blocked clone mid-run.
|
||||
ctx, cancel = context.WithTimeout(ctx, 750*time.Millisecond)
|
||||
defer cancel()
|
||||
}
|
||||
targetPath, err := cloner.EnsureWorkspaceClonedWithBasicAuth(
|
||||
@@ -102,3 +105,14 @@ func TestEnsureWorkspaceClonedWithBasicAuthKeepsCredentialScopedToGitChild(t *te
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// canonicalTempDir returns t.TempDir() with symlinks resolved so the Git child
|
||||
// and the assertions agree on paths (macOS /var -> /private/var).
|
||||
func canonicalTempDir(t *testing.T) string {
|
||||
t.Helper()
|
||||
resolved, err := filepath.EvalSymlinks(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return resolved
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ package cron
|
||||
|
||||
import (
|
||||
"context"
|
||||
"reflect"
|
||||
"time"
|
||||
|
||||
"go.uber.org/zap"
|
||||
@@ -36,10 +37,18 @@ type RoutinesHandler struct {
|
||||
|
||||
// NewRoutinesHandler builds a RoutinesHandler. now defaults to
|
||||
// time.Now().UTC() when nil — tests pass a controlled clock.
|
||||
//
|
||||
// A typed-nil collaborator (a concrete nil pointer assigned into the
|
||||
// RoutineTicker interface, as happens when Office is disabled) is
|
||||
// normalised to a genuinely nil interface so Tick takes the no-op branch
|
||||
// instead of dereferencing a nil receiver.
|
||||
func NewRoutinesHandler(ticker RoutineTicker, now func() time.Time, log *logger.Logger) *RoutinesHandler {
|
||||
if now == nil {
|
||||
now = func() time.Time { return time.Now().UTC() }
|
||||
}
|
||||
if isNilTicker(ticker) {
|
||||
ticker = nil
|
||||
}
|
||||
return &RoutinesHandler{
|
||||
ticker: ticker,
|
||||
now: now,
|
||||
@@ -47,6 +56,25 @@ func NewRoutinesHandler(ticker RoutineTicker, now func() time.Time, log *logger.
|
||||
}
|
||||
}
|
||||
|
||||
// isNilTicker reports whether ticker is absent, treating a typed-nil
|
||||
// pointer wrapped in the interface the same as a genuinely nil interface.
|
||||
func isNilTicker(ticker RoutineTicker) bool {
|
||||
if ticker == nil {
|
||||
return true
|
||||
}
|
||||
// Every RoutineTicker implementation today is a pointer receiver, so Ptr is
|
||||
// the only kind reached in practice; the remaining nil-able kinds are listed
|
||||
// defensively so a future value/func/channel implementer can never smuggle a
|
||||
// typed-nil past this guard and re-open the Office-disabled tick panic.
|
||||
v := reflect.ValueOf(ticker)
|
||||
switch v.Kind() {
|
||||
case reflect.Ptr, reflect.Map, reflect.Chan, reflect.Func, reflect.Slice, reflect.Interface:
|
||||
return v.IsNil()
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// Name implements Handler.
|
||||
func (h *RoutinesHandler) Name() string { return "routines" }
|
||||
|
||||
|
||||
@@ -54,3 +54,18 @@ func TestRoutinesHandler_NilTickerIsNoOp(t *testing.T) {
|
||||
t.Fatalf("Tick: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRoutinesHandler_TypedNilTickerIsNoOp reproduces the Office-disabled
|
||||
// regression: a concrete nil pointer assigned into the RoutineTicker
|
||||
// interface is a non-nil interface value, so a naive h.ticker == nil guard
|
||||
// is bypassed and Tick dereferences the nil receiver. A handler built from a
|
||||
// typed-nil collaborator by any wiring path must no-op instead of panicking.
|
||||
func TestRoutinesHandler_TypedNilTickerIsNoOp(t *testing.T) {
|
||||
var typedNil *fakeRoutineTicker
|
||||
var ticker RoutineTicker = typedNil // non-nil interface wrapping a nil pointer
|
||||
|
||||
h := NewRoutinesHandler(ticker, nil, logger.Default())
|
||||
if err := h.Tick(context.Background()); err != nil {
|
||||
t.Fatalf("Tick: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -55,7 +55,7 @@ func TestHTTPCreateRepositoryRejectsInvalidLocalPathWithoutPersistence(t *testin
|
||||
|
||||
func TestHTTPInitializeLocalRepositoryCreatesRepository(t *testing.T) {
|
||||
router, repo := newRepositoryHTTPTestRouter(t)
|
||||
parentPath := t.TempDir()
|
||||
parentPath := canonicalTempDir(t)
|
||||
body := []byte(`{"name":"new-project","parent_path":` + strconv.Quote(parentPath) + `}`)
|
||||
request := httptest.NewRequest(
|
||||
http.MethodPost,
|
||||
@@ -316,7 +316,7 @@ func TestHTTPListDirectoryIncludesChoosableContract(t *testing.T) {
|
||||
|
||||
func TestHTTPCreateDirectoryCreatesFolder(t *testing.T) {
|
||||
router, _ := newRepositoryHTTPTestRouter(t)
|
||||
parent := t.TempDir()
|
||||
parent := canonicalTempDir(t)
|
||||
request := httptest.NewRequest(
|
||||
http.MethodPost,
|
||||
"/api/v1/fs/create-dir",
|
||||
|
||||
@@ -1597,8 +1597,8 @@ func (r *freshBranchIdentityRepository) CreateTaskRepository(_ context.Context,
|
||||
|
||||
func TestCommitFreshBranchUsesPersistedTaskRepositoryIdentity(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
realRepoPath := initHandlerGitRepository(t, filepath.Join(t.TempDir(), "real"))
|
||||
decoyRepoPath := initHandlerGitRepository(t, filepath.Join(t.TempDir(), "decoy"))
|
||||
realRepoPath := initHandlerGitRepository(t, filepath.Join(canonicalTempDir(t), "real"))
|
||||
decoyRepoPath := initHandlerGitRepository(t, filepath.Join(canonicalTempDir(t), "decoy"))
|
||||
repo := &freshBranchIdentityRepository{
|
||||
task: &models.Task{ID: "task-1", WorkspaceID: "ws-1"},
|
||||
taskRepos: []*models.TaskRepository{{
|
||||
@@ -1667,6 +1667,16 @@ func TestCommitFreshBranchRollsBackTaskWhenPersistedRepositoriesCannotBeLoaded(t
|
||||
}
|
||||
}
|
||||
|
||||
// canonicalTempDir returns t.TempDir() with symlinks resolved, matching how production canonicalizes local paths.
|
||||
func canonicalTempDir(t *testing.T) string {
|
||||
t.Helper()
|
||||
resolved, err := filepath.EvalSymlinks(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatalf("EvalSymlinks: %v", err)
|
||||
}
|
||||
return resolved
|
||||
}
|
||||
|
||||
func initHandlerGitRepository(t *testing.T, path string) string {
|
||||
t.Helper()
|
||||
if err := os.MkdirAll(path, 0o755); err != nil {
|
||||
|
||||
@@ -310,6 +310,22 @@ func TestLocalRepositoryStatus_ListsDirtyFiles(t *testing.T) {
|
||||
}
|
||||
|
||||
// initRealGitRepo creates a git repo with a single commit on `main`.
|
||||
// canonicalRepoTestPath resolves dir the same way the production
|
||||
// CreateRepository path does (filepath.EvalSymlinks), so a stored LocalPath
|
||||
// matches what resolveRepositoryLocalPath re-resolves it to. Tests that write
|
||||
// LocalPath through the repository-entity layer directly (bypassing the
|
||||
// canonicalizing service.CreateRepository) must use this, or the saved raw
|
||||
// t.TempDir() path fails the sameCanonicalPath guard on platforms whose temp
|
||||
// dir has a symlinked ancestor (e.g. macOS /var -> /private/var).
|
||||
func canonicalRepoTestPath(t *testing.T, dir string) string {
|
||||
t.Helper()
|
||||
resolved, err := filepath.EvalSymlinks(dir)
|
||||
if err != nil {
|
||||
t.Fatalf("EvalSymlinks(%q): %v", dir, err)
|
||||
}
|
||||
return resolved
|
||||
}
|
||||
|
||||
func initRealGitRepo(t *testing.T, dir string) {
|
||||
t.Helper()
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
@@ -388,7 +404,7 @@ func persistFreshBranchRepository(t *testing.T, svc *Service, repoPath string) s
|
||||
t.Fatalf("CreateWorkspace: %v", err)
|
||||
}
|
||||
if err := svc.repoEntities.CreateRepository(ctx, &models.Repository{
|
||||
ID: repositoryID, WorkspaceID: workspaceID, Name: "Repository", SourceType: sourceTypeLocal, LocalPath: repoPath,
|
||||
ID: repositoryID, WorkspaceID: workspaceID, Name: "Repository", SourceType: sourceTypeLocal, LocalPath: canonicalRepoTestPath(t, repoPath),
|
||||
}); err != nil {
|
||||
t.Fatalf("CreateRepository: %v", err)
|
||||
}
|
||||
|
||||
@@ -747,7 +747,7 @@ func TestListBranches_SavedRepositoryOutsideDiscoveryRoots(t *testing.T) {
|
||||
t.Fatalf("CreateWorkspace: %v", err)
|
||||
}
|
||||
if err := svc.repoEntities.CreateRepository(ctx, &models.Repository{
|
||||
ID: "outside-repo", WorkspaceID: "ws-1", Name: "outside", SourceType: sourceTypeLocal, LocalPath: repoPath,
|
||||
ID: "outside-repo", WorkspaceID: "ws-1", Name: "outside", SourceType: sourceTypeLocal, LocalPath: canonicalRepoTestPath(t, repoPath),
|
||||
}); err != nil {
|
||||
t.Fatalf("CreateRepository: %v", err)
|
||||
}
|
||||
@@ -848,7 +848,7 @@ func TestListBranches_SavedLinkedWorktreeOutsideDiscoveryRoots(t *testing.T) {
|
||||
t.Fatalf("CreateWorkspace: %v", err)
|
||||
}
|
||||
if err := svc.repoEntities.CreateRepository(ctx, &models.Repository{
|
||||
ID: "linked-repo", WorkspaceID: "ws-1", Name: "linked", SourceType: sourceTypeLocal, LocalPath: linkedPath,
|
||||
ID: "linked-repo", WorkspaceID: "ws-1", Name: "linked", SourceType: sourceTypeLocal, LocalPath: canonicalRepoTestPath(t, linkedPath),
|
||||
}); err != nil {
|
||||
t.Fatalf("CreateRepository: %v", err)
|
||||
}
|
||||
|
||||
@@ -153,7 +153,7 @@ func TestRefreshRepositoryBranches_SavedRepositoryOutsideDiscoveryRoots(t *testi
|
||||
t.Fatalf("CreateWorkspace: %v", err)
|
||||
}
|
||||
if err := svc.repoEntities.CreateRepository(ctx, &models.Repository{
|
||||
ID: "outside-repo", WorkspaceID: "ws-1", Name: "outside", SourceType: sourceTypeLocal, LocalPath: repoPath,
|
||||
ID: "outside-repo", WorkspaceID: "ws-1", Name: "outside", SourceType: sourceTypeLocal, LocalPath: canonicalRepoTestPath(t, repoPath),
|
||||
}); err != nil {
|
||||
t.Fatalf("CreateRepository: %v", err)
|
||||
}
|
||||
|
||||
@@ -393,12 +393,14 @@ func (s *Service) executeTaskResourceCleanupJob(
|
||||
return fmt.Errorf("refresh task cleanup runtime inventory: %w", err)
|
||||
}
|
||||
s.registerTaskRuntimeStopOwners(targets, true)
|
||||
failedStops := s.stopTaskRuntimeTargets(ctx, job.TaskID, targets, taskResourceCleanupStopReason(job.Trigger), "task cleanup runtime stop failed")
|
||||
stopOutcome := s.stopTaskRuntimeTargets(ctx, job.TaskID, targets, taskResourceCleanupStopReason(job.Trigger), "task cleanup runtime stop failed")
|
||||
failedStops := stopOutcome.failed
|
||||
if cancelled, err := s.cancelIfTaskUnarchived(ctx, job); err != nil || cancelled {
|
||||
return err
|
||||
}
|
||||
errs := s.performTaskCleanup(ctx, job.TaskID, snapshot.Sessions, snapshot.Worktrees, targets,
|
||||
taskEnvironmentCleanup{env: snapshot.TaskEnvironment, deleteRow: snapshot.DeleteEnvironmentRow}, failedStops)
|
||||
taskEnvironmentCleanup{env: snapshot.TaskEnvironment, deleteRow: snapshot.DeleteEnvironmentRow},
|
||||
taskCleanupPreserveRows(stopOutcome))
|
||||
if cause := context.Cause(ctx); cause != nil {
|
||||
return errors.Join(append(errs, cause)...)
|
||||
}
|
||||
|
||||
@@ -53,6 +53,15 @@ type TaskExecutionStopper interface {
|
||||
RegisterExecutionStopOwner(sessionID, executionID string, force bool)
|
||||
}
|
||||
|
||||
// TaskRowLivenessProber classifies an executors_running row's backing-process
|
||||
// liveness in a runtime-aware way (a local process check is never applied to a
|
||||
// remote/SSH row). It is optional and satisfied by the lifecycle adapter. When
|
||||
// unwired, cleanup treats every row as Unknown so a not-found stop is never
|
||||
// mistaken for an absent runtime.
|
||||
type TaskRowLivenessProber interface {
|
||||
RowLiveness(row *models.ExecutorRunning) models.ProcessLiveness
|
||||
}
|
||||
|
||||
// TaskResourceCleanupActivityGate serializes durable cleanup with install-wide maintenance.
|
||||
type TaskResourceCleanupActivityGate interface {
|
||||
AcquireTaskResourceCleanup(context.Context) (TaskResourceCleanupActivityLease, error)
|
||||
@@ -243,6 +252,7 @@ type Service struct {
|
||||
discoveryConfig RepositoryDiscoveryConfig
|
||||
worktreeCleanup WorktreeCleanup
|
||||
executionStopper TaskExecutionStopper
|
||||
rowLivenessProber TaskRowLivenessProber
|
||||
contextWindowResetter func(context.Context, string) error
|
||||
cleanupActivity TaskResourceCleanupActivityGate
|
||||
branchMaterializer BranchMaterializer
|
||||
@@ -371,6 +381,13 @@ func (s *Service) SetExecutionStopper(stopper TaskExecutionStopper) {
|
||||
s.executionStopper = stopper
|
||||
}
|
||||
|
||||
// SetRowLivenessProber wires the runtime-aware executors_running liveness probe
|
||||
// (satisfied by the lifecycle adapter). It is optional; when unwired, cleanup
|
||||
// treats every row as Unknown.
|
||||
func (s *Service) SetRowLivenessProber(prober TaskRowLivenessProber) {
|
||||
s.rowLivenessProber = prober
|
||||
}
|
||||
|
||||
// SetContextWindowResetter wires the guarded context-window reset callback
|
||||
// owned by the orchestrator. It is optional for isolated task-service users;
|
||||
// those callers fall back to clearing the session metadata directly.
|
||||
|
||||
@@ -1980,9 +1980,10 @@ func (s *Service) runTaskCleanup(
|
||||
stopTargets = refreshedTargets
|
||||
s.registerTaskRuntimeStopOwners(stopTargets, true)
|
||||
|
||||
failedStops := s.stopTaskRuntimeTargets(cleanupCtx, id, stopTargets, stopReason, stopFailMsg)
|
||||
stopOutcome := s.stopTaskRuntimeTargets(cleanupCtx, id, stopTargets, stopReason, stopFailMsg)
|
||||
|
||||
cleanupErrors := s.performTaskCleanup(cleanupCtx, id, sessions, worktrees, stopTargets, envCleanup, failedStops)
|
||||
cleanupErrors := s.performTaskCleanup(cleanupCtx, id, sessions, worktrees, stopTargets, envCleanup,
|
||||
taskCleanupPreserveRows(stopOutcome))
|
||||
|
||||
if len(cleanupErrors) > 0 {
|
||||
s.logger.Warn(cleanupMsg+" with errors",
|
||||
@@ -2153,14 +2154,46 @@ func exactTaskStopTargetSessions(targetSets ...[]taskStopTarget) map[string]stru
|
||||
return exact
|
||||
}
|
||||
|
||||
func (s *Service) stopTaskRuntimeTargets(ctx context.Context, taskID string, stopTargets []taskStopTarget, stopReason, stopFailMsg string) map[string]struct{} {
|
||||
failedStops := make(map[string]struct{})
|
||||
// taskRuntimeStopOutcome separates the two independent effects a stop pass has on
|
||||
// downstream cleanup. failed drives durable-job retry accounting AND row
|
||||
// preservation (a still-uncertain stop must be retried and its row kept).
|
||||
// preserve keeps a row (and its worktree) from being torn down WITHOUT counting
|
||||
// as a retryable failure — used for a confirmed-dead but resume-safe row that was
|
||||
// repaired in place under the resume-safety invariant.
|
||||
type taskRuntimeStopOutcome struct {
|
||||
failed map[string]struct{}
|
||||
preserve map[string]struct{}
|
||||
}
|
||||
|
||||
// taskCleanupPreserveRows is the set of sessions whose executor row and worktree
|
||||
// must survive this cleanup pass: every failed (retryable) stop plus every
|
||||
// resume-safe row repaired in place. performTaskCleanup treats membership as
|
||||
// "do not delete this row/worktree".
|
||||
func taskCleanupPreserveRows(outcome taskRuntimeStopOutcome) map[string]struct{} {
|
||||
if len(outcome.preserve) == 0 {
|
||||
return outcome.failed
|
||||
}
|
||||
preserve := make(map[string]struct{}, len(outcome.failed)+len(outcome.preserve))
|
||||
for id := range outcome.failed {
|
||||
preserve[id] = struct{}{}
|
||||
}
|
||||
for id := range outcome.preserve {
|
||||
preserve[id] = struct{}{}
|
||||
}
|
||||
return preserve
|
||||
}
|
||||
|
||||
func (s *Service) stopTaskRuntimeTargets(ctx context.Context, taskID string, stopTargets []taskStopTarget, stopReason, stopFailMsg string) taskRuntimeStopOutcome {
|
||||
outcome := taskRuntimeStopOutcome{
|
||||
failed: make(map[string]struct{}),
|
||||
preserve: make(map[string]struct{}),
|
||||
}
|
||||
if s.executionStopper == nil || len(stopTargets) == 0 {
|
||||
return failedStops
|
||||
return outcome
|
||||
}
|
||||
for _, target := range stopTargets {
|
||||
if context.Cause(ctx) != nil {
|
||||
return failedStops
|
||||
return outcome
|
||||
}
|
||||
if target.executionID != "" {
|
||||
if err := s.executionStopper.StopExecution(ctx, target.executionID, stopReason, true); err != nil {
|
||||
@@ -2174,7 +2207,7 @@ func (s *Service) stopTaskRuntimeTargets(ctx context.Context, taskID string, sto
|
||||
zap.Error(err))
|
||||
continue
|
||||
}
|
||||
failedStops[target.sessionID] = struct{}{}
|
||||
outcome.failed[target.sessionID] = struct{}{}
|
||||
s.logger.Warn(stopFailMsg,
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("session_id", target.sessionID),
|
||||
@@ -2191,20 +2224,95 @@ func (s *Service) stopTaskRuntimeTargets(ctx context.Context, taskID string, sto
|
||||
zap.Error(err))
|
||||
continue
|
||||
}
|
||||
failedStops[target.sessionID] = struct{}{}
|
||||
// A session-level not-found is retryable by default (the execution may
|
||||
// simply not be registered yet). But when the owned row is a
|
||||
// confirmed-dead LOCAL runtime, the runtime really is gone — treat the
|
||||
// stop as already complete so the durable cleanup job stops retrying a
|
||||
// runtime that will never come back.
|
||||
if runtimeStopAlreadyComplete(err) {
|
||||
if running := s.confirmedDeadLocalRow(ctx, target.sessionID); running != nil {
|
||||
s.reconcileConfirmedDeadRow(ctx, taskID, target.sessionID, running, &outcome)
|
||||
continue
|
||||
}
|
||||
}
|
||||
outcome.failed[target.sessionID] = struct{}{}
|
||||
s.logger.Warn(stopFailMsg,
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("session_id", target.sessionID),
|
||||
zap.Error(err))
|
||||
}
|
||||
}
|
||||
return failedStops
|
||||
return outcome
|
||||
}
|
||||
|
||||
// reconcileConfirmedDeadRow applies the resume-safety deletion invariant to a
|
||||
// confirmed-dead local row whose stop reported not-found. A row holding a
|
||||
// resume_token or backing a resumable session is repaired in place (never
|
||||
// deleted) and marked preserve-without-retry; a row that is neither is left for
|
||||
// the normal cleanup path to prune. Neither outcome is a failed stop, so the
|
||||
// durable cleanup job does not retry solely because the runtime was absent.
|
||||
func (s *Service) reconcileConfirmedDeadRow(
|
||||
ctx context.Context,
|
||||
taskID, sessionID string,
|
||||
running *models.ExecutorRunning,
|
||||
outcome *taskRuntimeStopOutcome,
|
||||
) {
|
||||
if models.RowMustBePreserved(running, s.sessionStateForCleanup(ctx, sessionID)) {
|
||||
outcome.preserve[sessionID] = struct{}{}
|
||||
if err := s.executors.RepairExecutorRunningDead(ctx, sessionID); err != nil {
|
||||
s.logger.Warn("failed to repair confirmed-dead resume-safe runtime row",
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("session_id", sessionID),
|
||||
zap.Error(err))
|
||||
}
|
||||
s.logger.Info("session runtime confirmed dead and resume-safe; repairing row in place",
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("session_id", sessionID))
|
||||
return
|
||||
}
|
||||
s.logger.Info("session runtime confirmed dead and already gone; treating stop as complete",
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("session_id", sessionID))
|
||||
}
|
||||
|
||||
// sessionStateForCleanup best-effort reads a session's state for the
|
||||
// resume-safety check. A missing session is treated as an empty (non-resumable)
|
||||
// state so a row's preservation then hinges on its resume_token alone.
|
||||
func (s *Service) sessionStateForCleanup(ctx context.Context, sessionID string) models.TaskSessionState {
|
||||
if s.sessions == nil {
|
||||
return ""
|
||||
}
|
||||
session, err := s.sessions.GetTaskSession(ctx, sessionID)
|
||||
if err != nil || session == nil {
|
||||
return ""
|
||||
}
|
||||
return session.State
|
||||
}
|
||||
|
||||
func runtimeStopAlreadyComplete(err error) bool {
|
||||
return errors.Is(err, runtimeapi.ErrNotFound)
|
||||
}
|
||||
|
||||
// confirmedDeadLocalRow returns the executors_running row backing sessionID when
|
||||
// it is a confirmed-dead LOCAL runtime — a local process handle that no longer
|
||||
// exists on this host — and nil otherwise. It returns nil when no prober is
|
||||
// wired, the row is missing/unreadable, or the runtime-aware liveness is
|
||||
// anything other than Dead, so an alive, unknown, or remote row is never treated
|
||||
// as absent.
|
||||
func (s *Service) confirmedDeadLocalRow(ctx context.Context, sessionID string) *models.ExecutorRunning {
|
||||
if s.rowLivenessProber == nil || s.executors == nil {
|
||||
return nil
|
||||
}
|
||||
running, err := s.executors.GetExecutorRunningBySessionID(ctx, sessionID)
|
||||
if err != nil || running == nil {
|
||||
return nil
|
||||
}
|
||||
if s.rowLivenessProber.RowLiveness(running) != models.ProcessLivenessDead {
|
||||
return nil
|
||||
}
|
||||
return running
|
||||
}
|
||||
|
||||
// performTaskCleanup handles post-deletion cleanup operations.
|
||||
// Handles worktree cleanup, executor_running records, and quick-chat workspace directories.
|
||||
// Agent stopping is handled separately in the DeleteTask background goroutine.
|
||||
|
||||
@@ -58,6 +58,8 @@ type stubExecutors struct {
|
||||
runningByTaskErr error
|
||||
runningBySession *models.ExecutorRunning
|
||||
runningBySessErr error
|
||||
deletedSessions []string
|
||||
repairedSessions []string
|
||||
}
|
||||
|
||||
func (s *stubExecutors) ListExecutorsRunningByTaskID(_ context.Context, _ string) ([]*models.ExecutorRunning, error) {
|
||||
@@ -68,7 +70,13 @@ func (s *stubExecutors) GetExecutorRunningBySessionID(_ context.Context, _ strin
|
||||
return s.runningBySession, s.runningBySessErr
|
||||
}
|
||||
|
||||
func (s *stubExecutors) DeleteExecutorRunningBySessionID(_ context.Context, _ string) error {
|
||||
func (s *stubExecutors) DeleteExecutorRunningBySessionID(_ context.Context, sessionID string) error {
|
||||
s.deletedSessions = append(s.deletedSessions, sessionID)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *stubExecutors) RepairExecutorRunningDead(_ context.Context, sessionID string) error {
|
||||
s.repairedSessions = append(s.repairedSessions, sessionID)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -200,7 +208,7 @@ func TestRefreshTaskRuntimeStopTargets_ExactRuntimeSupersedesEmptySessionFallbac
|
||||
targets,
|
||||
"archive",
|
||||
"stop failed",
|
||||
)
|
||||
).failed
|
||||
if len(failed) != 0 {
|
||||
t.Fatalf("late exact execution left retryable fallback failure: %v", failed)
|
||||
}
|
||||
@@ -270,7 +278,7 @@ func TestStopTaskRuntimeTargets_TerminalStopFailureDoesNotBlockCleanup(t *testin
|
||||
{sessionID: "sess-cancelled", terminal: true},
|
||||
}
|
||||
|
||||
failed := svc.stopTaskRuntimeTargets(context.Background(), "task-a", targets, "archive", "stop failed")
|
||||
failed := svc.stopTaskRuntimeTargets(context.Background(), "task-a", targets, "archive", "stop failed").failed
|
||||
if len(failed) != 0 {
|
||||
t.Errorf("terminal stop failure must not add to failedStops; got %v", failed)
|
||||
}
|
||||
@@ -288,7 +296,7 @@ func TestStopTaskRuntimeTargets_ExactRuntimeAbsenceIsIdempotent(t *testing.T) {
|
||||
[]taskStopTarget{{sessionID: "sess-running", executionID: "exec-running"}},
|
||||
"archive",
|
||||
"stop failed",
|
||||
)
|
||||
).failed
|
||||
if len(failed) != 0 {
|
||||
t.Errorf("exact runtime absence must be idempotent; got %v", failed)
|
||||
}
|
||||
@@ -306,12 +314,128 @@ func TestStopTaskRuntimeTargets_SessionRuntimeAbsenceRemainsRetryable(t *testing
|
||||
[]taskStopTarget{{sessionID: "sess-running"}},
|
||||
"archive",
|
||||
"stop failed",
|
||||
)
|
||||
).failed
|
||||
if _, ok := failed["sess-running"]; !ok {
|
||||
t.Errorf("session-level absence must remain retryable; got %v", failed)
|
||||
}
|
||||
}
|
||||
|
||||
// stubRowLivenessProber is a minimal TaskRowLivenessProber for tests.
|
||||
type stubRowLivenessProber struct {
|
||||
liveness models.ProcessLiveness
|
||||
}
|
||||
|
||||
func (s stubRowLivenessProber) RowLiveness(*models.ExecutorRunning) models.ProcessLiveness {
|
||||
return s.liveness
|
||||
}
|
||||
|
||||
// TestStopTaskRuntimeTargets_SessionAbsenceDeadLocalIsIdempotent is the
|
||||
// regression test for the infinite cleanup-retry bug: a session-only stop that
|
||||
// returns not-found for a CONFIRMED-DEAD LOCAL row must be treated as an
|
||||
// already-stopped runtime (not a failed stop) so the durable cleanup job stops
|
||||
// retrying forever.
|
||||
func TestStopTaskRuntimeTargets_SessionAbsenceDeadLocalIsIdempotent(t *testing.T) {
|
||||
svc, _, _ := createTestService(t)
|
||||
execs := &stubExecutors{
|
||||
runningBySession: &models.ExecutorRunning{
|
||||
SessionID: "sess-dead", Runtime: agentruntime.RuntimeStandalone, LocalPID: 4242,
|
||||
},
|
||||
}
|
||||
svc.executors = execs
|
||||
svc.executionStopper = &stubStopper{
|
||||
stopSessionErr: fmt.Errorf("gone: %w", runtimeapi.ErrNotFound),
|
||||
}
|
||||
svc.rowLivenessProber = stubRowLivenessProber{liveness: models.ProcessLivenessDead}
|
||||
|
||||
outcome := svc.stopTaskRuntimeTargets(
|
||||
context.Background(),
|
||||
"task-dead",
|
||||
[]taskStopTarget{{sessionID: "sess-dead"}},
|
||||
"archive",
|
||||
"stop failed",
|
||||
)
|
||||
if len(outcome.failed) != 0 {
|
||||
t.Errorf("session absence for a confirmed-dead local row must be idempotent; got %v", outcome.failed)
|
||||
}
|
||||
// A tokenless, non-resumable dead-local row is prunable: it must NOT be
|
||||
// preserved and must NOT be repaired in place — the normal cleanup path
|
||||
// deletes it.
|
||||
if len(outcome.preserve) != 0 {
|
||||
t.Errorf("tokenless dead-local row must be prunable, not preserved; got %v", outcome.preserve)
|
||||
}
|
||||
if len(execs.repairedSessions) != 0 {
|
||||
t.Errorf("tokenless dead-local row must not be repaired in place; got %v", execs.repairedSessions)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStopTaskRuntimeTargets_DeadLocalResumeSafeRowPreservedNotRetried is the
|
||||
// resume-safety regression: a confirmed-dead local row that still carries a
|
||||
// resume_token must be repaired in place (never torn down) AND excluded from the
|
||||
// retryable failedStops set, so the durable cleanup job neither retries forever
|
||||
// nor loses the only resume handle.
|
||||
func TestStopTaskRuntimeTargets_DeadLocalResumeSafeRowPreservedNotRetried(t *testing.T) {
|
||||
svc, _, _ := createTestService(t)
|
||||
execs := &stubExecutors{
|
||||
runningBySession: &models.ExecutorRunning{
|
||||
SessionID: "sess-tok", Runtime: agentruntime.RuntimeStandalone,
|
||||
LocalPID: 4242, ResumeToken: "tok-tok",
|
||||
},
|
||||
}
|
||||
svc.executors = execs
|
||||
svc.executionStopper = &stubStopper{
|
||||
stopSessionErr: fmt.Errorf("gone: %w", runtimeapi.ErrNotFound),
|
||||
}
|
||||
svc.rowLivenessProber = stubRowLivenessProber{liveness: models.ProcessLivenessDead}
|
||||
|
||||
outcome := svc.stopTaskRuntimeTargets(
|
||||
context.Background(),
|
||||
"task-tok",
|
||||
[]taskStopTarget{{sessionID: "sess-tok"}},
|
||||
"archive",
|
||||
"stop failed",
|
||||
)
|
||||
if _, retried := outcome.failed["sess-tok"]; retried {
|
||||
t.Errorf("resume-safe dead-local row must not be a retryable failed stop; got %v", outcome.failed)
|
||||
}
|
||||
if _, preserved := outcome.preserve["sess-tok"]; !preserved {
|
||||
t.Errorf("resume-safe dead-local row must be preserved from teardown; got %v", outcome.preserve)
|
||||
}
|
||||
if len(execs.repairedSessions) != 1 || execs.repairedSessions[0] != "sess-tok" {
|
||||
t.Errorf("resume-safe dead-local row must be repaired in place; got %v", execs.repairedSessions)
|
||||
}
|
||||
if len(execs.deletedSessions) != 0 {
|
||||
t.Errorf("resume-safe dead-local row must never be deleted; got %v", execs.deletedSessions)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStopTaskRuntimeTargets_SessionAbsenceUnknownRemainsRetryable guards the
|
||||
// anti-blanket-ignore rule for the cleanup worker: a session-only not-found stop
|
||||
// for a row whose liveness is Unknown (remote/no local handle) stays a failed,
|
||||
// retryable stop.
|
||||
func TestStopTaskRuntimeTargets_SessionAbsenceUnknownRemainsRetryable(t *testing.T) {
|
||||
svc, _, _ := createTestService(t)
|
||||
svc.executors = &stubExecutors{
|
||||
runningBySession: &models.ExecutorRunning{
|
||||
SessionID: "sess-remote", Runtime: agentruntime.RuntimeSSH,
|
||||
},
|
||||
}
|
||||
svc.executionStopper = &stubStopper{
|
||||
stopSessionErr: fmt.Errorf("not registered yet: %w", runtimeapi.ErrNotFound),
|
||||
}
|
||||
svc.rowLivenessProber = stubRowLivenessProber{liveness: models.ProcessLivenessUnknown}
|
||||
|
||||
failed := svc.stopTaskRuntimeTargets(
|
||||
context.Background(),
|
||||
"task-remote",
|
||||
[]taskStopTarget{{sessionID: "sess-remote"}},
|
||||
"archive",
|
||||
"stop failed",
|
||||
).failed
|
||||
if _, ok := failed["sess-remote"]; !ok {
|
||||
t.Errorf("session absence for an unknown/remote row must remain retryable; got %v", failed)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStopTaskRuntimeTargets_NonTerminalStopFailureBlocksCleanup(t *testing.T) {
|
||||
svc, _, _ := createTestService(t)
|
||||
svc.executors = &stubExecutors{}
|
||||
@@ -321,7 +445,7 @@ func TestStopTaskRuntimeTargets_NonTerminalStopFailureBlocksCleanup(t *testing.T
|
||||
{sessionID: "sess-running", terminal: false},
|
||||
}
|
||||
|
||||
failed := svc.stopTaskRuntimeTargets(context.Background(), "task-b", targets, "archive", "stop failed")
|
||||
failed := svc.stopTaskRuntimeTargets(context.Background(), "task-b", targets, "archive", "stop failed").failed
|
||||
if _, ok := failed["sess-running"]; !ok {
|
||||
t.Error("non-terminal stop failure must add session to failedStops")
|
||||
}
|
||||
@@ -424,6 +548,90 @@ func TestCleanupTaskResources_NonTerminalSessionStopFailureBlocksCleanup(t *test
|
||||
}
|
||||
}
|
||||
|
||||
// seedDeadLocalSessionOnly seeds a session and a session-only executor_running
|
||||
// row (empty agent_execution_id) so cleanup takes the StopSession path rather
|
||||
// than StopExecution. resumeToken, when set, makes the row resume-safe.
|
||||
func seedDeadLocalSessionOnly(t *testing.T, repo interface {
|
||||
CreateWorkspace(context.Context, *models.Workspace) error
|
||||
CreateWorkflow(context.Context, *models.Workflow) error
|
||||
CreateTask(context.Context, *models.Task) error
|
||||
CreateTaskSession(context.Context, *models.TaskSession) error
|
||||
UpsertExecutorRunning(context.Context, *models.ExecutorRunning) error
|
||||
}, wsID, wfID, taskID, sessID, resumeToken string, state models.TaskSessionState) {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
_ = repo.CreateWorkspace(ctx, &models.Workspace{ID: wsID, Name: "WS"})
|
||||
_ = repo.CreateWorkflow(ctx, &models.Workflow{ID: wfID, WorkspaceID: wsID, Name: "WF"})
|
||||
_ = repo.CreateTask(ctx, &models.Task{ID: taskID, WorkspaceID: wsID, WorkflowID: wfID, WorkflowStepID: "step-1", Title: "T", Priority: "medium"})
|
||||
_ = repo.CreateTaskSession(ctx, &models.TaskSession{ID: sessID, TaskID: taskID, State: state})
|
||||
if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{
|
||||
ID: sessID, SessionID: sessID, TaskID: taskID, ExecutorID: "executor-1",
|
||||
Runtime: agentruntime.RuntimeStandalone, Status: models.ExecutorRunningStatusRunning,
|
||||
LocalPID: 4242, ResumeToken: resumeToken,
|
||||
}); err != nil {
|
||||
t.Fatalf("seed executor_running: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCleanupTaskResources_NonTerminalDeadLocalSessionGetsCleanedUp is the
|
||||
// end-to-end companion for the confirmed-dead prune path: a non-terminal but
|
||||
// non-resumable (CREATED, no resume token) session with a session-only
|
||||
// executor_running row whose StopSession reports the runtime not-found and whose
|
||||
// local liveness is Dead must have its executor_running row removed after
|
||||
// cleanup — the durable job does not retry solely because the runtime was gone.
|
||||
func TestCleanupTaskResources_NonTerminalDeadLocalSessionGetsCleanedUp(t *testing.T) {
|
||||
svc, _, repo := createTestService(t)
|
||||
|
||||
stopper := newRecordingTaskExecutionStopper()
|
||||
// Mirror the not-found result executor.Stop now surfaces for a genuinely
|
||||
// missing session (the runtime seam normalizes to runtimeapi.ErrNotFound).
|
||||
stopper.stopSessionErr = fmt.Errorf("execution not found: %w: gone", runtimeapi.ErrNotFound)
|
||||
svc.SetExecutionStopper(stopper)
|
||||
svc.SetRowLivenessProber(stubRowLivenessProber{liveness: models.ProcessLivenessDead})
|
||||
svc.setCleanupDoneForTestHook(make(chan struct{}, 1))
|
||||
|
||||
seedDeadLocalSessionOnly(t, repo, "ws-dl", "wf-dl", "task-dl", "sess-dead-local", "", models.TaskSessionStateCreated)
|
||||
|
||||
svc.CleanupTaskResources(context.Background(), "task-dl", false)
|
||||
waitForCleanupDone(t, svc)
|
||||
|
||||
if _, err := repo.GetExecutorRunningBySessionID(context.Background(), "sess-dead-local"); err == nil {
|
||||
t.Error("executor_running row must be removed for a confirmed-dead local non-resumable session")
|
||||
}
|
||||
}
|
||||
|
||||
// TestCleanupTaskResources_DeadLocalResumeSafeSessionRepairedNotDeleted is the
|
||||
// resume-safety end-to-end case: a confirmed-dead local row that still carries a
|
||||
// resume_token must be repaired in place (row preserved, token intact,
|
||||
// local_pid cleared) rather than deleted.
|
||||
func TestCleanupTaskResources_DeadLocalResumeSafeSessionRepairedNotDeleted(t *testing.T) {
|
||||
svc, _, repo := createTestService(t)
|
||||
|
||||
stopper := newRecordingTaskExecutionStopper()
|
||||
stopper.stopSessionErr = fmt.Errorf("execution not found: %w: gone", runtimeapi.ErrNotFound)
|
||||
svc.SetExecutionStopper(stopper)
|
||||
svc.SetRowLivenessProber(stubRowLivenessProber{liveness: models.ProcessLivenessDead})
|
||||
svc.setCleanupDoneForTestHook(make(chan struct{}, 1))
|
||||
|
||||
// CREATED (non-resumable) state isolates the resume_token as the sole
|
||||
// preservation driver.
|
||||
seedDeadLocalSessionOnly(t, repo, "ws-rs", "wf-rs", "task-rs", "sess-resume-safe", "tok-keep", models.TaskSessionStateCreated)
|
||||
|
||||
svc.CleanupTaskResources(context.Background(), "task-rs", false)
|
||||
waitForCleanupDone(t, svc)
|
||||
|
||||
row, err := repo.GetExecutorRunningBySessionID(context.Background(), "sess-resume-safe")
|
||||
if err != nil {
|
||||
t.Fatalf("resume-safe dead-local row must be preserved, not deleted: %v", err)
|
||||
}
|
||||
if row.ResumeToken != "tok-keep" {
|
||||
t.Errorf("resume_token must survive repair; got %q", row.ResumeToken)
|
||||
}
|
||||
if row.Status != models.ExecutorRunningStatusStopped || row.LocalPID != 0 {
|
||||
t.Errorf("dead row must be repaired to stopped with cleared local_pid; got status=%q local_pid=%d", row.Status, row.LocalPID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCleanupTaskResources_SkipsBorrowedInheritedWorktree(t *testing.T) {
|
||||
svc, _, repo := createTestService(t)
|
||||
ctx := context.Background()
|
||||
|
||||
@@ -2555,6 +2555,7 @@ type recordingTaskExecutionStopper struct {
|
||||
stopExecutionCh chan stopExecutionCall
|
||||
stopExecutionErr error
|
||||
stopExecutionErrByID map[string]error
|
||||
stopSessionErr error
|
||||
claimExecutionFunc func(sessionID, executionID string, force bool) bool
|
||||
}
|
||||
|
||||
@@ -2567,7 +2568,7 @@ func (s *recordingTaskExecutionStopper) StopTask(context.Context, string, string
|
||||
}
|
||||
|
||||
func (s *recordingTaskExecutionStopper) StopSession(context.Context, string, string, bool) error {
|
||||
return nil
|
||||
return s.stopSessionErr
|
||||
}
|
||||
|
||||
func (s *recordingTaskExecutionStopper) StopExecution(_ context.Context, executionID, reason string, force bool) error {
|
||||
|
||||
@@ -24,7 +24,7 @@ func TestAttachWorkspaceSourcesDerivesLegacyPrimaryBranchForWorktreeProjection(t
|
||||
}
|
||||
primaryPath := t.TempDir()
|
||||
seedBareGitDir(t, primaryPath, "ref: refs/heads/main\n")
|
||||
if err := repo.CreateRepository(ctx, &models.Repository{ID: "repo-legacy-primary", WorkspaceID: "ws-legacy-primary", Name: "primary", LocalPath: primaryPath}); err != nil {
|
||||
if err := repo.CreateRepository(ctx, &models.Repository{ID: "repo-legacy-primary", WorkspaceID: "ws-legacy-primary", Name: "primary", LocalPath: canonicalRepoTestPath(t, primaryPath)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
task, err := svc.CreateTask(ctx, &CreateTaskRequest{WorkspaceID: "ws-legacy-primary", WorkflowID: "wf-legacy-primary", WorkflowStepID: "step", Title: "Task", Repositories: []TaskRepositoryInput{{RepositoryID: "repo-legacy-primary"}}})
|
||||
|
||||
@@ -6,8 +6,19 @@ import (
|
||||
"testing"
|
||||
)
|
||||
|
||||
// canonicalTempDir resolves symlinks in t.TempDir() so tests hand production
|
||||
// code a canonical owned control root (macOS /var -> /private/var).
|
||||
func canonicalTempDir(t *testing.T) string {
|
||||
t.Helper()
|
||||
d, err := filepath.EvalSymlinks(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatalf("EvalSymlinks(TempDir): %v", err)
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
func TestCreateOwnedDirectoryLinkCreatesLiveLinkInsideOwnedRoot(t *testing.T) {
|
||||
root := filepath.Join(t.TempDir(), "tasks", "task-1")
|
||||
root := filepath.Join(canonicalTempDir(t), "tasks", "task-1")
|
||||
target := filepath.Join(t.TempDir(), "source")
|
||||
if err := os.MkdirAll(target, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -72,7 +83,7 @@ func seedOwnedDirectoryLink(t *testing.T, root, name, target string) {
|
||||
// inspected entry is still on disk afterwards, since it may be a link the user
|
||||
// or the repository keeps on purpose.
|
||||
func TestIsSelfReferentialDirectoryLinkDetectsSelfLink(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
root := canonicalTempDir(t)
|
||||
seedOwnedDirectoryLink(t, root, "self", root)
|
||||
|
||||
selfLink, err := IsSelfReferentialDirectoryLink(root, "self")
|
||||
@@ -85,7 +96,7 @@ func TestIsSelfReferentialDirectoryLinkDetectsSelfLink(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestIsSelfReferentialDirectoryLinkIgnoresForeignLink(t *testing.T) {
|
||||
root, target := t.TempDir(), t.TempDir()
|
||||
root, target := canonicalTempDir(t), t.TempDir()
|
||||
if err := os.WriteFile(filepath.Join(target, "live.txt"), []byte("one"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -130,7 +141,7 @@ func TestIsSelfReferentialDirectoryLinkIgnoresMissingEntry(t *testing.T) {
|
||||
// call must be a no-op. A path-text comparison rejected Windows junctions here
|
||||
// because filepath.EvalSymlinks does not traverse them.
|
||||
func TestEnsureOwnedDirectoryLinkIsIdempotent(t *testing.T) {
|
||||
root := filepath.Join(t.TempDir(), "tasks", "task-1")
|
||||
root := filepath.Join(canonicalTempDir(t), "tasks", "task-1")
|
||||
target := t.TempDir()
|
||||
if err := os.WriteFile(filepath.Join(target, "live.txt"), []byte("one"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -156,7 +167,7 @@ func TestEnsureOwnedDirectoryLinkIsIdempotent(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestEnsureOwnedDirectoryLinkRejectsDifferentTarget(t *testing.T) {
|
||||
root := filepath.Join(t.TempDir(), "tasks", "task-1")
|
||||
root := filepath.Join(canonicalTempDir(t), "tasks", "task-1")
|
||||
target, other := t.TempDir(), t.TempDir()
|
||||
if _, _, err := EnsureOwnedDirectoryLink(root, "api", target); err != nil {
|
||||
t.Fatalf("EnsureOwnedDirectoryLink: %v", err)
|
||||
|
||||
@@ -21,7 +21,7 @@ func TestPruneQuarantinedWorkspacePreservesOtherRecoverableRegistration(t *testi
|
||||
t.Fatalf("NewManager: %v", err)
|
||||
}
|
||||
|
||||
originalRoot := t.TempDir()
|
||||
originalRoot := canonicalTempDir(t)
|
||||
quarantineRoot := t.TempDir()
|
||||
firstPath := filepath.Join(originalRoot, "task-1", "repo")
|
||||
secondPath := filepath.Join(originalRoot, "task-2", "repo")
|
||||
|
||||
@@ -0,0 +1,131 @@
|
||||
---
|
||||
spec: docs/specs/platform/cron-office-disabled-safety.md
|
||||
created: 2026-08-03
|
||||
status: complete
|
||||
---
|
||||
|
||||
# Implementation Plan: Three Backend Regressions
|
||||
|
||||
Three independently reviewable backend regressions observed in
|
||||
`~/.kandev/logs/backend-logs.log`, repaired with TDD. Each task owns a disjoint
|
||||
package and is separately reviewable.
|
||||
|
||||
## Related Specs
|
||||
|
||||
- `docs/specs/platform/cron-office-disabled-safety.md` (Task 01, new repair spec)
|
||||
- `docs/specs/tasks/title-length-limit.md` (Task 02, amended)
|
||||
- `docs/specs/tasks/runtime-cleanup.md` (Task 03, amended)
|
||||
|
||||
## Root Causes
|
||||
|
||||
### 1. Office-disabled routines cron panic
|
||||
|
||||
When Office is disabled, `services.OfficeSvcs` is nil so `officeRoutines` stays a
|
||||
nil `*office/routines.RoutineService` pointer
|
||||
(`backendapp/main.go:1280-1283`). `startCronScheduler`
|
||||
(`backendapp/cron.go:39`) passes that nil pointer straight into
|
||||
`schedulercron.NewRoutinesHandler(routineSvc, nil, log)`, which stores it in the
|
||||
`RoutineTicker` interface field. A typed-nil pointer inside an interface is a
|
||||
non-nil interface value, so `RoutinesHandler.Tick`'s `h.ticker == nil` guard
|
||||
(`scheduler/cron/routines.go:57-61`) is bypassed. `TickScheduledTriggers` then
|
||||
runs on a nil receiver and dereferences the service's repository, panicking every
|
||||
30 seconds. The loop's per-handler panic recovery re-arms each tick.
|
||||
|
||||
### 2. GitHub review/issue titles exceed the 60-rune limit
|
||||
|
||||
`buildReviewTaskRequest` (`orchestrator/event_handlers_github.go:451`) builds
|
||||
`fmt.Sprintf("PR #%d: %s", pr.Number, pr.Title)` with no length cap, and the
|
||||
issue watcher (`event_handlers_github.go:1368`) builds
|
||||
`fmt.Sprintf("Issue #%d: %s", ...)` the same way. Both feed `CreateReviewTask` /
|
||||
`CreateIssueTask` → `Service.CreateTask`, whose `validateTaskTitle(req.Title)`
|
||||
(`task/service/service_tasks.go:341`) rejects titles over 60 runes.
|
||||
`TruncateTaskTitle` is only applied to the auto-title derivation path
|
||||
(`service_tasks.go:212`), never to caller-supplied titles, so long PR/issue
|
||||
titles make the whole review/issue task fail to create.
|
||||
|
||||
### 3. Missing-session runtime cleanup preserves stale rows
|
||||
|
||||
`Executor.Stop` (`orchestrator/executor/executor_interaction.go:21-27`) maps
|
||||
every session-lookup error to `ErrExecutionNotFound`, and `StopExecution`
|
||||
(`:173-192`) wraps every stop error — including `lifecycle.ErrExecutionNotFound`
|
||||
— as `ErrExecutionNotFound` without letting callers distinguish "already gone"
|
||||
from "real failure". `handleMissingSessionOnStartup`
|
||||
(`orchestrator/service.go:1725-1748`) preserves the row whenever
|
||||
`StopAgentWithReason` returns any error, including a legitimate not-found for a
|
||||
dead process. In the durable worker, `executeTaskResourceCleanupJob`
|
||||
(`task/service/resource_cleanup_jobs.go:396,413-419`) counts every failed stop —
|
||||
not-found included — as a `failedStops` entry and returns an error, so the job
|
||||
re-enters `retry_wait` and exhausts its full retry budget (the documented
|
||||
eight-claim bound, whose eighth failed attempt becomes terminal `failed`) even
|
||||
though the runtime is already gone. Confirmed-dead local runtimes are never
|
||||
pruned and every cleanup for them burns the whole retry budget before failing.
|
||||
|
||||
## Backend Changes
|
||||
|
||||
- **Task 01 (cron):** In `backendapp/cron.go`, only assign the routines service
|
||||
into a `schedulercron.RoutineTicker` variable when non-nil, so a genuinely nil
|
||||
interface reaches `NewRoutinesHandler`. Optionally harden
|
||||
`NewRoutinesHandler`/`Tick` to defend against a typed-nil concrete value.
|
||||
Preserve normal forwarding when Office is enabled.
|
||||
- **Task 02 (titles):** Route the complete generated title through
|
||||
`taskservice.TruncateTaskTitle` in `buildReviewTaskRequest` and the GitHub
|
||||
issue task builder, before the request leaves the orchestrator.
|
||||
- **Task 03 (cleanup):** Add a runtime-aware "already stopped" classification so
|
||||
a not-found stop for a confirmed-dead local row is a success, then thread it
|
||||
through startup missing-session handling and the durable cleanup worker so such
|
||||
rows are pruned/repaired under `RowMustBePreserved` and no longer counted as
|
||||
failed stops. Preserve alive/unknown/remote rows and non-not-found errors.
|
||||
|
||||
## Tests (RED before GREEN)
|
||||
|
||||
- **Task 01:** Two complementary layers. Behavioral: `scheduler/cron/routines_test.go`
|
||||
proves a handler built from a typed-nil `RoutineTicker` (the exact production
|
||||
panic shape) takes the no-op branch on `Tick` without panicking. Static guard:
|
||||
`backendapp/scheduler_wiring_test.go` asserts (via AST) that `startCronScheduler`
|
||||
never passes the raw `routineSvc` pointer straight into `NewRoutinesHandler`,
|
||||
so the nil-gating (`RoutineTicker` interface variable) cannot be dropped and
|
||||
reintroduce the typed-nil interface. The backendapp check is intentionally
|
||||
static — a runtime `startCronScheduler` test would require standing up the full
|
||||
`*Repositories`/dispatcher graph for no additional behavioral coverage beyond
|
||||
the handler test.
|
||||
- **Task 02:** `orchestrator/event_handlers_github_review_test.go` (and the issue
|
||||
builder test) — ASCII and multibyte over-limit titles keep the `PR #`/`Issue #`
|
||||
prefix, stay ≤60 runes, and end with `…`.
|
||||
- **Task 03:** `orchestrator/reconcile_restart_test.go` and
|
||||
`task/service/service_tasks_stop_test.go` /
|
||||
`task/service/resource_cleanup_jobs_test.go` — dead/alive/unknown-remote/
|
||||
missing-session/terminal-session cases; confirmed-dead not-found prunes/repairs
|
||||
and the cleanup job does not retry; alive/unknown preserved; non-not-found stays
|
||||
retryable.
|
||||
|
||||
## Implementation Waves And Parallel Candidates
|
||||
|
||||
Task 02 and Task 03 both modify `internal/orchestrator` (Task 02 the GitHub
|
||||
title builders, Task 03 the stop-error classification), so they are not
|
||||
package-disjoint — but their files and symbols are disjoint, so they do not
|
||||
collide. Task 01 is fully isolated in `backendapp`+`scheduler/cron`. All three
|
||||
share no schema, migration, generated contract, lockfile, or package config.
|
||||
|
||||
- Wave 1 (parallel-safe): Task 01, Task 02.
|
||||
- Wave 1 (sequential within itself): Task 03.
|
||||
|
||||
`parallelism: sequential` remains the default per task; waves are a human aid,
|
||||
not authorization to delegate.
|
||||
|
||||
- [x] [Task 01: Office-disabled routines cron no-op](task-01-office-disabled-routines-noop.md)
|
||||
- [x] [Task 02: Truncate generated review/issue task titles](task-02-truncate-generated-titles.md)
|
||||
- [x] [Task 03: Prune confirmed-dead missing-session runtime rows](task-03-confirmed-dead-runtime-cleanup.md)
|
||||
|
||||
## Documentation Impact
|
||||
|
||||
Durable specs are amended/added in this design package. No public CLI, config,
|
||||
API, or UI documentation changes are required.
|
||||
|
||||
## Risks
|
||||
|
||||
- Task 03 must classify only typed not-found sentinels for confirmed-dead local
|
||||
rows; matching error strings or ignoring all not-found could hide real stop
|
||||
failures or prune alive/remote rows. Judge "dead" with the existing
|
||||
runtime-aware liveness probe, and gate deletion on `RowMustBePreserved`.
|
||||
- Task 01's typed-nil fix is small but easy to reintroduce; the regression test
|
||||
must construct the typed-nil interface explicitly, not a plain `nil`.
|
||||
+75
@@ -0,0 +1,75 @@
|
||||
---
|
||||
id: "01-office-disabled-routines-noop"
|
||||
title: "Office-disabled routines cron no-op"
|
||||
status: complete
|
||||
wave: 1
|
||||
depends_on: []
|
||||
plan: "plan.md"
|
||||
spec: "../../specs/platform/cron-office-disabled-safety.md"
|
||||
parallel-safe: true
|
||||
---
|
||||
|
||||
# Task 01: Office-Disabled Routines Cron No-Op
|
||||
|
||||
## Root Cause
|
||||
|
||||
`backendapp/main.go:1280-1283` leaves `officeRoutines` a nil
|
||||
`*office/routines.RoutineService` pointer when Office is disabled.
|
||||
`startCronScheduler` (`backendapp/cron.go:39`) passes it into
|
||||
`schedulercron.NewRoutinesHandler`, storing a typed-nil pointer in the
|
||||
`RoutineTicker` interface field. `RoutinesHandler.Tick`'s `h.ticker == nil`
|
||||
guard (`scheduler/cron/routines.go:57-61`) is bypassed because a typed-nil
|
||||
pointer in an interface is a non-nil interface, so `TickScheduledTriggers` runs
|
||||
on a nil receiver and panics every 30 seconds.
|
||||
|
||||
## Acceptance
|
||||
|
||||
- With Office disabled, the shared cron loop ticks without a routines
|
||||
nil-pointer panic and other handlers keep running.
|
||||
- A `RoutinesHandler` constructed from an absent routines service treats each
|
||||
tick as a no-op returning success.
|
||||
- With Office enabled, each tick still forwards to
|
||||
`RoutineService.TickScheduledTriggers` (behavior unchanged).
|
||||
|
||||
## Regression Test (RED first)
|
||||
|
||||
- In `scheduler/cron/routines_test.go`: build a handler whose ticker is a
|
||||
typed-nil concrete `RoutineTicker` (i.e. a `(*fakeTicker)(nil)` assigned into
|
||||
the interface, mirroring the production typed-nil) and assert `Tick` returns
|
||||
nil without panic. Add/confirm a positive case that a real ticker is forwarded.
|
||||
- Add a `backendapp` test that `startCronScheduler` (or the routines-handler
|
||||
wiring helper) with a nil routines service produces a handler that ticks safely.
|
||||
|
||||
## Verification
|
||||
|
||||
```bash
|
||||
cd apps/backend && go test -tags fts5 -run 'TestRoutinesHandler' ./internal/scheduler/cron
|
||||
cd apps/backend && go test -tags fts5 -run 'TestStartCronScheduler|TestCron' ./internal/backendapp
|
||||
```
|
||||
|
||||
## Files likely touched
|
||||
|
||||
- `apps/backend/internal/backendapp/cron.go`
|
||||
- `apps/backend/internal/backendapp/main.go` (only if wiring moves here)
|
||||
- `apps/backend/internal/scheduler/cron/routines.go` (optional hardening)
|
||||
- `apps/backend/internal/scheduler/cron/routines_test.go`
|
||||
- `apps/backend/internal/backendapp/*_test.go` (wiring regression)
|
||||
|
||||
## Dependencies
|
||||
|
||||
None.
|
||||
|
||||
## Parallelism
|
||||
|
||||
`sequential` within itself; `parallel-safe` against Tasks 02 and 03 (disjoint
|
||||
packages, no shared schema/contract/config).
|
||||
|
||||
## Inputs
|
||||
|
||||
- Repair spec: `docs/specs/platform/cron-office-disabled-safety.md`.
|
||||
- Confirmed 22 recurring panics in `~/.kandev/logs/backend-logs.log`.
|
||||
|
||||
## Output contract
|
||||
|
||||
Report the RED panic reproduction, the wiring change, files changed, exact
|
||||
verification result, residual risks, and update this task plus `plan.md` status.
|
||||
+83
@@ -0,0 +1,83 @@
|
||||
---
|
||||
id: "02-truncate-generated-titles"
|
||||
title: "Truncate generated review/issue task titles"
|
||||
status: complete
|
||||
wave: 1
|
||||
depends_on: []
|
||||
plan: "plan.md"
|
||||
spec: "../../specs/tasks/title-length-limit.md"
|
||||
parallel-safe: true
|
||||
---
|
||||
|
||||
# Task 02: Truncate Generated Review/Issue Task Titles
|
||||
|
||||
## Root Cause
|
||||
|
||||
`buildReviewTaskRequest` (`orchestrator/event_handlers_github.go:451`) builds
|
||||
`fmt.Sprintf("PR #%d: %s", pr.Number, pr.Title)` and the issue watcher
|
||||
(`event_handlers_github.go:1368`) builds `fmt.Sprintf("Issue #%d: %s", ...)`,
|
||||
neither capped. Both feed `Service.CreateTask`, whose
|
||||
`validateTaskTitle(req.Title)` (`task/service/service_tasks.go:341`) rejects
|
||||
titles over 60 runes. `TruncateTaskTitle` is only applied to the auto-title
|
||||
derivation path, not caller-supplied titles, so long PR/issue titles drop the
|
||||
review/issue task entirely.
|
||||
|
||||
## Acceptance
|
||||
|
||||
- The complete generated review title and issue title are passed through
|
||||
`taskservice.TruncateTaskTitle` before leaving the orchestrator.
|
||||
- Over-limit titles keep the `PR #<n>:` / `Issue #<n>:` prefix, are ≤60 runes,
|
||||
and end with a single `…`. Multibyte titles are counted by code point.
|
||||
- Titles at or under 60 runes are unchanged.
|
||||
|
||||
## Audit (in scope, do not broaden)
|
||||
|
||||
`buildReviewTaskRequest` and the GitHub issue builder (`event_handlers_github.go`)
|
||||
are the two builders that feed `CreateTask` without truncation. Confirm; fix both.
|
||||
Adjacent builders are already safe and are OUT of scope unless a test shows
|
||||
otherwise: `event_handlers_automation.go` already calls `TruncateTaskTitle`
|
||||
(lines 314/321/324); `source_sentry.go` truncates via `truncateSentryTitle`;
|
||||
`office/routines/service.go:503` already truncates. `source_jira.go`,
|
||||
`source_linear.go`, `source_gitlab.go`, `source_azuredevops.go` build
|
||||
`IssueTaskRequest` titles — note them, but only fix if a regression test
|
||||
demonstrates they reach `CreateTask` without truncation.
|
||||
|
||||
## Regression Test (RED first)
|
||||
|
||||
- In `orchestrator/event_handlers_github_review_test.go`: table-driven
|
||||
`buildReviewTaskRequest` cases — short ASCII (unchanged), long ASCII
|
||||
(prefix kept, ≤60 runes, ends `…`), long multibyte (no split, ≤60 runes). Add
|
||||
the equivalent for the issue builder.
|
||||
|
||||
## Verification
|
||||
|
||||
```bash
|
||||
cd apps/backend && go test -tags fts5 -run 'TestBuildReviewTaskRequest|TestBuildIssueTaskRequest|Title' ./internal/orchestrator
|
||||
```
|
||||
|
||||
## Files likely touched
|
||||
|
||||
- `apps/backend/internal/orchestrator/event_handlers_github.go`
|
||||
- `apps/backend/internal/orchestrator/event_handlers_github_review_test.go`
|
||||
- `apps/backend/internal/task/service/task_title.go` (read-only reference)
|
||||
|
||||
## Dependencies
|
||||
|
||||
None.
|
||||
|
||||
## Parallelism
|
||||
|
||||
`sequential` within itself; `parallel-safe` against Tasks 01 and 03.
|
||||
|
||||
## Inputs
|
||||
|
||||
- Amended spec: `docs/specs/tasks/title-length-limit.md` (watcher-title bullet
|
||||
and scenarios).
|
||||
- Confirmed 5 "task title is too long" review-task failures in the logs
|
||||
(PRs #12096, #12057, #12054).
|
||||
|
||||
## Output contract
|
||||
|
||||
Report the RED failures, the truncation change, the audit outcome for adjacent
|
||||
builders, files changed, exact verification result, residual risks, and update
|
||||
this task plus `plan.md` status.
|
||||
+101
@@ -0,0 +1,101 @@
|
||||
---
|
||||
id: "03-confirmed-dead-runtime-cleanup"
|
||||
title: "Prune confirmed-dead missing-session runtime rows"
|
||||
status: complete
|
||||
wave: 1
|
||||
depends_on: []
|
||||
plan: "plan.md"
|
||||
spec: "../../specs/tasks/runtime-cleanup.md"
|
||||
parallel-safe: false
|
||||
---
|
||||
|
||||
# Task 03: Prune Confirmed-Dead Missing-Session Runtime Rows
|
||||
|
||||
## Root Cause
|
||||
|
||||
- `Executor.Stop` (`orchestrator/executor/executor_interaction.go:21-27`) maps
|
||||
every session-lookup error to `ErrExecutionNotFound`, and `StopExecution`
|
||||
(`:173-192`) wraps every stop error — including `lifecycle.ErrExecutionNotFound`
|
||||
— as `ErrExecutionNotFound`, so callers cannot tell "already gone" from a real
|
||||
failure.
|
||||
- `handleMissingSessionOnStartup` (`orchestrator/service.go:1725-1748`) preserves
|
||||
the row on any `StopAgentWithReason` error, including a legitimate not-found for
|
||||
a dead process.
|
||||
- `executeTaskResourceCleanupJob`
|
||||
(`task/service/resource_cleanup_jobs.go:396,413-419`) counts every failed stop
|
||||
(not-found included) into `failedStops`, returns an error, and the job
|
||||
re-enters `retry_wait` forever.
|
||||
|
||||
## Acceptance
|
||||
|
||||
- A not-found stop for a **confirmed-dead local** row (runtime-aware liveness ==
|
||||
Dead) is recorded as a successful stop; the row is pruned or repaired under
|
||||
`RowMustBePreserved` (token/worktree preserved when required).
|
||||
- Such a row is NOT counted as a `failedStops` entry, so the durable cleanup job
|
||||
does not retry solely because the owned runtime is absent.
|
||||
- **Alive** and **Unknown/remote** (SSH/containerized/no local handle) rows are
|
||||
preserved on a not-found stop and the outcome stays retryable.
|
||||
- Non-not-found session/task lookup or stop errors stay retryable; not every
|
||||
session/execution not-found is blanket-ignored.
|
||||
- Runtime-specific persisted handles (e.g. `agent_execution_id`) are used to
|
||||
decide the stop result where available.
|
||||
|
||||
## Regression Test (RED first)
|
||||
|
||||
- `orchestrator/reconcile_restart_test.go`: missing-session row that is
|
||||
confirmed-dead-local + not-found stop → row pruned/repaired; alive → preserved;
|
||||
unknown/remote (SSH) → preserved with handle+token; non-not-found stop error →
|
||||
preserved retryable.
|
||||
- `task/service/service_tasks_stop_test.go` and
|
||||
`task/service/resource_cleanup_jobs_test.go`: confirmed-dead not-found stop is
|
||||
not a `failedStops` entry and the job completes (no retry); alive/unknown remain
|
||||
failures/preserved; terminal-session case behaves per resume-safety.
|
||||
|
||||
## Verification
|
||||
|
||||
```bash
|
||||
cd apps/backend && go test -tags fts5 -run 'TestReconcile|MissingSession|StopExecution|Stop' ./internal/orchestrator ./internal/orchestrator/executor
|
||||
cd apps/backend && go test -tags fts5 -run 'TaskResourceCleanup|Stop|Cleanup' ./internal/task/service
|
||||
```
|
||||
|
||||
## Files likely touched
|
||||
|
||||
- `apps/backend/internal/orchestrator/executor/executor_interaction.go`
|
||||
- `apps/backend/internal/orchestrator/service.go`
|
||||
- `apps/backend/internal/orchestrator/reconcile_restart_test.go`
|
||||
- `apps/backend/internal/task/service/service_tasks.go`
|
||||
- `apps/backend/internal/task/service/service_tasks_stop_test.go`
|
||||
- `apps/backend/internal/task/service/resource_cleanup_jobs.go`
|
||||
- `apps/backend/internal/task/service/resource_cleanup_jobs_test.go`
|
||||
- `apps/backend/internal/task/models/resume_safety.go` (read-only reference)
|
||||
|
||||
## Dependencies
|
||||
|
||||
None on Tasks 01/02, but internally sequential: the error-classification change
|
||||
in the executor precedes the startup and worker consumers.
|
||||
|
||||
## Parallelism
|
||||
|
||||
`sequential`; spans `orchestrator`, `orchestrator/executor`, and `task/service`
|
||||
and must land as one reviewable change so callers and classification stay
|
||||
consistent. Not parallel-safe.
|
||||
|
||||
## Inputs
|
||||
|
||||
- Amended spec: `docs/specs/tasks/runtime-cleanup.md` (confirmed-dead
|
||||
not-found-as-stopped bullets, failure modes, and scenarios).
|
||||
- Existing runtime-aware liveness: `lifecycle.RowProcessLiveness` /
|
||||
`models.ProcessLiveness` and `RowMustBePreserved`
|
||||
(`task/models/resume_safety.go`).
|
||||
- Existing sentinels: `lifecycle.ErrExecutionNotFound`,
|
||||
`executor.ErrExecutionNotFound`, `runtimeapi.ErrNotFound`,
|
||||
`models.ErrTaskSessionNotFound`, `models.ErrExecutorRunningNotFound`.
|
||||
- Confirmed "resumed task resource cleanup job failed: 1 runtime stop operations
|
||||
failed" in the logs (job 7e3d58bd..., task 76cbddb5...).
|
||||
|
||||
## Output contract
|
||||
|
||||
Report the RED failures for each case (dead/alive/unknown-remote/missing-session/
|
||||
terminal), the classification and consumer changes, files changed, exact
|
||||
verification results, evidence the cleanup job no longer retries a confirmed-dead
|
||||
runtime, residual risks, and update this task plus `plan.md` status.
|
||||
@@ -0,0 +1,106 @@
|
||||
---
|
||||
status: draft
|
||||
created: 2026-08-03
|
||||
owner: cfl
|
||||
---
|
||||
|
||||
# Shared Cron Loop Safety When Office Is Disabled
|
||||
|
||||
## Why
|
||||
|
||||
The shared cron loop (ADR-0004 Phase 5) drives every backend timer — heartbeat,
|
||||
budget, Office routines, and Office recovery — on a single goroutine. Office is
|
||||
an optional feature. When Office is disabled the routines collaborator does not
|
||||
exist, and the cron loop must keep running the other handlers instead of
|
||||
crashing. A crash here is severe because the loop's panic recovery re-arms every
|
||||
tick: the backend logs a nil-pointer panic every 30 seconds for the life of the
|
||||
process and never fires heartbeat or budget work.
|
||||
|
||||
## What
|
||||
|
||||
- The shared cron loop starts and keeps ticking whether or not Office is
|
||||
enabled.
|
||||
- When the Office routines service is absent, the routines cron handler performs
|
||||
no work and returns success on every tick. It does not panic, log an error, or
|
||||
abort the loop.
|
||||
- When the Office routines service is present, the routines cron handler forwards
|
||||
each tick to `RoutineService.TickScheduledTriggers` exactly as before. Enabling
|
||||
Office does not change routine dispatch behavior.
|
||||
- The remaining cron handlers (heartbeat, budget, Office recovery) run
|
||||
independently of whether the routines handler is a no-op.
|
||||
- Wiring a nil collaborator into any cron handler is safe: a genuinely absent
|
||||
collaborator produces a no-op handler, never a handler that dereferences a nil
|
||||
value during `Tick`.
|
||||
|
||||
## Data Model
|
||||
|
||||
No data model change. The behavior concerns in-process wiring of the cron loop
|
||||
handlers only.
|
||||
|
||||
## API Surface
|
||||
|
||||
No user-facing HTTP or WebSocket surface. The affected contract is the internal
|
||||
cron handler seam:
|
||||
|
||||
```go
|
||||
// RoutineTicker is the surface RoutinesHandler needs from the office
|
||||
// routines service.
|
||||
type RoutineTicker interface {
|
||||
TickScheduledTriggers(ctx context.Context, now time.Time) error
|
||||
}
|
||||
|
||||
func NewRoutinesHandler(ticker RoutineTicker, now func() time.Time, log *logger.Logger) *RoutinesHandler
|
||||
```
|
||||
|
||||
`RoutinesHandler.Tick` treats a nil ticker as a no-op. The safety guarantee is
|
||||
that this nil check is reached whenever no routines service is configured,
|
||||
regardless of how the absence is expressed by the caller.
|
||||
|
||||
## Failure Modes
|
||||
|
||||
- **Typed-nil collaborator (the regression).** A concrete nil pointer
|
||||
(`(*office/routines.RoutineService)(nil)`) assigned into the `RoutineTicker`
|
||||
interface produces a non-nil interface value. `RoutinesHandler.Tick`'s
|
||||
`h.ticker == nil` guard is then bypassed, `TickScheduledTriggers` runs on a nil
|
||||
receiver, dereferences the service's repository, and panics. The cron loop's
|
||||
per-handler panic recovery converts this into a recurring error every tick.
|
||||
Desired behavior: the handler must observe a genuinely nil interface when no
|
||||
service exists, so the no-op branch is taken.
|
||||
- **Office enabled but routines service still nil.** Should never happen in
|
||||
practice, but the handler must still no-op rather than panic.
|
||||
- **Routines service returns an error while Office is enabled.** Unchanged: the
|
||||
error propagates from `Tick` to the loop, which logs it without aborting other
|
||||
handlers.
|
||||
|
||||
## Persistence Guarantees
|
||||
|
||||
None. This behavior is process-local wiring and has no persisted state.
|
||||
|
||||
## Scenarios
|
||||
|
||||
- **GIVEN** the backend starts with Office disabled, **WHEN** the shared cron
|
||||
loop ticks, **THEN** the routines handler performs no work, returns success,
|
||||
does not panic, and the heartbeat, budget, and Office-recovery handlers keep
|
||||
running.
|
||||
- **GIVEN** the backend starts with Office disabled, **WHEN** the cron loop runs
|
||||
for multiple ticks, **THEN** no nil-pointer panic is logged for the routines
|
||||
handler on any tick.
|
||||
- **GIVEN** the backend starts with Office enabled and a real routines service,
|
||||
**WHEN** the cron loop ticks, **THEN** the routines handler forwards each tick
|
||||
to `RoutineService.TickScheduledTriggers` and normal routine dispatch is
|
||||
unchanged.
|
||||
- **GIVEN** a routines cron handler is constructed with a nil routines
|
||||
collaborator by any wiring path, **WHEN** it ticks, **THEN** it returns success
|
||||
without dereferencing the collaborator.
|
||||
|
||||
## Out of Scope
|
||||
|
||||
- Changing routine dispatch, concurrency policy, fingerprinting, or cron
|
||||
expression evaluation inside the Office routines service.
|
||||
- Changing heartbeat, budget, or Office-recovery handler behavior.
|
||||
- Adding a user-facing surface for cron health.
|
||||
- Making the cron tick interval configurable.
|
||||
|
||||
## Implementation Plan
|
||||
|
||||
See [the implementation plan](../../plans/backend-regressions-office-disabled-cleanup/plan.md).
|
||||
@@ -55,6 +55,28 @@ worktrees, or executor rows behind and the machine slowly runs out of memory.
|
||||
- A cleanup operation treats a classifiable missing owned resource as already
|
||||
complete. This includes a missing worktree and a missing `TaskEnvironment`
|
||||
row captured by the durable cleanup snapshot.
|
||||
- Stopping a runtime that is already gone is a successful, idempotent stop, not a
|
||||
failed stop. When a stop operation is asked to stop an execution or session
|
||||
that no longer exists, and the row it owns is a confirmed-dead local runtime,
|
||||
the cleanup treats the runtime as already stopped so the runtime row can be
|
||||
pruned or repaired under the resume-safety invariant and the durable cleanup
|
||||
job does not retry solely because the owned runtime is absent.
|
||||
- "Confirmed dead" is judged in a runtime-aware way. A row is confirmed dead only
|
||||
when it is a local runtime whose local liveness handle refers to a process that
|
||||
no longer exists on this host. An alive local runtime, or a runtime whose
|
||||
liveness is Unknown (remote SSH, containerized, or a local row with no liveness
|
||||
handle), is never treated as dead on the basis of a not-found stop result; its
|
||||
row is preserved and the outcome remains retryable.
|
||||
- Missing-runtime handling never blanket-ignores every session-level or
|
||||
execution-level not-found error. Only a not-found result for a runtime whose row
|
||||
is confirmed dead is reclassified as a successful stop. A not-found result for
|
||||
an alive or unknown/remote row, or a session/task lookup error that is not a
|
||||
typed not-found sentinel, remains a preserved, retryable outcome so a transient
|
||||
store error is never mistaken for an absent runtime.
|
||||
- Where a runtime-specific persisted stop handle exists (for example
|
||||
`agent_execution_id`, or a remote/container handle), missing-runtime handling
|
||||
uses it to decide the stop result rather than inferring absence from a generic
|
||||
error.
|
||||
- Archive, delete, cascade, workspace-delete, and quick-chat expiration persist a
|
||||
cleanup intent and resource snapshot before mutating or deleting task state.
|
||||
Cleanup is performed by a durable worker rather than a detached goroutine.
|
||||
@@ -174,6 +196,18 @@ The durable cleanup job wraps that resource lifecycle:
|
||||
- If a runtime row points at a missing in-memory execution, cleanup attempts the
|
||||
runtime-specific persisted handle when available. If no handle can be used, the
|
||||
row is preserved with a warning instead of being silently dropped.
|
||||
- If a stop operation reports the execution or session is not found and the owned
|
||||
row is a confirmed-dead local runtime, cleanup records the stop as successful,
|
||||
prunes or repairs the row under the resume-safety invariant, proceeds with any
|
||||
now-safe shared-environment/worktree teardown, and does not count the row as a
|
||||
failed stop. The durable cleanup job therefore does not re-enter `retry_wait`
|
||||
solely because that owned runtime was confirmed absent.
|
||||
- If a stop operation reports not found but the owned row is alive, or its
|
||||
liveness is Unknown (remote/containerized/no local handle), cleanup preserves
|
||||
the row and treats the outcome as retryable rather than pruning it.
|
||||
- If a session or task lookup during stop fails with an error that is not a typed
|
||||
not-found sentinel, cleanup treats it as a retryable failure and preserves the
|
||||
row; it does not reinterpret the error as an absent runtime.
|
||||
- If worktree or task environment cleanup fails after runtime shutdown is
|
||||
confirmed, the runtime tracking row can still be removed because it no longer
|
||||
identifies a live process. The resource cleanup error is logged and handled by
|
||||
@@ -241,6 +275,29 @@ The durable cleanup job wraps that resource lifecycle:
|
||||
- **GIVEN** the backend restarts with an `executors_running` row for an archived
|
||||
task, **WHEN** startup reconciliation runs, **THEN** it attempts cleanup for the
|
||||
row instead of treating the archived task as active.
|
||||
- **GIVEN** an `executors_running` row whose task session is missing and whose
|
||||
local liveness handle refers to a dead host process, **WHEN** startup
|
||||
reconciliation stops it and the stop reports the execution is not found, **THEN**
|
||||
the row is pruned or repaired under the resume-safety invariant instead of being
|
||||
preserved indefinitely.
|
||||
- **GIVEN** a resumed durable cleanup job whose only remaining runtime target is a
|
||||
confirmed-dead local runtime, **WHEN** the cleanup worker stops it and the stop
|
||||
reports not found, **THEN** the job records a successful stop, completes owned
|
||||
resource teardown, and does not re-enter `retry_wait` because of that runtime.
|
||||
- **GIVEN** an `executors_running` row that is still alive on this host, **WHEN**
|
||||
a stop reports not found for it, **THEN** the row is preserved and the outcome is
|
||||
treated as retryable rather than pruned.
|
||||
- **GIVEN** an `executors_running` row for a remote (SSH) or containerized runtime
|
||||
whose local liveness is Unknown, **WHEN** a stop reports not found, **THEN** the
|
||||
row is preserved with its remote handle and resume token intact and is not
|
||||
pruned on the basis of a host-local not-found.
|
||||
- **GIVEN** a session lookup during a stop fails with a non-not-found store error,
|
||||
**WHEN** cleanup evaluates the result, **THEN** it preserves the row and retries
|
||||
rather than treating the runtime as absent.
|
||||
- **GIVEN** a confirmed-dead local row that still carries a resume token, **WHEN**
|
||||
a not-found stop is reclassified as successful, **THEN** the row is repaired in
|
||||
place (token and worktree preserved) rather than deleted, per
|
||||
`RowMustBePreserved`.
|
||||
- **GIVEN** agentctl is stopped while an ACP child process ignores stdin EOF,
|
||||
**WHEN** the stop timeout expires, **THEN** the ACP process group is killed and
|
||||
no ACP child is reparented to PID 1.
|
||||
|
||||
@@ -17,6 +17,8 @@ Long task titles crowd navigation and task chrome, especially when a remote pull
|
||||
- When a remote pull request, merge request, or issue supplies a title longer than 60 characters, the task dialog uses the first 59 characters followed by `…`. The full remote title remains available in the generated task description or linked remote item; only the task title is shortened.
|
||||
- The `create_task_kandev` MCP tool advertises and enforces the 60-character limit and asks agents to use a concise, few-word title.
|
||||
- Backend task creation and title updates reject titles longer than 60 characters regardless of whether the caller uses HTTP, WebSocket, MCP, Office, automation, or another service adapter.
|
||||
- The GitHub review-PR and GitHub issue watchers, which construct `PR #<n>: <title>` and `Issue #<n>: <title>` respectively, shorten the complete generated title (including the `PR #<n>:` or `Issue #<n>:` prefix) to at most 60 characters before it reaches task creation. A title that is shortened ends with a single `…` and never exceeds 60 runes counted by Unicode code point, so multibyte titles are not split mid-character. Because backend task creation rejects overlong titles, a watcher that skips this shortening drops the review/issue task entirely instead of creating a shortened one — this is the concrete regression this limit fixes for the GitHub paths.
|
||||
- Every other backend watcher that turns a remote item into a task through the shared watcher-dispatch pipeline — Linear (`[<identifier>] <title>`), Jira (`[<key>] <summary>`), GitLab merge-request and issue (`[<path>!<n>]`/`[<path>#<n>] <title>`), and Azure DevOps work-item and pull-request (`[<project>#<n>] <title>`) — has its complete generated title shortened to at most 60 runes (ending with a single `…` when shortened, counted by Unicode code point) before it reaches task creation. The dispatch coordinator applies this once at the shared seam after each source builds its request, so a source cannot skip it and every current and future watcher source inherits the guarantee automatically. The automation-trigger path likewise routes its generated title through the shared truncation helper. New watcher paths that auto-generate a title from a remote item are covered by the coordinator with no per-source work.
|
||||
- Existing stored titles longer than 60 characters remain readable and are not rewritten. Updates that omit `title` continue to work; any submitted replacement title must satisfy the new limit.
|
||||
|
||||
## API surface
|
||||
@@ -41,6 +43,9 @@ Long task titles crowd navigation and task chrome, especially when a remote pull
|
||||
- **GIVEN** an MCP caller supplies a title longer than 60 characters, **WHEN** it calls `create_task_kandev`, **THEN** the call returns a validation error and creates no task.
|
||||
- **GIVEN** an HTTP or WebSocket caller supplies an overlong title for task creation or rename, **WHEN** the request is handled, **THEN** it returns the channel's validation response and leaves stored data unchanged.
|
||||
- **GIVEN** an existing task with a legacy title longer than 60 characters, **WHEN** a caller updates another task field without submitting `title`, **THEN** the existing title remains unchanged.
|
||||
- **GIVEN** a GitHub review watcher observes a pull request whose `PR #<n>: <title>` string exceeds 60 characters, **WHEN** the review task is created, **THEN** the generated title is shortened to at most 60 runes ending with `…`, the `PR #<n>:` prefix is retained, and the review task is created rather than dropped.
|
||||
- **GIVEN** a GitHub issue watcher observes an issue whose `Issue #<n>: <title>` string exceeds 60 characters, **WHEN** the issue task is created, **THEN** the generated title is shortened to at most 60 runes ending with `…`, the `Issue #<n>:` prefix is retained, and the issue task is created rather than dropped.
|
||||
- **GIVEN** a watcher-generated title contains multibyte characters near the 60-rune boundary, **WHEN** it is shortened, **THEN** the result is counted by Unicode code point and no character is split.
|
||||
|
||||
## Out of scope
|
||||
|
||||
|
||||
Reference in New Issue
Block a user