weichenandCursor
aea391cdaf
Add goose v19 task_runs/task_run_events tables with PG store for restart replay, orphan interruption on startup, strict /ready checks, and strip client work_dir from chat/SSE paths per P0 path sandbox requirements. Co-authored-by: Cursor <cursoragent@cursor.com>
106 lines
2.7 KiB
Go
106 lines
2.7 KiB
Go
package taskrun
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"testing"
|
|
"time"
|
|
|
|
"chat-agent-system/internal/bizdev"
|
|
"chat-agent-system/internal/stream"
|
|
)
|
|
|
|
func TestManager_DoneEmittedAfterFinalize(t *testing.T) {
|
|
m := NewManager(nil, nil)
|
|
finalizeCalled := false
|
|
|
|
m.SetExecutor(
|
|
func(_ context.Context, _ string, _ bizdev.TaskRunRequest, _ *stream.Streamer) (*bizdev.TaskRunResult, error) {
|
|
return &bizdev.TaskRunResult{Success: true, WorkDir: "/tmp/ws"}, nil
|
|
},
|
|
nil,
|
|
nil,
|
|
)
|
|
m.SetFinalizer(func(_ context.Context, _ *RunRecord, _ *bizdev.TaskRunResult) (string, error) {
|
|
finalizeCalled = true
|
|
time.Sleep(20 * time.Millisecond)
|
|
return "completed", nil
|
|
})
|
|
|
|
runID := "run_finalize_order"
|
|
record := &RunRecord{RunID: runID, Status: "running"}
|
|
_, err := m.CreateAndStart(context.Background(), runID, record, bizdev.TaskRunRequest{TaskKey: "intake"}, bizdev.StageRunRequest{}, "")
|
|
if err != nil {
|
|
t.Fatalf("CreateAndStart: %v", err)
|
|
}
|
|
|
|
deadline := time.Now().Add(3 * time.Second)
|
|
for time.Now().Before(deadline) {
|
|
rec, getErr := m.GetRun(runID)
|
|
if getErr == nil && rec.Status != "running" {
|
|
break
|
|
}
|
|
time.Sleep(20 * time.Millisecond)
|
|
}
|
|
|
|
m.mu.RLock()
|
|
ar := m.runs[runID]
|
|
m.mu.RUnlock()
|
|
if ar == nil {
|
|
t.Fatal("active run evicted too early")
|
|
}
|
|
ar.mu.RLock()
|
|
defer ar.mu.RUnlock()
|
|
if !finalizeCalled {
|
|
t.Fatal("finalize was not called")
|
|
}
|
|
if len(ar.events) == 0 {
|
|
t.Fatal("expected events")
|
|
}
|
|
last := ar.events[len(ar.events)-1]
|
|
if last.Event.Type != stream.EventDone {
|
|
t.Fatalf("last event type=%s want done", last.Event.Type)
|
|
}
|
|
}
|
|
|
|
func TestManager_FinalizeFailureDoesNotEmitDone(t *testing.T) {
|
|
m := NewManager(nil, nil)
|
|
m.SetExecutor(
|
|
func(_ context.Context, _ string, _ bizdev.TaskRunRequest, _ *stream.Streamer) (*bizdev.TaskRunResult, error) {
|
|
return &bizdev.TaskRunResult{Success: true}, nil
|
|
},
|
|
nil,
|
|
nil,
|
|
)
|
|
m.SetFinalizer(func(_ context.Context, _ *RunRecord, _ *bizdev.TaskRunResult) (string, error) {
|
|
return "", errors.New("db write failed")
|
|
})
|
|
|
|
runID := "run_finalize_fail"
|
|
record := &RunRecord{RunID: runID, Status: "running"}
|
|
_, _ = m.CreateAndStart(context.Background(), runID, record, bizdev.TaskRunRequest{TaskKey: "intake"}, bizdev.StageRunRequest{}, "")
|
|
|
|
deadline := time.Now().Add(3 * time.Second)
|
|
for time.Now().Before(deadline) {
|
|
rec, getErr := m.GetRun(runID)
|
|
if getErr == nil && rec.Status == "finalizing_failed" {
|
|
m.mu.RLock()
|
|
ar := m.runs[runID]
|
|
m.mu.RUnlock()
|
|
if ar == nil {
|
|
t.Fatal("missing active run")
|
|
}
|
|
ar.mu.RLock()
|
|
defer ar.mu.RUnlock()
|
|
for _, ev := range ar.events {
|
|
if ev.Event.Type == stream.EventDone {
|
|
t.Fatal("done must not be emitted when finalize fails")
|
|
}
|
|
}
|
|
return
|
|
}
|
|
time.Sleep(20 * time.Millisecond)
|
|
}
|
|
t.Fatal("expected finalizing_failed status")
|
|
}
|