test(office): prove a dropped children-completed wake gets recovered ↗

One-shot failure dispatcher

This helper fails the first on_children_completed dispatch and records every call so the test can compare operation ids.

type oneShotFailureDispatcher struct {
	inner       *queueRunDispatcher
	failTrigger engine.Trigger
	failArmed   bool
	failErr     error

	mu    sync.Mutex
	calls []dispatcherCall
}

func (d *oneShotFailureDispatcher) HandleTrigger(
	ctx context.Context, taskID string, trigger engine.Trigger, payload any, opID string,
) error {
	d.mu.Lock()
	d.calls = append(d.calls, dispatcherCall{taskID, trigger, payload, opID})
	d.mu.Unlock()
	if d.failArmed && trigger == d.failTrigger {
		d.failArmed = false
		return d.failErr
	}
	return d.inner.HandleTrigger(ctx, taskID, trigger, payload, opID)
}

Counting dispatcher

This helper counts dispatch attempts so the control test can prove the reconciler never tries to dispatch behind a successful edge delivery.

type countingDispatcher struct {
	inner *queueRunDispatcher

	mu sync.Mutex
	n  int
}

func (d *countingDispatcher) HandleTrigger(
	ctx context.Context, taskID string, trigger engine.Trigger, payload any, opID string,
) error {
	d.mu.Lock()
	d.n++
	d.mu.Unlock()
	return d.inner.HandleTrigger(ctx, taskID, trigger, payload, opID)
}

func (d *countingDispatcher) Count() int {
	d.mu.Lock()
	defer d.mu.Unlock()
	return d.n
}

Recovery test: failed edge is re-delivered

This test injects a failure on the edge path, then proves Tick re-delivers the same wake with the same operation id as a real run.

func TestParentWakeReconciler_RecoversFailedEdgeDispatch(t *testing.T) {
	svc, eb := newTestServiceWithBus(t)
	ctx := context.Background()

	dispatcher := &oneShotFailureDispatcher{
		inner:       &queueRunDispatcher{svc: svc},
		failTrigger: engine.TriggerOnChildrenCompleted,
		failArmed:   true,
		failErr:     fmt.Errorf("injected on_children_completed dispatch failure"),
	}
	svc.SetWorkflowEngineDispatcher(dispatcher)

	adoptOffice(t, svc, "ws-1")
	seedStuckParent(t, svc, "ws-1", "parent-1", "worker-1")

	publishChildDone(t, eb, "parent-1", "worker-1")

	calls := dispatcher.Calls()
	if len(calls) != 1 {
		t.Fatalf("want exactly one dispatch attempt after the failed edge publish, got %d", len(calls))
	}
	edgeOpID := calls[0].opID
	runs, _ := svc.ListRuns(ctx, "ws-1")
	if len(runs) != 0 {
		t.Fatalf("failed edge dispatch left a queued run: %#v", runs)
	}
}

handler := service.NewParentWakeReconciler(service.NewSchedulerIntegration(svc, 0))
	if err := handler.Tick(ctx); err != nil {
		t.Fatalf("tick: %v", err)
	}

	calls = dispatcher.Calls()
	if len(calls) != 2 {
		t.Fatalf("want a second dispatch attempt from the reconciler, got %d", len(calls))
	}
	if calls[1].opID != edgeOpID {
		t.Fatalf("reconciler recovery used a different operation id: edge=%q reconciler=%q", edgeOpID, calls[1].opID)
	}

	runsAfter, _ := svc.ListRuns(ctx, "ws-1")
	if len(runsAfter) != 1 {
		t.Fatalf("want exactly one queued run after recovery, got %d", len(runsAfter))
	}
	if runsAfter[0].Reason != service.RunReasonTaskChildrenCompleted {
		t.Fatalf("recovered run reason = %q", runsAfter[0].Reason)
	}

Control test: successful edge is not double-queued

This control uses the same setup but lets the edge succeed, then proves Tick adds nothing.

func TestParentWakeReconciler_DoesNotDoubleQueueASuccessfulEdgeDispatch(t *testing.T) {
	svc, eb := newTestServiceWithBus(t)
	ctx := context.Background()

	dispatcher := &countingDispatcher{inner: &queueRunDispatcher{svc: svc}}
	svc.SetWorkflowEngineDispatcher(dispatcher)

	adoptOffice(t, svc, "ws-1")
	seedStuckParent(t, svc, "ws-1", "parent-1", "worker-1")

	publishChildDone(t, eb, "parent-1", "worker-1")

	runs, _ := svc.ListRuns(ctx, "ws-1")
	if len(runs) != 1 {
		t.Fatalf("successful edge dispatch did not queue a run: %#v", runs)
	}

	handler := service.NewParentWakeReconciler(service.NewSchedulerIntegration(svc, 0))
	if err := handler.Tick(ctx); err != nil {
		t.Fatalf("tick: %v", err)
	}

	if dispatcher.Count() != 1 {
		t.Fatalf("want the reconciler to skip dispatch entirely, got %d attempts", dispatcher.Count())
	}
	runsAfter, _ := svc.ListRuns(ctx, "ws-1")
	if len(runsAfter) != 1 {
		t.Fatalf("tick added a run behind a successful edge dispatch: %#v", runsAfter)
	}
}

Shared event helper

This helper publishes the exact task.moved shape that triggers finalizeDone, so both tests use the same edge path.

func publishChildDone(t *testing.T, eb bus.EventBus, parentID, workerID string) {
	t.Helper()
	ctx := context.Background()
	moveEvent := bus.NewEvent("task.moved", "test", map[string]string{
		"task_id":                   parentID + "-child-0",
		"workspace_id":              "ws-1",
		"from_step_id":              "step-1",
		"to_step_id":                "step-done",
		"to_step_name":              "Done",
		"from_step_name":            "In Progress",
		"assignee_agent_profile_id": workerID,
		"parent_id":                 parentID,
	})
	if err := eb.Publish(ctx, "task.moved", moveEvent); err != nil {
		t.Fatalf("publish task.moved: %v", err)
	}
}