Refresh downstream readiness after gate pass
This commit is contained in:
@@ -285,6 +285,7 @@ Behavior:
|
|||||||
- recomputes the gate for the task
|
- recomputes the gate for the task
|
||||||
- keeps the task in `verifying` while required checks are still pending
|
- keeps the task in `verifying` while required checks are still pending
|
||||||
- moves the task to `done` when all required checks pass
|
- moves the task to `done` when all required checks pass
|
||||||
|
- refreshes dependent readiness immediately when the task enters or leaves `done`, so newly unblocked work emits `task_ready` in the same flow
|
||||||
- moves the task to `failed` when one or more required checks fail
|
- moves the task to `failed` when one or more required checks fail
|
||||||
|
|
||||||
### `orch verify status`
|
### `orch verify status`
|
||||||
|
|||||||
@@ -2025,6 +2025,7 @@ func (s *OrchStore) RecordCheck(ctx context.Context, input VerifyRecordInput) (V
|
|||||||
return VerifyRecordResult{}, err
|
return VerifyRecordResult{}, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
previousTaskStatus := task.Status
|
||||||
nextStatus := task.Status
|
nextStatus := task.Status
|
||||||
switch {
|
switch {
|
||||||
case gate == nil:
|
case gate == nil:
|
||||||
@@ -2090,6 +2091,12 @@ func (s *OrchStore) RecordCheck(ctx context.Context, input VerifyRecordInput) (V
|
|||||||
attempt.UpdatedAt = now
|
attempt.UpdatedAt = now
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if nextStatus != previousTaskStatus {
|
||||||
|
if err := refreshReadyStates(ctx, tx, task.RunID, now); err != nil {
|
||||||
|
return VerifyRecordResult{}, err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if err := updateRunAggregateStatus(ctx, tx, task.RunID, now); err != nil {
|
if err := updateRunAggregateStatus(ctx, tx, task.RunID, now); err != nil {
|
||||||
return VerifyRecordResult{}, err
|
return VerifyRecordResult{}, err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,163 @@
|
|||||||
|
package store
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
dbpkg "ai-workflow-skill/packages/coord-core/db"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestRecordCheckRefreshesDependentReadyStateWhenGatePasses(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
dbPath := filepath.Join(t.TempDir(), "coord.db")
|
||||||
|
|
||||||
|
sqlDB, err := dbpkg.Open(ctx, dbPath)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("open db: %v", err)
|
||||||
|
}
|
||||||
|
defer sqlDB.Close()
|
||||||
|
|
||||||
|
if err := dbpkg.ApplyMigrations(ctx, sqlDB); err != nil {
|
||||||
|
t.Fatalf("apply migrations: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
orchStore := NewOrchStore(sqlDB)
|
||||||
|
inboxStore := NewInboxStore(sqlDB)
|
||||||
|
|
||||||
|
if _, err := orchStore.CreateRun(ctx, CreateRunInput{
|
||||||
|
RunID: "run_verify_dep_001",
|
||||||
|
Goal: "Verify gated prerequisite unlocks downstream work",
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("create run: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if _, err := orchStore.AddTask(ctx, AddTaskInput{
|
||||||
|
RunID: "run_verify_dep_001",
|
||||||
|
TaskID: "T1",
|
||||||
|
Title: "Build backend",
|
||||||
|
Summary: "Implement the gated prerequisite",
|
||||||
|
DefaultTo: "worker-a",
|
||||||
|
RequiredChecks: []string{"lint", "test"},
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("add prerequisite task: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if _, err := orchStore.AddTask(ctx, AddTaskInput{
|
||||||
|
RunID: "run_verify_dep_001",
|
||||||
|
TaskID: "T2",
|
||||||
|
Title: "Build frontend",
|
||||||
|
Summary: "Waits for backend verification",
|
||||||
|
DefaultTo: "worker-b",
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("add dependent task: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if _, err := orchStore.AddDependency(ctx, AddDependencyInput{
|
||||||
|
RunID: "run_verify_dep_001",
|
||||||
|
TaskID: "T2",
|
||||||
|
DependsOnTaskID: "T1",
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("add dependency: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
dependent, _, err := orchStore.GetTaskWithLatestAttempt(ctx, "run_verify_dep_001", "T2")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("load dependent task before verify: %v", err)
|
||||||
|
}
|
||||||
|
if dependent.Status != "planned" {
|
||||||
|
t.Fatalf("expected dependent task planned before prerequisite passes, got %q", dependent.Status)
|
||||||
|
}
|
||||||
|
|
||||||
|
dispatchResult, err := orchStore.DispatchTask(ctx, DispatchInput{
|
||||||
|
RunID: "run_verify_dep_001",
|
||||||
|
TaskID: "T1",
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("dispatch prerequisite task: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if _, err := inboxStore.ClaimThread(ctx, ClaimInput{
|
||||||
|
ThreadID: dispatchResult.Thread.ThreadID,
|
||||||
|
Agent: "worker-a",
|
||||||
|
LeaseSeconds: 300,
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("claim prerequisite thread: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if _, _, err := inboxStore.CompleteThread(ctx, CompleteInput{
|
||||||
|
ThreadID: dispatchResult.Thread.ThreadID,
|
||||||
|
Agent: "worker-a",
|
||||||
|
Summary: "Backend complete",
|
||||||
|
Body: "Ready for verification.",
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("complete prerequisite thread: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
reconcileResult, err := orchStore.ReconcileRun(ctx, "run_verify_dep_001")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("reconcile run: %v", err)
|
||||||
|
}
|
||||||
|
if len(reconcileResult.UpdatedTasks) != 1 || reconcileResult.UpdatedTasks[0].Status != "verifying" {
|
||||||
|
t.Fatalf("expected prerequisite task to enter verifying, got %#v", reconcileResult.UpdatedTasks)
|
||||||
|
}
|
||||||
|
|
||||||
|
if _, err := orchStore.RecordCheck(ctx, VerifyRecordInput{
|
||||||
|
RunID: "run_verify_dep_001",
|
||||||
|
TaskID: "T1",
|
||||||
|
CheckName: "lint",
|
||||||
|
Status: "passed",
|
||||||
|
Summary: "lint clean",
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("record first verification check: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
cursor := latestRunEventID(t, ctx, sqlDB, "run_verify_dep_001")
|
||||||
|
|
||||||
|
if _, err := orchStore.RecordCheck(ctx, VerifyRecordInput{
|
||||||
|
RunID: "run_verify_dep_001",
|
||||||
|
TaskID: "T1",
|
||||||
|
CheckName: "test",
|
||||||
|
Status: "passed",
|
||||||
|
Summary: "tests clean",
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("record final verification check: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
dependent, _, err = orchStore.GetTaskWithLatestAttempt(ctx, "run_verify_dep_001", "T2")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("load dependent task after verify: %v", err)
|
||||||
|
}
|
||||||
|
if dependent.Status != "ready" {
|
||||||
|
t.Fatalf("expected dependent task ready after prerequisite gate passes, got %q", dependent.Status)
|
||||||
|
}
|
||||||
|
|
||||||
|
events, _, found, err := orchStore.findRunEventsAfter(ctx, "run_verify_dep_001", cursor, []string{"task_ready"})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("find ready events after final verification: %v", err)
|
||||||
|
}
|
||||||
|
if !found {
|
||||||
|
t.Fatal("expected task_ready event after prerequisite gate passes")
|
||||||
|
}
|
||||||
|
if len(events) != 1 || events[0].TaskID != "T2" {
|
||||||
|
t.Fatalf("expected dependent task_ready event, got %#v", events)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func latestRunEventID(t *testing.T, ctx context.Context, db queryRower, runID string) int64 {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
var eventID int64
|
||||||
|
if err := db.QueryRowContext(
|
||||||
|
ctx,
|
||||||
|
`SELECT COALESCE(MAX(event_id), 0)
|
||||||
|
FROM events
|
||||||
|
WHERE run_id = ?`,
|
||||||
|
runID,
|
||||||
|
).Scan(&eventID); err != nil {
|
||||||
|
t.Fatalf("query latest event id for %s: %v", runID, err)
|
||||||
|
}
|
||||||
|
return eventID
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user