feat(story,scheduler): add epic-proposal tool + AskUser-timeout escalation (Phase 7c)

Two independent pieces, completing Phase 7.

Epic-proposal tool: AgentChannel gains a 5th method, ProposeEpic(ctx,
EpicProposal{Name, Description, StoryIDs}) (epicID, err), implemented on
storeChannel -- matches an existing epic by exact name or creates one
(DiscoverySource: "agent"), sets epic_id on each resolvable story (skips,
doesn't fail, on an unresolved ID), emits KindEpicProposed attached to the
epic's own ID with payload {epic_id, name, story_ids}. Wired into both
transports exactly like Phase 6 wired role into spawn_subtask: a new
propose_epic tool in the native tool-use loop (internal/agentloop/tools.go)
and the MCP transport (internal/executor/agentmcp.go). This is the mechanism
for a discovery/planner-role agent to act on its own judgment that several
stories it's been given form one cohesive initiative -- the judgment itself
lives in the calling agent's instructions/model, not in this code.

AskUser-timeout escalation: extends the existing Scheduler (Phase 5's
retry-then-escalate watcher) rather than adding a new component, since
"stuck task needs escalation" is exactly what it already does. Finds
role-typed BLOCKED tasks whose question has been outstanding longer than
SchedulerConfig.AskUserTimeoutSeconds (default 10 minutes) using
task.UpdatedAt as the outstanding-since timestamp -- no new column needed,
since UpdateTaskQuestion already stamps it the instant a question is
recorded and nothing else touches the row while BLOCKED. Resolves the next
ladder tier from the latest execution's EscalationRung, records the
system-authored fallback answer as an audit-trail task.Interaction, clears
the question, sets the new tasks.needs_review flag, emits KindEscalated
(now carrying a trigger field: "failure" vs "ask_user_timeout" for the
existing failure-retry path vs this one), and resumes via Pool.SubmitResume
at the escalated tier -- degrading to same-tier resume with final:true if
the ladder's exhausted or no role config exists, since unblocking the task
takes priority over having somewhere higher to escalate to.
GET /api/tasks?needs_review=true surfaces auto-decided tasks for human
review.

go build/vet/test -race -count=1 all pass, full suite (20 packages), run
twice to rule out flakiness in the new tests. (One pre-existing, unrelated
test -- TestHandleRunTask_CascadesRetryToFailedDeps, a tempdir-cleanup race
-- appeared once under full-suite load per the implementing agent's report
and did not reproduce in this verification's runs either; not a regression
from this work.)

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01V1moSNCJRcP6kykA4tyUSs

1 files changed, 253 insertions, 0 deletions

diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go
index 18c9e42..258bb5d 100644
--- a/internal/scheduler/scheduler_test.go
+++ b/internal/scheduler/scheduler_test.go
@@ -25,6 +25,11 @@ type fakeStore struct {
 	agentUpdates []task.AgentConfig
 	stateUpdates []task.State
 	events       []*event.Event
+
+	// ask-user-timeout fakes (Phase 7c).
+	questionUpdates    []string
+	interactions       []task.Interaction
+	needsReviewUpdates []bool
 }
 
 func newFakeStore() *fakeStore {
@@ -92,6 +97,36 @@ func (f *fakeStore) CreateEvent(e *event.Event) error {
 	return nil
 }
 
+func (f *fakeStore) UpdateTaskQuestion(taskID, questionJSON string) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.questionUpdates = append(f.questionUpdates, questionJSON)
+	if t, ok := f.tasks[taskID]; ok {
+		t.QuestionJSON = questionJSON
+	}
+	return nil
+}
+
+func (f *fakeStore) AppendTaskInteraction(taskID string, interaction task.Interaction) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.interactions = append(f.interactions, interaction)
+	if t, ok := f.tasks[taskID]; ok {
+		t.Interactions = append(t.Interactions, interaction)
+	}
+	return nil
+}
+
+func (f *fakeStore) UpdateTaskNeedsReview(id string, needsReview bool) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.needsReviewUpdates = append(f.needsReviewUpdates, needsReview)
+	if t, ok := f.tasks[id]; ok {
+		t.NeedsReview = needsReview
+	}
+	return nil
+}
+
 func (f *fakeStore) eventsOfKind(k event.Kind) []*event.Event {
 	f.mu.Lock()
 	defer f.mu.Unlock()
@@ -109,6 +144,17 @@ type fakePool struct {
 	mu        sync.Mutex
 	submitted []*task.Task
 	err       error
+
+	// resumed records every SubmitResume call (Phase 7c's ask-user-timeout
+	// path); resumeErr lets tests exercise the error-logging path.
+	resumed   []resumeCall
+	resumeErr error
+}
+
+// resumeCall captures one SubmitResume invocation's arguments.
+type resumeCall struct {
+	task *task.Task
+	exec *storage.Execution
 }
 
 func (f *fakePool) Submit(_ context.Context, t *task.Task) error {
@@ -118,6 +164,13 @@ func (f *fakePool) Submit(_ context.Context, t *task.Task) error {
 	return f.err
 }
 
+func (f *fakePool) SubmitResume(_ context.Context, t *task.Task, exec *storage.Execution) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.resumed = append(f.resumed, resumeCall{task: t, exec: exec})
+	return f.resumeErr
+}
+
 func (f *fakePool) submitCount() int {
 	f.mu.Lock()
 	defer f.mu.Unlock()
@@ -178,6 +231,30 @@ func failedExec(rung int) *storage.Execution {
 	}
 }
 
+// blockedTask builds a role-typed task BLOCKED on an ask_user question,
+// with UpdatedAt standing in for "outstanding since" (see
+// tickAskUserTimeouts' doc comment on why task.UpdatedAt, not a new column).
+func blockedTask(id, roleName, agentType, questionJSON string, updatedAt time.Time) *task.Task {
+	return &task.Task{
+		ID:           id,
+		Name:         "test",
+		Agent:        task.AgentConfig{Type: agentType, Role: roleName, MaxBudgetUSD: 0.5},
+		State:        task.StateBlocked,
+		QuestionJSON: questionJSON,
+		UpdatedAt:    updatedAt,
+	}
+}
+
+func blockedExec(rung int, sessionID string) *storage.Execution {
+	return &storage.Execution{
+		ID:             uuid.NewString(),
+		Status:         "BLOCKED",
+		EscalationRung: rung,
+		SessionID:      sessionID,
+		StartTime:      time.Now(),
+	}
+}
+
 // TestScheduler_RetriesSameRung_WhileUnderMaxRetries proves that a task
 // whose current rung has fewer attempts than tier.MaxRetries is resubmitted
 // at the same rung, with no escalation event and no Agent.Type/Model change.
@@ -367,3 +444,179 @@ func TestScheduler_Convergence_DoesNotReprocessSameExecution(t *testing.T) {
 		t.Fatalf("expected 0 submissions, got %d", pool.submitCount())
 	}
 }
