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

2 files changed, 498 insertions, 16 deletions

diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go
index b3756fc..6df854b 100644
--- a/internal/scheduler/scheduler.go
+++ b/internal/scheduler/scheduler.go
@@ -6,10 +6,18 @@
 // event.KindEscalated event either way. If the ladder is exhausted or the
 // budget denies the escalation, the task is left FAILED for human attention.
 //
-// Explicit non-goals for this phase (see the Phase 5 task description):
-// no AskUser-timeout escalation, no DAG/cascade-fail logic. Handling for
-// TIMED_OUT/CANCELLED/BUDGET_EXCEEDED tasks follows the same shape as FAILED
-// but isn't implemented yet — only FAILED is polled.
+// Phase 7c extended this same "watch for stuck role-typed tasks and take
+// escalation action" responsibility to a second trigger: a role-typed task
+// BLOCKED on an ask_user question nobody has answered within
+// SchedulerConfig.AskUserTimeoutSeconds (see tickAskUserTimeouts below). This
+// is a natural extension of what Scheduler already does, not a new
+// component — it owns retry/escalation for role-typed tasks generally, and a
+// stuck question is just another way a role-typed task gets stuck.
+//
+// Explicit non-goal still remaining: no DAG/cascade-fail logic here (that's
+// executor.Pool.cascadeFail's job). Handling for TIMED_OUT/CANCELLED/
+// BUDGET_EXCEEDED tasks follows the same shape as FAILED but isn't
+// implemented yet — only FAILED is polled for the failure-retry path.
 package scheduler
 
 import (
@@ -19,6 +27,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/thepeterstone/claudomator/internal/event"
 	"github.com/thepeterstone/claudomator/internal/role"
 	"github.com/thepeterstone/claudomator/internal/storage"
@@ -33,6 +42,14 @@ type Store interface {
 	UpdateTaskAgent(id string, agent task.AgentConfig) error
 	UpdateTaskState(id string, newState task.State) error
 	CreateEvent(e *event.Event) error
+	// UpdateTaskQuestion, AppendTaskInteraction, and UpdateTaskNeedsReview
+	// back tickAskUserTimeouts' resume-with-fallback-answer flow (Phase 7c):
+	// clearing the stale question, recording the system-authored answer as an
+	// interaction (mirroring api.answerTaskQuestion's audit trail for a real
+	// human answer), and flagging the task for later human review.
+	UpdateTaskQuestion(taskID, questionJSON string) error
+	AppendTaskInteraction(taskID string, interaction task.Interaction) error
+	UpdateTaskNeedsReview(id string, needsReview bool) error
 }
 
 // Pool is the subset of *executor.Pool the Scheduler needs. Satisfied by
@@ -41,6 +58,11 @@ type Store interface {
 // executor package's runner/sandbox machinery.
 type Pool interface {
 	Submit(ctx context.Context, t *task.Task) error
+	// SubmitResume re-queues a BLOCKED (or otherwise interrupted) task using
+	// a resume execution carrying ResumeSessionID/ResumeAnswer. Used by
+	// tickAskUserTimeouts to resume a task with a system-authored fallback
+	// answer the same way api.answerTaskQuestion resumes it with a real one.
+	SubmitResume(ctx context.Context, t *task.Task, exec *storage.Execution) error
 }
 
 // BudgetGate reports whether an escalation to provider estimated at estCost
@@ -50,13 +72,22 @@ type BudgetGate interface {
 }
 
 // Scheduler polls for role-typed FAILED tasks and retries or escalates them
-// per their active role_configs escalation ladder.
+// per their active role_configs escalation ladder. It also polls for
+// role-typed tasks BLOCKED on a stale ask_user question (see
+// tickAskUserTimeouts).
 type Scheduler struct {
 	Store  Store
 	Pool   Pool
 	Budget BudgetGate // nil means "no budget gating" (always allow)
 	Logger *slog.Logger
 
+	// AskUserTimeout is how long a role-typed task may sit BLOCKED on an
+	// unanswered ask_user question before tickAskUserTimeouts resumes it with
+	// a system-authored fallback answer. <= 0 means DefaultAskUserTimeout
+	// (see askUserTimeout()); set from config.SchedulerConfig.AskUserTimeout()
+	// in production (internal/cli/serve.go).
+	AskUserTimeout time.Duration
+
 	// handled dedupes processing within a single running process: once a
 	// decision (retry/escalate/decline) has been made for a given
 	// execution ID, it is never reconsidered again by this Scheduler
@@ -68,6 +99,13 @@ type Scheduler struct {
 	// this map, so a task can be reconsidered once more after a restart —
 	// intentional: it's an idempotent bookkeeping decision, not orchestration
 	// state, so re-deriving it once is harmless.
+	//
+	// tickAskUserTimeouts does NOT need an equivalent guard: successfully
+	// resuming a task moves it out of BLOCKED (to QUEUED), which structurally
+	// removes it from the next tick's BLOCKED query — the same
+	// "idempotency via a state the next poll won't rediscover" pattern
+	// StoryOrchestrator's ensureEvaluators/ensureArbitration use, rather than
+	// an in-memory marker.
 	mu      sync.Mutex
 	handled map[string]bool
 }
@@ -75,6 +113,30 @@ type Scheduler struct {
 // DefaultPollInterval is used by Run when pollInterval <= 0.
 const DefaultPollInterval = 30 * time.Second
 
+// DefaultAskUserTimeout is used when Scheduler.AskUserTimeout <= 0. Mirrors
+// config.SchedulerConfig.AskUserTimeout's default and reasoning (10 minutes:
+// long enough for an actively-working human to notice and answer a
+// clarification request, short enough that a role-typed task doesn't stall
+// for hours on input that may never come — the common case for a
+// self-hosted, typically single-operator deployment).
+const DefaultAskUserTimeout = 10 * time.Minute
+
+// askUserTimeout returns the effective ask-user timeout, defaulting to
+// DefaultAskUserTimeout when unset.
+func (s *Scheduler) askUserTimeout() time.Duration {
+	if s.AskUserTimeout <= 0 {
+		return DefaultAskUserTimeout
+	}
+	return s.AskUserTimeout
+}
+
+// fallbackAnswer is the system-authored answer injected into a role-typed
+// task's ask_user question once it has gone unanswered for longer than
+// askUserTimeout(). Clearly marked as a system fallback, not a real human
+// answer, so anyone reading the task's interaction history or event stream
+// later understands why the agent proceeded without a real decision.
+const fallbackAnswer = "[auto-escalated: no human response within timeout; proceeding with best judgment]"
+
 // Run polls for role-typed FAILED tasks every pollInterval until ctx is
 // cancelled.
 func (s *Scheduler) Run(ctx context.Context, pollInterval time.Duration) {
@@ -93,20 +155,22 @@ func (s *Scheduler) Run(ctx context.Context, pollInterval time.Duration) {
 	}
 }
 
-// Tick runs a single poll pass. Exported so tests can drive it directly
+// Tick runs a single poll pass: FAILED-task retry/escalation, then
+// ask_user-timeout escalation. Exported so tests can drive it directly
 // without waiting on a ticker.
 func (s *Scheduler) Tick(ctx context.Context) {
 	tasks, err := s.Store.ListTasks(storage.TaskFilter{State: task.StateFailed})
 	if err != nil {
 		s.logf("scheduler: list failed tasks", "error", err)
-		return
-	}
-	for _, t := range tasks {
-		if t.Agent.Role == "" {
-			continue
+	} else {
+		for _, t := range tasks {
+			if t.Agent.Role == "" {
+				continue
+			}
+			s.processTask(ctx, t)
 		}
-		s.processTask(ctx, t)
 	}
+	s.tickAskUserTimeouts(ctx)
 }
 
 func (s *Scheduler) logf(msg string, args ...any) {
@@ -240,7 +304,7 @@ func (s *Scheduler) escalate(ctx context.Context, t *task.Task, fromRung, toRung
 		s.logf("scheduler: escalate: update task state", "taskID", t.ID, "error", err)
 		return
 	}
-	s.emitEscalated(t.ID, fromRung, toRung, fromProvider, target.Provider, false, "")
+	s.emitEscalated(t.ID, fromRung, toRung, fromProvider, target.Provider, false, "", "failure")
 
 	resubmit := *t
 	resubmit.Agent = newAgent
@@ -254,10 +318,15 @@ func (s *Scheduler) escalate(ctx context.Context, t *task.Task, fromRung, toRung
 // (ladder exhausted or budget denied) and leaves it FAILED for human
 // attention.
 func (s *Scheduler) decline(ctx context.Context, t *task.Task, atRung int, consideredProvider, reason string) {
-	s.emitEscalated(t.ID, atRung, atRung, t.Agent.Type, consideredProvider, true, reason)
+	s.emitEscalated(t.ID, atRung, atRung, t.Agent.Type, consideredProvider, true, reason, "failure")
 }
 
-func (s *Scheduler) emitEscalated(taskID string, fromRung, toRung int, fromProvider, toProvider string, final bool, reason string) {
+// emitEscalated records an event.KindEscalated event. trigger distinguishes
+// what caused the scheduler to reconsider this task's tier: "failure" for
+// the FAILED-task retry/escalation path above, "ask_user_timeout" for
+// escalateAskUserTimeout below — so someone reading the event stream later
+// can tell a stuck-question escalation from a stuck-failure one.
+func (s *Scheduler) emitEscalated(taskID string, fromRung, toRung int, fromProvider, toProvider string, final bool, reason, trigger string) {
 	payload, _ := json.Marshal(struct {
 		FromRung     int    `json:"from_rung"`
 		ToRung       int    `json:"to_rung"`
@@ -265,7 +334,8 @@ func (s *Scheduler) emitEscalated(taskID string, fromRung, toRung int, fromProvi
 		ToProvider   string `json:"to_provider,omitempty"`
 		Final        bool   `json:"final"`
 		Reason       string `json:"reason,omitempty"`
-	}{FromRung: fromRung, ToRung: toRung, FromProvider: fromProvider, ToProvider: toProvider, Final: final, Reason: reason})
+		Trigger      string `json:"trigger,omitempty"`
+	}{FromRung: fromRung, ToRung: toRung, FromProvider: fromProvider, ToProvider: toProvider, Final: final, Reason: reason, Trigger: trigger})
 	if err := s.Store.CreateEvent(&event.Event{
 		TaskID:  taskID,
 		Kind:    event.KindEscalated,
@@ -275,3 +345,162 @@ func (s *Scheduler) emitEscalated(taskID string, fromRung, toRung int, fromProvi
 		s.logf("scheduler: emit escalated event", "taskID", taskID, "error", err)
 	}
 }
+
+// tickAskUserTimeouts finds role-typed tasks BLOCKED on an ask_user question
+// that has been outstanding longer than askUserTimeout() and resumes each
+// with a system-authored fallback answer, escalating to the next tier of the
+// role's ladder where possible so a different (hopefully more capable)
+// provider/model picks up where the original agent stalled.
+//
+// The "outstanding since" timestamp is task.UpdatedAt, not a new column:
+// storage.DB.UpdateTaskQuestion (the last write made to a task's row on its
+// way into BLOCKED — see executor.Pool.handleRunResult's BlockedError
+// branch, which calls UpdateTaskState(BLOCKED) then
+// UpdateTaskQuestion(questionJSON) in that order) stamps updated_at at the
+// exact moment the question was recorded, and nothing else touches the row
+// while it sits BLOCKED awaiting an answer. Reusing it avoids an entirely
+// redundant "asked_at" column carrying the same information a second time.
+func (s *Scheduler) tickAskUserTimeouts(ctx context.Context) {
+	tasks, err := s.Store.ListTasks(storage.TaskFilter{State: task.StateBlocked})
+	if err != nil {
+		s.logf("scheduler: list blocked tasks", "error", err)
+		return
+	}
+	timeout := s.askUserTimeout()
+	for _, t := range tasks {
+		if t.Agent.Role == "" || t.QuestionJSON == "" {
+			// Not role-typed, or BLOCKED on pending subtasks rather than a
+			// question (see task.go's state machine: BLOCKED covers both).
+			continue
+		}
+		if time.Since(t.UpdatedAt) < timeout {
+			continue // still within the grace period
+		}
+		s.escalateAskUserTimeout(ctx, t)
+	}
+}
+
+// escalateAskUserTimeout resumes a single BLOCKED, role-typed task whose
+// question has timed out. It resolves the role's escalation ladder from the
+// tier the task was dispatched at (latest execution's EscalationRung) and
+// picks the next tier up, mirroring processTask's escalate() above but
+// applied to "stuck on a question" rather than "stuck on a failure" — if no
+// higher tier exists (ladder exhausted, no active role config, etc.) it
+// still resumes the task (unblocking it is the priority) at its current
+// tier, just without a provider/model change, and marks the escalation
+// event final:true so that distinction is visible in the event stream.
+func (s *Scheduler) escalateAskUserTimeout(ctx context.Context, t *task.Task) {
+	execs, err := s.Store.ListExecutions(t.ID)
+	if err != nil || len(execs) == 0 {
+		s.logf("scheduler: ask-user-timeout: no executions", "taskID", t.ID)
+		return
+	}
+	latest := execs[0] // ListExecutions orders DESC by start_time.
+	if latest.SessionID == "" {
+		s.logf("scheduler: ask-user-timeout: no resumable session", "taskID", t.ID)
+		return
+	}
+
+	currentRung := latest.EscalationRung
+	if currentRung < 0 {
+		currentRung = 0
+	}
+
+	// Captured before any Store mutation below: fakeStore-style test doubles
+	// (and, in principle, a caching Store) may hand back the same *task.Task
+	// pointer from ListTasks that UpdateTaskAgent then mutates in place, so
+	// reading t.Agent.Type *after* that call would silently pick up the new
+	// value instead of the original one — the same "read fromProvider before
+	// mutating" care processTask's escalate() takes above.
+	fromProvider := t.Agent.Type
+
+	newAgent := t.Agent
+	toRung := currentRung
+	toProvider := fromProvider
+	reason := ""
+	final := false
+
+	row, rcErr := s.Store.GetActiveRoleConfig(t.Agent.Role)
+	if rcErr != nil {
+		reason = "no active role config; resuming at same tier"
+		final = true
+	} else {
+		var rc role.RoleConfig
+		if jsonErr := json.Unmarshal([]byte(row.ConfigJSON), &rc); jsonErr != nil {
+			reason = "failed to decode role config; resuming at same tier"
+			final = true
+		} else if len(rc.EscalationLadder) == 0 {
+			reason = "empty escalation ladder; resuming at same tier"
+			final = true
+		} else {
+			nextRung := currentRung + 1
+			if nextRung >= len(rc.EscalationLadder) || len(rc.EscalationLadder[nextRung].Candidates) == 0 {
+				reason = "escalation ladder exhausted; resuming at same tier"
+				final = true
+			} else {
+				target := rc.EscalationLadder[nextRung].Candidates[0]
+				newAgent.Type = target.Provider
+				newAgent.Model = target.Model
+				toRung = nextRung
+				toProvider = target.Provider
+			}
+		}
+	}
+
+	// Record the system-authored fallback answer as an interaction before
+	// clearing the question, mirroring the audit trail api.answerTaskQuestion
+	// leaves for a real human answer.
+	if t.QuestionJSON != "" {
+		var qData struct {
+			Text    string   `json:"text"`
+			Options []string `json:"options"`
+		}
+		if json.Unmarshal([]byte(t.QuestionJSON), &qData) == nil {
+			if err := s.Store.AppendTaskInteraction(t.ID, task.Interaction{
+				QuestionText: qData.Text,
+				Options:      qData.Options,
+				Answer:       fallbackAnswer,
+				AskedAt:      t.UpdatedAt,
+			}); err != nil {
+				s.logf("scheduler: ask-user-timeout: append interaction", "taskID", t.ID, "error", err)
+			}
+		}
+	}
+	if err := s.Store.UpdateTaskQuestion(t.ID, ""); err != nil {
+		s.logf("scheduler: ask-user-timeout: clear question", "taskID", t.ID, "error", err)
+		return
+	}
+	if newAgent.Type != fromProvider || newAgent.Model != t.Agent.Model {
+		if err := s.Store.UpdateTaskAgent(t.ID, newAgent); err != nil {
+			s.logf("scheduler: ask-user-timeout: update task agent", "taskID", t.ID, "error", err)
+			return
+		}
+	}
+	if err := s.Store.UpdateTaskState(t.ID, task.StateQueued); err != nil {
+		s.logf("scheduler: ask-user-timeout: update task state", "taskID", t.ID, "error", err)
+		return
+	}
+	if err := s.Store.UpdateTaskNeedsReview(t.ID, true); err != nil {
+		s.logf("scheduler: ask-user-timeout: mark needs_review", "taskID", t.ID, "error", err)
+	}
+
+	s.emitEscalated(t.ID, currentRung, toRung, fromProvider, toProvider, final, reason, "ask_user_timeout")
+
+	// SubmitResume requires the task passed in to carry a resumable State
+	// (see executor.resumablePoolStates, which includes BLOCKED) — mirroring
+	// api.answerTaskQuestion, which passes the pre-transition BLOCKED task
+	// struct even though the DB row has already moved to QUEUED above.
+	resume := *t
+	resume.Agent = newAgent
+	resume.State = task.StateBlocked
+	resumeExec := &storage.Execution{
+		ID:              uuid.NewString(),
+		TaskID:          t.ID,
+		ResumeSessionID: latest.SessionID,
+		ResumeAnswer:    fallbackAnswer,
+		SandboxDir:      latest.SandboxDir,
+	}
+	if err := s.Pool.SubmitResume(ctx, &resume, resumeExec); err != nil {
+		s.logf("scheduler: ask-user-timeout: submit resume", "taskID", t.ID, "error", err)
+	}
+}
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)
+	}
+}