+
+// TestScheduler_AskUserTimeout_EscalatesAndResumes proves the core
+// ask-user-timeout flow: a BLOCKED role-typed task whose question has been
+// outstanding longer than the configured timeout gets resumed at the next
+// escalation tier with a system-authored answer, needs_review gets set, and
+// a KindEscalated event with the ask-user-timeout trigger is emitted.
+func TestScheduler_AskUserTimeout_EscalatesAndResumes(t *testing.T) {
+	store := newFakeStore()
+	seedRoleConfig(t, store, twoTierLadder())
+
+	askedAt := time.Now().Add(-20 * time.Minute)
+	tk := blockedTask("t1", "coder", "local", `{"text":"Which approach?","options":["a","b"]}`, askedAt)
+	store.tasks[tk.ID] = tk
+	store.execsByTask[tk.ID] = []*storage.Execution{blockedExec(0, "sess-1")}
+
+	pool := &fakePool{}
+	sch := &Scheduler{Store: store, Pool: pool, AskUserTimeout: 10 * time.Minute}
+	sch.Tick(context.Background())
+
+	if len(pool.resumed) != 1 {
+		t.Fatalf("expected 1 resume submission, got %d", len(pool.resumed))
+	}
+	rc := pool.resumed[0]
+	if rc.task.Agent.Type != "anthropic" || rc.task.Agent.Model != "claude-sonnet-5" {
+		t.Errorf("resumed task should carry the escalated tier's provider/model: got %q/%q", rc.task.Agent.Type, rc.task.Agent.Model)
+	}
+	if rc.task.State != task.StateBlocked {
+		t.Errorf("resumed task struct should retain BLOCKED state (matches api.answerTaskQuestion's convention for SubmitResume's resumable-state check), got %v", rc.task.State)
+	}
+	if rc.exec.ResumeSessionID != "sess-1" {
+		t.Errorf("resume exec ResumeSessionID: got %q want sess-1", rc.exec.ResumeSessionID)
+	}
+	if rc.exec.ResumeAnswer != fallbackAnswer {
+		t.Errorf("resume exec ResumeAnswer: got %q want the fallback answer marker", rc.exec.ResumeAnswer)
+	}
+
+	if len(store.needsReviewUpdates) != 1 || !store.needsReviewUpdates[0] {
+		t.Errorf("expected needs_review to be set true exactly once, got %+v", store.needsReviewUpdates)
+	}
+	if len(store.interactions) != 1 || store.interactions[0].Answer != fallbackAnswer || store.interactions[0].QuestionText != "Which approach?" {
+		t.Errorf("expected the fallback answer recorded as an interaction, got %+v", store.interactions)
+	}
+	if len(store.questionUpdates) != 1 || store.questionUpdates[0] != "" {
+		t.Errorf("expected the question to be cleared, got %+v", store.questionUpdates)
+	}
+	if len(store.agentUpdates) != 1 {
+		t.Fatalf("expected 1 UpdateTaskAgent call, got %d", len(store.agentUpdates))
+	}
+
+	evs := store.eventsOfKind(event.KindEscalated)
+	if len(evs) != 1 {
+		t.Fatalf("expected 1 KindEscalated event, got %d", len(evs))
+	}
+	var payload struct {
+		FromRung     int    `json:"from_rung"`
+		ToRung       int    `json:"to_rung"`
+		FromProvider string `json:"from_provider"`
+		ToProvider   string `json:"to_provider"`
+		Final        bool   `json:"final"`
+		Trigger      string `json:"trigger"`
+	}
+	if err := json.Unmarshal(evs[0].Payload, &payload); err != nil {
+		t.Fatalf("unmarshal event payload: %v", err)
+	}
+	if payload.Trigger != "ask_user_timeout" {
+		t.Errorf("expected trigger=ask_user_timeout, got %q", payload.Trigger)
+	}
+	if payload.Final {
+		t.Errorf("escalating to a higher tier should not be final")
+	}
+	if payload.FromRung != 0 || payload.ToRung != 1 || payload.FromProvider != "local" || payload.ToProvider != "anthropic" {
+		t.Errorf("expected escalation local(rung0) -> anthropic(rung1), got %+v", payload)
+	}
+}
+
+// TestScheduler_AskUserTimeout_WithinWindow_LeftAlone proves that a BLOCKED
+// task still within the configured timeout window is left completely alone:
+// no resume, no needs_review, no event.
+func TestScheduler_AskUserTimeout_WithinWindow_LeftAlone(t *testing.T) {
+	store := newFakeStore()
+	seedRoleConfig(t, store, twoTierLadder())
+
+	tk := blockedTask("t1", "coder", "local", `{"text":"Which approach?"}`, time.Now())
+	store.tasks[tk.ID] = tk
+	store.execsByTask[tk.ID] = []*storage.Execution{blockedExec(0, "sess-1")}
+
+	pool := &fakePool{}
+	sch := &Scheduler{Store: store, Pool: pool, AskUserTimeout: 10 * time.Minute}
+	sch.Tick(context.Background())
+
+	if len(pool.resumed) != 0 {
+		t.Fatalf("task within the timeout window should not be resumed, got %d resumes", len(pool.resumed))
+	}
+	if len(store.needsReviewUpdates) != 0 {
+		t.Errorf("task within the timeout window should not be flagged needs_review, got %+v", store.needsReviewUpdates)
+	}
+	if len(store.eventsOfKind(event.KindEscalated)) != 0 {
+		t.Errorf("task within the timeout window should not emit a KindEscalated event")
+	}
+	if len(store.questionUpdates) != 0 {
+		t.Errorf("task within the timeout window should not have its question cleared")
+	}
+}
+
+// TestScheduler_AskUserTimeout_IgnoresSubtaskBlocked proves that a BLOCKED
+// task with no pending question (i.e. blocked waiting on subtasks, per
+// task.go's state machine, not on ask_user) is never touched by
+// tickAskUserTimeouts, however long it's been BLOCKED.
+func TestScheduler_AskUserTimeout_IgnoresSubtaskBlocked(t *testing.T) {
+	store := newFakeStore()
+	seedRoleConfig(t, store, twoTierLadder())
+
+	tk := blockedTask("t1", "coder", "local", "", time.Now().Add(-1*time.Hour))
+	store.tasks[tk.ID] = tk
+	store.execsByTask[tk.ID] = []*storage.Execution{blockedExec(0, "sess-1")}
+
+	pool := &fakePool{}
+	sch := &Scheduler{Store: store, Pool: pool, AskUserTimeout: 10 * time.Minute}
+	sch.Tick(context.Background())
+
+	if len(pool.resumed) != 0 {
+		t.Fatalf("subtask-blocked task should not be resumed, got %d resumes", len(pool.resumed))
+	}
+}
+
+// TestScheduler_AskUserTimeout_LadderExhausted_StillResumesAtSameTier proves
+// the documented fallback for when no higher tier exists: the task is still
+// unblocked (resumed) so it doesn't stay stuck forever, but at its current
+// tier's provider/model (no UpdateTaskAgent call), and the escalation event
+// is marked final:true to distinguish "resumed without escalating" from a
+// genuine tier bump.
+func TestScheduler_AskUserTimeout_LadderExhausted_StillResumesAtSameTier(t *testing.T) {
+	store := newFakeStore()
+	seedRoleConfig(t, store, twoTierLadder())
+
+	askedAt := time.Now().Add(-20 * time.Minute)
+	tk := blockedTask("t1", "coder", "anthropic", `{"text":"Which approach?"}`, askedAt)
+	store.tasks[tk.ID] = tk
+	store.execsByTask[tk.ID] = []*storage.Execution{blockedExec(1, "sess-1")} // already at the last tier
+
+	pool := &fakePool{}
+	sch := &Scheduler{Store: store, Pool: pool, AskUserTimeout: 10 * time.Minute}
+	sch.Tick(context.Background())
+
+	if len(pool.resumed) != 1 {
+		t.Fatalf("expected the task to still be resumed even with no higher tier, got %d resumes", len(pool.resumed))
+	}
+	if len(store.agentUpdates) != 0 {
+		t.Errorf("no higher tier exists, so Agent should be unchanged: got %d UpdateTaskAgent calls", len(store.agentUpdates))
+	}
+	rc := pool.resumed[0]
+	if rc.task.Agent.Type != "anthropic" {
+		t.Errorf("resumed task should keep its current provider, got %q", rc.task.Agent.Type)
+	}
+
+	evs := store.eventsOfKind(event.KindEscalated)
+	if len(evs) != 1 {
+		t.Fatalf("expected 1 KindEscalated event, got %d", len(evs))
+	}
+	var payload struct {
+		Final   bool   `json:"final"`
+		Trigger string `json:"trigger"`
+	}
+	if err := json.Unmarshal(evs[0].Payload, &payload); err != nil {
+		t.Fatalf("unmarshal event payload: %v", err)
+	}
+	if !payload.Final {
+		t.Errorf("resuming without an escalation should be marked final=true")
+	}
+	if payload.Trigger != "ask_user_timeout" {
+		t.Errorf("expected trigger=ask_user_timeout, got %q", payload.Trigger)
+	}
+	if len(store.needsReviewUpdates) != 1 || !store.needsReviewUpdates[0] {
+		t.Errorf("expected needs_review to still be set even without an escalation, got %+v", store.needsReviewUpdates)
+	}
+}