summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--CLAUDE.md59
-rw-r--r--cmd/spike/main.go7
-rw-r--r--internal/agentchannel/agentchannel.go21
-rw-r--r--internal/agentloop/tools.go38
-rw-r--r--internal/api/agentmcp_endpoint_test.go3
-rw-r--r--internal/api/server.go9
-rw-r--r--internal/api/server_test.go58
-rw-r--r--internal/cli/serve.go7
-rw-r--r--internal/config/config.go22
-rw-r--r--internal/executor/agentmcp.go23
-rw-r--r--internal/executor/agentmcp_test.go41
-rw-r--r--internal/executor/channel.go70
-rw-r--r--internal/executor/channel_test.go206
-rw-r--r--internal/executor/container_test.go3
-rw-r--r--internal/executor/executor.go7
-rw-r--r--internal/executor/executor_test.go7
-rw-r--r--internal/executor/nativerunner_test.go45
-rw-r--r--internal/scheduler/scheduler.go261
-rw-r--r--internal/scheduler/scheduler_test.go253
-rw-r--r--internal/storage/db.go36
-rw-r--r--internal/storage/epic.go11
-rw-r--r--internal/task/task.go6
22 files changed, 1156 insertions, 37 deletions
diff --git a/CLAUDE.md b/CLAUDE.md
index 32bca69..a728f1f 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -78,7 +78,7 @@ Config defaults to `~/.claudomator/config.toml`. Data is stored in `~/.claudomat
| `internal/agentloop` | Provider-neutral tool-use control-flow loop shared by all `NativeRunner`s |
| `internal/sandbox` | `Sandbox` interface (`HostSandbox`/`DockerSandbox`) + pre-tool-use guardrail hooks |
| `internal/role` | `RoleConfig`/`Tier`/`Rung` — per-role system prompt + provider/model escalation ladder |
-| `internal/scheduler` | `Scheduler` — polls role-typed FAILED tasks and retries/escalates them per their role's ladder; `StoryOrchestrator` (Phase 7b) — polls stories and drives Builder → Evaluators → Arbitration → REVIEW_READY |
+| `internal/scheduler` | `Scheduler` — polls role-typed FAILED tasks and retries/escalates them per their role's ladder, and (Phase 7c) role-typed BLOCKED tasks whose ask_user question has timed out; `StoryOrchestrator` (Phase 7b) — polls stories and drives Builder → Evaluators → Arbitration → REVIEW_READY |
| `internal/story` | `Epic`/`Story` — planning layer above the flat task tree; data model + CRUD only (see `internal/scheduler.StoryOrchestrator` for the orchestration that drives it) |
| `web` | Embedded static UI (`embed.go`) |
@@ -186,7 +186,7 @@ Schema is auto-migrated additively on `storage.Open()` — new columns are `ALTE
tasks: id, name, description, config_json, priority, timeout_ns, retry_json,
tags_json, depends_on_json, parent_task_id, state, rejection_comment,
question_json, summary, elaboration_input, interactions_json,
- created_at, updated_at
+ needs_review, created_at, updated_at
executions: id, task_id, start_time, end_time, exit_code, status, stdout_path,
stderr_path, artifact_dir, cost_usd, error_msg, session_id,
@@ -469,6 +469,55 @@ spawned here with no matching active `role_configs` row dispatches in the
same degraded (no role resolution) mode `Pool.execute()` already logs a
warning and falls back to for any other role-typed task.
+### Epic-proposal tool (Phase 7c)
+
+A 5th `AgentChannel` method, `ProposeEpic(ctx, agentchannel.EpicProposal{Name,
+Description, StoryIDs})`, lets a discovery/planner-role agent that has been
+handed several story IDs act on its own judgment that they form a cohesive
+initiative. `storeChannel.ProposeEpic` (`internal/executor/channel.go`)
+matches an existing epic by exact `Name` (`Store.GetEpicByName`) or creates a
+new one, then sets `epic_id` on each resolved story via `Store.GetStory` +
+`Store.UpdateStory` — a story ID that doesn't resolve is skipped, not
+fatal to the call. Emits `event.KindEpicProposed` attached to the epic's own
+ID (payload `{epic_id, name, story_ids}`, `story_ids` being only the ones
+actually grouped), matching the story orchestrator's convention of attaching
+planning-layer ceremony events to the entity they're about. Exposed as a
+`propose_epic` tool on both transports — `internal/agentloop/tools.go` (native
+tool-use loop) and `internal/executor/agentmcp.go` (MCP) — with the same
+`{name, description?, story_ids}` schema Phase 6 used for `spawn_subtask`'s
+`role` parameter. This phase only builds the mechanism; judging *whether*
+stories belong together is the calling agent's job via its instructions/model.
+
+### AskUser-timeout escalation (Phase 7c)
+
+`internal/scheduler.Scheduler` (the same retry-then-escalate watcher described
+above) also polls BLOCKED role-typed tasks whose `question_json` has been
+outstanding longer than `config.SchedulerConfig.AskUserTimeout()` (default 10
+minutes; `[scheduler] ask_user_timeout_seconds` in TOML) — `tickAskUserTimeouts`,
+called from `Tick` alongside the existing FAILED-task pass. The "outstanding
+since" timestamp is `task.UpdatedAt` itself, not a new column:
+`storage.DB.UpdateTaskQuestion` (the last write made to a task's row on its
+way into BLOCKED) already stamps `updated_at` the moment the question was
+recorded, and nothing else touches the row while it sits BLOCKED. For each
+timed-out question, the scheduler resolves the role's active escalation
+ladder from the tier the task was dispatched at (`latest execution's
+EscalationRung`) and, if a higher tier exists, escalates to its first
+candidate (mirroring `processTask`'s existing "always `Candidates[0]`"
+simplification); if not (ladder exhausted, no active role config, etc.), it
+still resumes the task at its current tier rather than leaving it stuck
+forever, marking that escalation `final: true`. Either way it records the
+system-authored answer `"[auto-escalated: no human response within timeout;
+proceeding with best judgment]"` as a `task.Interaction` (mirroring
+`api.answerTaskQuestion`'s audit trail for a real human answer), clears
+`question_json`, sets `tasks.needs_review = true`, and resumes the task via
+`Pool.SubmitResume` — the same mechanism `POST /api/tasks/{id}/answer` uses,
+just at the escalated tier's provider/model instead of the one that asked.
+Emits `event.KindEscalated` with a `trigger` field (`"failure"` for the
+existing FAILED-task path, `"ask_user_timeout"` for this one) so the two are
+distinguishable in the event stream. `GET /api/tasks?needs_review=true`
+surfaces flagged tasks for a human to double-check the system's stand-in
+answer later.
+
---
@@ -488,6 +537,10 @@ Same pattern as `task.Priority`/`RetryConfig` below: `internal/role.RoleConfig.T
`Scheduler.handled` (keyed by execution ID) prevents re-emitting a `final: true` `KindEscalated` event on every poll tick once a role-typed task's ladder is exhausted, but it's per-process and resets on restart. A restart can produce one extra "reconsideration" (and, if still exhausted/denied, one more `KindEscalated` event) — not an infinite loop, just not persisted. A future phase could persist this via a `tasks` column if that turns out to matter.
+### AskUser-timeout escalation resumes at the same tier when the ladder is exhausted
+
+The phase description for `Scheduler.escalateAskUserTimeout` didn't specify behavior when a BLOCKED task is already at its role's top escalation tier (or has no active role config at all). Leaving the task stuck BLOCKED forever seemed worse than the alternative, so it still resumes the task (unblocking it is the priority) at its *current* provider/model — no `UpdateTaskAgent` call — and marks that `KindEscalated` event `final: true` so it's visible in the event stream as "resumed without escalating" rather than a genuine tier bump. A later phase could reconsider this (e.g. leaving the task BLOCKED and only flagging it) if it turns out best-effort auto-resumption at the same tier isn't the right call for every role.
+
### StoryOrchestrator's Arbitration outcome always routes to REVIEW_READY
`internal/scheduler.StoryOrchestrator.finalizeArbitration` does not parse the
@@ -560,7 +613,7 @@ In `executor.go`, `withFailureHistory` creates a copy of the task struct (`copy
| Method | Endpoint | Description |
|--------|----------|-------------|
-| GET | `/api/tasks` | List tasks; `?state=RUNNING&since=<RFC3339>&limit=50` |
+| GET | `/api/tasks` | List tasks; `?state=RUNNING&since=<RFC3339>&limit=50&needs_review=true` |
| POST | `/api/tasks` | Create task (JSON body) |
| GET | `/api/tasks/{id}` | Get task |
| DELETE | `/api/tasks/{id}` | Delete task + subtasks + executions |
diff --git a/cmd/spike/main.go b/cmd/spike/main.go
index dd87021..7fc0bf2 100644
--- a/cmd/spike/main.go
+++ b/cmd/spike/main.go
@@ -46,6 +46,7 @@ type recordingChannel struct {
summary string
progress []string
subtasks []executor.SubtaskSpec
+ epics []executor.EpicProposal
}
func (c *recordingChannel) AskUser(_ context.Context, q string) (string, error) {
@@ -72,6 +73,12 @@ func (c *recordingChannel) RecordProgress(_ context.Context, m string) error {
c.mu.Unlock()
return nil
}
+func (c *recordingChannel) ProposeEpic(_ context.Context, spec executor.EpicProposal) (string, error) {
+ c.mu.Lock()
+ c.epics = append(c.epics, spec)
+ c.mu.Unlock()
+ return "spike-epic-1", nil
+}
func main() {
agent := "claude"
diff --git a/internal/agentchannel/agentchannel.go b/internal/agentchannel/agentchannel.go
index 42eacbc..83c5f8d 100644
--- a/internal/agentchannel/agentchannel.go
+++ b/internal/agentchannel/agentchannel.go
@@ -44,6 +44,20 @@ type SubtaskSpec struct {
Role string
}
+// EpicProposal describes an epic a discovery/planner-role agent wants to
+// group one or more stories under (internal/story.Epic/Story — the planning
+// layer above the flat task tree). Matching an existing epic is by exact
+// Name (see storeChannel.ProposeEpic in internal/executor/channel.go) — the
+// simplest reasonable de-dup, not fuzzy matching.
+type EpicProposal struct {
+ Name string
+ Description string
+ // StoryIDs are the story IDs to group under this epic. An ID that
+ // doesn't resolve to an existing story is skipped rather than failing
+ // the whole call — one bad ID shouldn't block grouping the rest.
+ StoryIDs []string
+}
+
// AgentChannel is how a Runner reports agent-originated signals to the rest of
// the system. Implementations translate these into stored artifacts and events.
// The transport by which a Runner detects these signals — post-exit files
@@ -59,6 +73,13 @@ type AgentChannel interface {
SpawnSubtask(ctx context.Context, spec SubtaskSpec) (taskID string, err error)
// RecordProgress records a free-form progress note from the agent.
RecordProgress(ctx context.Context, message string) error
+ // ProposeEpic groups spec.StoryIDs under a new or existing epic (matched
+ // by name), returning the epic's ID. Called by a discovery/planner-role
+ // agent that has independently judged a set of stories to form a
+ // cohesive initiative — the judgment itself is the calling agent's job
+ // (via its instructions/model); this just gives it a mechanism to act on
+ // it.
+ ProposeEpic(ctx context.Context, spec EpicProposal) (epicID string, err error)
}
// BlockedError is returned by Run when the agent asked the user a question and
diff --git a/internal/agentloop/tools.go b/internal/agentloop/tools.go
index fc003cb..d803c23 100644
--- a/internal/agentloop/tools.go
+++ b/internal/agentloop/tools.go
@@ -11,9 +11,11 @@ import (
"github.com/thepeterstone/claudomator/internal/sandbox"
)
-// agentToolSpecs returns the eight tools available to the loop: the four
-// agent back-channel tools (mirroring the MCP tools ContainerRunner exposes)
-// plus the four sandbox tools, as provider-neutral ToolSpecs. Ported verbatim
+// agentToolSpecs returns the nine tools available to the loop: the five
+// agent back-channel tools (mirroring the MCP tools ContainerRunner exposes;
+// propose_epic was added in Phase 7c) plus the four sandbox tools, as
+// provider-neutral ToolSpecs. The original four (ask_user/report_summary/
+// spawn_subtask/record_progress) plus the sandbox tools were ported verbatim
// (same names/descriptions/JSON schemas) from the former
// executor.agentToolDefs (internal/executor/localtools.go).
func agentToolSpecs() []provider.ToolSpec {
@@ -67,6 +69,19 @@ func agentToolSpecs() []provider.ToolSpec {
},
},
{
+ Name: "propose_epic",
+ Description: "Group one or more stories under a new or existing epic (matched by exact name) when they form a cohesive initiative. Only call this when you've been given several story IDs and independently judge that they belong together.",
+ ParametersJSONSchema: map[string]any{
+ "type": "object",
+ "properties": map[string]any{
+ "name": strProp("short descriptive name for the epic; matched by exact name to reuse an existing epic instead of creating a duplicate"),
+ "description": strProp("optional longer description of the initiative"),
+ "story_ids": map[string]any{"type": "array", "items": map[string]any{"type": "string"}, "description": "the story IDs to group under this epic"},
+ },
+ "required": []string{"name", "story_ids"},
+ },
+ },
+ {
Name: "read_file",
Description: "Read the contents of a file in the sandbox working directory.",
ParametersJSONSchema: map[string]any{
@@ -179,6 +194,23 @@ func (l *Loop) dispatchTool(ctx context.Context, name, argsJSON string) (result
}
return "Noted.", false, nil
+ case "propose_epic":
+ var a struct {
+ Name string `json:"name"`
+ Description string `json:"description"`
+ StoryIDs []string `json:"story_ids"`
+ }
+ _ = json.Unmarshal([]byte(argsJSON), &a)
+ id, peErr := l.Channel.ProposeEpic(ctx, agentchannel.EpicProposal{
+ Name: a.Name,
+ Description: a.Description,
+ StoryIDs: a.StoryIDs,
+ })
+ if peErr != nil {
+ return "", false, peErr
+ }
+ return "Proposed epic " + id, false, nil
+
case "read_file":
var a struct {
Path string `json:"path"`
diff --git a/internal/api/agentmcp_endpoint_test.go b/internal/api/agentmcp_endpoint_test.go
index b9cee3a..7ea71aa 100644
--- a/internal/api/agentmcp_endpoint_test.go
+++ b/internal/api/agentmcp_endpoint_test.go
@@ -27,6 +27,9 @@ func (f *fakeAgentChannel) RecordProgress(_ context.Context, m string) error {
f.progress = append(f.progress, m)
return nil
}
+func (f *fakeAgentChannel) ProposeEpic(context.Context, executor.EpicProposal) (string, error) {
+ return "epic-1", nil
+}
type tokenRT struct {
token string
diff --git a/internal/api/server.go b/internal/api/server.go
index 8bd5b0e..8d03021 100644
--- a/internal/api/server.go
+++ b/internal/api/server.go
@@ -9,6 +9,7 @@ import (
"log/slog"
"net/http"
"os"
+ "strconv"
"strings"
"time"
@@ -520,6 +521,14 @@ func (s *Server) handleListTasks(w http.ResponseWriter, r *http.Request) {
}
filter.Since = t
}
+ if nr := r.URL.Query().Get("needs_review"); nr != "" {
+ b, err := strconv.ParseBool(nr)
+ if err != nil {
+ writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid needs_review: " + nr})
+ return
+ }
+ filter.NeedsReview = &b
+ }
tasks, err := s.store.ListTasks(filter)
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()})
diff --git a/internal/api/server_test.go b/internal/api/server_test.go
index 24e1d30..460c669 100644
--- a/internal/api/server_test.go
+++ b/internal/api/server_test.go
@@ -518,6 +518,64 @@ func TestListTasks_WithTasks(t *testing.T) {
}
}
+// TestListTasks_NeedsReviewFilter proves GET /api/tasks?needs_review=true
+// only returns tasks flagged needs_review (set by
+// internal/scheduler.Scheduler's ask-user-timeout escalation).
+func TestListTasks_NeedsReviewFilter(t *testing.T) {
+ srv, store := testServer(t)
+
+ flagged := &task.Task{
+ ID: "nr-1", Name: "flagged",
+ RepositoryURL: "https://github.com/user/repo",
+ Agent: task.AgentConfig{Type: "claude", Instructions: "x"}, Priority: task.PriorityNormal,
+ Retry: task.RetryConfig{MaxAttempts: 1, Backoff: "linear"},
+ Tags: []string{}, DependsOn: []string{}, State: task.StatePending,
+ }
+ if err := store.CreateTask(flagged); err != nil {
+ t.Fatalf("CreateTask flagged: %v", err)
+ }
+ if err := store.UpdateTaskNeedsReview(flagged.ID, true); err != nil {
+ t.Fatalf("UpdateTaskNeedsReview: %v", err)
+ }
+
+ notFlagged := &task.Task{
+ ID: "nr-2", Name: "not flagged",
+ RepositoryURL: "https://github.com/user/repo",
+ Agent: task.AgentConfig{Type: "claude", Instructions: "x"}, Priority: task.PriorityNormal,
+ Retry: task.RetryConfig{MaxAttempts: 1, Backoff: "linear"},
+ Tags: []string{}, DependsOn: []string{}, State: task.StatePending,
+ }
+ if err := store.CreateTask(notFlagged); err != nil {
+ t.Fatalf("CreateTask notFlagged: %v", err)
+ }
+
+ req := httptest.NewRequest("GET", "/api/tasks?needs_review=true", nil)
+ w := httptest.NewRecorder()
+ srv.Handler().ServeHTTP(w, req)
+
+ if w.Code != http.StatusOK {
+ t.Fatalf("status: want 200, got %d; body: %s", w.Code, w.Body.String())
+ }
+ var tasks []task.Task
+ if err := json.NewDecoder(w.Body).Decode(&tasks); err != nil {
+ t.Fatalf("decode: %v", err)
+ }
+ if len(tasks) != 1 {
+ t.Fatalf("want 1 needs_review task, got %d: %+v", len(tasks), tasks)
+ }
+ if tasks[0].ID != "nr-1" || !tasks[0].NeedsReview {
+ t.Errorf("expected flagged task nr-1 with NeedsReview=true, got %+v", tasks[0])
+ }
+
+ // Sanity check: an invalid needs_review value is a 400, not a silent no-op.
+ req2 := httptest.NewRequest("GET", "/api/tasks?needs_review=notabool", nil)
+ w2 := httptest.NewRecorder()
+ srv.Handler().ServeHTTP(w2, req2)
+ if w2.Code != http.StatusBadRequest {
+ t.Errorf("invalid needs_review: want 400, got %d", w2.Code)
+ }
+}
+
// stateWalkPaths defines the sequence of intermediate states needed to reach each target state.
var stateWalkPaths = map[task.State][]task.State{
task.StatePending: {},
diff --git a/internal/cli/serve.go b/internal/cli/serve.go
index 7ee8d49..57d4681 100644
--- a/internal/cli/serve.go
+++ b/internal/cli/serve.go
@@ -251,9 +251,10 @@ func serve(addr, basePath string) error {
// internal/scheduler). Only `serve` runs this background loop — the
// one-shot `run` command has no long-lived process to host it in.
sch := &scheduler.Scheduler{
- Store: store,
- Pool: pool,
- Logger: logger,
+ Store: store,
+ Pool: pool,
+ Logger: logger,
+ AskUserTimeout: cfg.Scheduler.AskUserTimeout(),
}
if accountant != nil {
// Only assign when non-nil: a nil *budget.Accountant boxed into the
diff --git a/internal/config/config.go b/internal/config/config.go
index 9e065b5..19548af 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -166,6 +166,13 @@ type SchedulerConfig struct {
// PollIntervalSeconds is how often the scheduler checks for role-typed
// FAILED tasks to retry or escalate. Defaults to 30 when zero/unset.
PollIntervalSeconds int `toml:"poll_interval_seconds"`
+ // AskUserTimeoutSeconds is how long a role-typed task may sit BLOCKED on
+ // an unanswered ask_user question before the scheduler resumes it itself
+ // with a system-authored fallback answer (escalating to the next tier of
+ // the role's ladder where possible) and flags it tasks.needs_review for
+ // a human to double-check later. Defaults to 600 (10 minutes) when
+ // zero/unset.
+ AskUserTimeoutSeconds int `toml:"ask_user_timeout_seconds"`
}
// PollInterval returns the configured poll interval, defaulting to 30s.
@@ -176,6 +183,21 @@ func (c SchedulerConfig) PollInterval() time.Duration {
return time.Duration(c.PollIntervalSeconds) * time.Second
}
+// AskUserTimeout returns the configured ask-user timeout, defaulting to 10
+// minutes when zero/unset. Ten minutes is long enough that a human actively
+// working alongside the agent (this is a self-hosted, typically
+// single-operator system, not a large on-call team watching a queue) has a
+// realistic chance to notice a clarification request and answer it, but
+// short enough that a role-typed task doesn't sit stalled for hours — often
+// the common case, not the exception, in this deployment shape — waiting on
+// input that may never come.
+func (c SchedulerConfig) AskUserTimeout() time.Duration {
+ if c.AskUserTimeoutSeconds <= 0 {
+ return 10 * time.Minute
+ }
+ return time.Duration(c.AskUserTimeoutSeconds) * time.Second
+}
+
func Default() (*Config, error) {
home, err := os.UserHomeDir()
if err != nil {
diff --git a/internal/executor/agentmcp.go b/internal/executor/agentmcp.go
index 2177fbd..c0088c5 100644
--- a/internal/executor/agentmcp.go
+++ b/internal/executor/agentmcp.go
@@ -73,11 +73,17 @@ type recordProgressInput struct {
Message string `json:"message" jsonschema:"a short progress note describing what you are doing"`
}
+type proposeEpicInput struct {
+ Name string `json:"name" jsonschema:"short descriptive name for the epic; matched by exact name to reuse an existing epic instead of creating a duplicate"`
+ Description string `json:"description,omitempty" jsonschema:"optional longer description of the initiative"`
+ StoryIDs []string `json:"story_ids" jsonschema:"the story IDs to group under this epic"`
+}
+
func textResult(text string) *mcp.CallToolResult {
return &mcp.CallToolResult{Content: []mcp.Content{&mcp.TextContent{Text: text}}}
}
-// newAgentServer builds an MCP server exposing the four agent tools bound to ch.
+// newAgentServer builds an MCP server exposing the five agent tools bound to ch.
func newAgentServer(ch AgentChannel) *mcp.Server {
s := mcp.NewServer(&mcp.Implementation{Name: "claudomator", Version: "1"}, nil)
@@ -137,6 +143,21 @@ func newAgentServer(ch AgentChannel) *mcp.Server {
return textResult("Noted."), nil, nil
})
+ mcp.AddTool(s, &mcp.Tool{
+ Name: "propose_epic",
+ Description: "Group one or more stories under a new or existing epic (matched by exact name) when they form a cohesive initiative. Only call this when you've been given several story IDs and independently judge that they belong together.",
+ }, func(ctx context.Context, _ *mcp.CallToolRequest, in proposeEpicInput) (*mcp.CallToolResult, any, error) {
+ id, err := ch.ProposeEpic(ctx, EpicProposal{
+ Name: in.Name,
+ Description: in.Description,
+ StoryIDs: in.StoryIDs,
+ })
+ if err != nil {
+ return nil, nil, err
+ }
+ return textResult("Proposed epic " + id), nil, nil
+ })
+
return s
}
diff --git a/internal/executor/agentmcp_test.go b/internal/executor/agentmcp_test.go
index 4aeb7cb..c73d6db 100644
--- a/internal/executor/agentmcp_test.go
+++ b/internal/executor/agentmcp_test.go
@@ -12,11 +12,13 @@ import (
// recordingChannel is a fake AgentChannel that records tool invocations.
type recordingChannel struct {
- asked string
- summary string
- spawned []SubtaskSpec
- progress []string
- spawnID string
+ asked string
+ summary string
+ spawned []SubtaskSpec
+ progress []string
+ spawnID string
+ proposedEpics []EpicProposal
+ epicID string
}
func (c *recordingChannel) AskUser(_ context.Context, q string) (string, error) {
@@ -35,6 +37,10 @@ func (c *recordingChannel) RecordProgress(_ context.Context, m string) error {
c.progress = append(c.progress, m)
return nil
}
+func (c *recordingChannel) ProposeEpic(_ context.Context, spec EpicProposal) (string, error) {
+ c.proposedEpics = append(c.proposedEpics, spec)
+ return c.epicID, nil
+}
func resultText(t *testing.T, res *mcp.CallToolResult) string {
t.Helper()
@@ -178,6 +184,31 @@ func TestAgentServer_RecordProgress(t *testing.T) {
}
}
+func TestAgentServer_ProposeEpic(t *testing.T) {
+ ch := &recordingChannel{epicID: "epic-1"}
+ cs := connectInMemory(t, newAgentServer(ch))
+ res, err := cs.CallTool(context.Background(), &mcp.CallToolParams{
+ Name: "propose_epic",
+ Arguments: map[string]any{"name": "Checkout revamp", "description": "relaunch", "story_ids": []string{"s1", "s2"}},
+ })
+ if err != nil {
+ t.Fatalf("CallTool: %v", err)
+ }
+ if len(ch.proposedEpics) != 1 {
+ t.Fatalf("expected 1 proposed epic, got %d", len(ch.proposedEpics))
+ }
+ spec := ch.proposedEpics[0]
+ if spec.Name != "Checkout revamp" || spec.Description != "relaunch" {
+ t.Errorf("epic spec not propagated: %+v", spec)
+ }
+ if len(spec.StoryIDs) != 2 || spec.StoryIDs[0] != "s1" || spec.StoryIDs[1] != "s2" {
+ t.Errorf("story_ids not propagated: %+v", spec.StoryIDs)
+ }
+ if txt := resultText(t, res); !strings.Contains(txt, "epic-1") {
+ t.Errorf("expected returned epic ID in result, got %q", txt)
+ }
+}
+
type bearerRT struct {
token string
base http.RoundTripper
diff --git a/internal/executor/channel.go b/internal/executor/channel.go
index 30d826c..ef64ede 100644
--- a/internal/executor/channel.go
+++ b/internal/executor/channel.go
@@ -2,12 +2,15 @@ package executor
import (
"context"
+ "database/sql"
"encoding/json"
+ "errors"
"sync"
"time"
"github.com/thepeterstone/claudomator/internal/agentchannel"
"github.com/thepeterstone/claudomator/internal/event"
+ "github.com/thepeterstone/claudomator/internal/story"
"github.com/thepeterstone/claudomator/internal/task"
"github.com/google/uuid"
)
@@ -29,12 +32,24 @@ type AgentChannel = agentchannel.AgentChannel
type SubtaskSpec = agentchannel.SubtaskSpec
type BlockedError = agentchannel.BlockedError
+// EpicProposal is the Phase 7c sibling of SubtaskSpec above: an alias so
+// existing code that refers to executor.EpicProposal (agentmcp.go, tests)
+// compiles against the same type agentchannel.AgentChannel.ProposeEpic uses.
+type EpicProposal = agentchannel.EpicProposal
+
var ErrAgentBlocked = agentchannel.ErrAgentBlocked
// channelStore is the subset of storage the default channel needs.
type channelStore interface {
CreateTask(t *task.Task) error
CreateEvent(e *event.Event) error
+ // CreateEpic, GetEpicByName, GetStory, and UpdateStory back
+ // storeChannel.ProposeEpic (Phase 7c): matching an existing epic by exact
+ // name (or creating a new one), then attaching the given stories to it.
+ CreateEpic(e *story.Epic) error
+ GetEpicByName(name string) (*story.Epic, error)
+ GetStory(id string) (*story.Story, error)
+ UpdateStory(st *story.Story) error
}
// pendingAsker is implemented by channels that buffer an ask_user call so the
@@ -153,4 +168,59 @@ func (c *storeChannel) RecordProgress(_ context.Context, message string) error {
Actor: event.ActorAgent,
Payload: payload,
})
+}
+
+// ProposeEpic groups spec.StoryIDs under a new or existing epic, matched by
+// exact spec.Name — the simplest reasonable de-dup; a fuzzy-matching pass is
+// left to a later phase if it turns out to matter. Story IDs that don't
+// resolve are skipped (not fatal to the call): one bad ID shouldn't prevent
+// grouping the rest. The event is attached to the epic's own ID, matching
+// the story orchestrator's convention (KindEvalVerdict/KindArbitrationDecided)
+// of attaching planning-layer ceremony events to the entity they're about
+// rather than to a task ID — events.task_id has no enforced FK, so this is
+// exactly the tolerance already built in for that.
+func (c *storeChannel) ProposeEpic(_ context.Context, spec agentchannel.EpicProposal) (string, error) {
+ epic, err := c.store.GetEpicByName(spec.Name)
+ if err != nil {
+ if !errors.Is(err, sql.ErrNoRows) {
+ return "", err
+ }
+ epic = &story.Epic{
+ ID: uuid.NewString(),
+ Name: spec.Name,
+ Description: spec.Description,
+ DiscoverySource: "agent",
+ }
+ if err := c.store.CreateEpic(epic); err != nil {
+ return "", err
+ }
+ }
+
+ grouped := make([]string, 0, len(spec.StoryIDs))
+ for _, sid := range spec.StoryIDs {
+ st, err := c.store.GetStory(sid)
+ if err != nil {
+ continue // unresolved story ID: skip, don't fail the whole call
+ }
+ st.EpicID = epic.ID
+ if err := c.store.UpdateStory(st); err != nil {
+ continue
+ }
+ grouped = append(grouped, sid)
+ }
+
+ payload, _ := json.Marshal(struct {
+ EpicID string `json:"epic_id"`
+ Name string `json:"name"`
+ StoryIDs []string `json:"story_ids"`
+ }{EpicID: epic.ID, Name: epic.Name, StoryIDs: grouped})
+ if err := c.store.CreateEvent(&event.Event{
+ TaskID: epic.ID,
+ Kind: event.KindEpicProposed,
+ Actor: event.ActorAgent,
+ Payload: payload,
+ }); err != nil {
+ return epic.ID, err
+ }
+ return epic.ID, nil
} \ No newline at end of file
diff --git a/internal/executor/channel_test.go b/internal/executor/channel_test.go
index 157cb0c..e05228a 100644
--- a/internal/executor/channel_test.go
+++ b/internal/executor/channel_test.go
@@ -2,10 +2,14 @@ package executor
import (
"context"
+ "database/sql"
+ "encoding/json"
"errors"
"testing"
+ "github.com/thepeterstone/claudomator/internal/agentchannel"
"github.com/thepeterstone/claudomator/internal/event"
+ "github.com/thepeterstone/claudomator/internal/story"
"github.com/thepeterstone/claudomator/internal/task"
)
@@ -13,6 +17,15 @@ type fakeChannelStore struct {
createdTasks []*task.Task
createdEvents []*event.Event
createTaskErr error
+
+ // Epic/story fakes back ProposeEpic (Phase 7c). epics is keyed by ID;
+ // GetEpicByName does a linear scan by Name, mirroring the real
+ // storage.DB.GetEpicByName's "exact match, simplest reasonable" behavior.
+ epics map[string]*story.Epic
+ createEpicErr error
+ stories map[string]*story.Story
+ updateStoryErr error
+ updatedStories []*story.Story
}
func (f *fakeChannelStore) CreateTask(t *task.Task) error {
@@ -28,6 +41,46 @@ func (f *fakeChannelStore) CreateEvent(e *event.Event) error {
return nil
}
+func (f *fakeChannelStore) CreateEpic(e *story.Epic) error {
+ if f.createEpicErr != nil {
+ return f.createEpicErr
+ }
+ if f.epics == nil {
+ f.epics = make(map[string]*story.Epic)
+ }
+ f.epics[e.ID] = e
+ return nil
+}
+
+func (f *fakeChannelStore) GetEpicByName(name string) (*story.Epic, error) {
+ for _, e := range f.epics {
+ if e.Name == name {
+ return e, nil
+ }
+ }
+ return nil, sql.ErrNoRows
+}
+
+func (f *fakeChannelStore) GetStory(id string) (*story.Story, error) {
+ st, ok := f.stories[id]
+ if !ok {
+ return nil, sql.ErrNoRows
+ }
+ return st, nil
+}
+
+func (f *fakeChannelStore) UpdateStory(st *story.Story) error {
+ if f.updateStoryErr != nil {
+ return f.updateStoryErr
+ }
+ if f.stories == nil {
+ f.stories = make(map[string]*story.Story)
+ }
+ f.stories[st.ID] = st
+ f.updatedStories = append(f.updatedStories, st)
+ return nil
+}
+
func TestStoreChannel_AskUser_BuffersAndBlocks(t *testing.T) {
ch := newStoreChannel(&fakeChannelStore{}, "task-1")
answer, err := ch.AskUser(context.Background(), `{"text":"q"}`)
@@ -204,3 +257,156 @@ func TestStoreChannel_RecordProgress_EmitsAgentMessage(t *testing.T) {
t.Errorf("TaskID: got %q want task-1", e.TaskID)
}
}
+
+// TestStoreChannel_ProposeEpic_CreatesEpicAndGroupsStories proves (a) from
+// the phase description: a new epic name creates a fresh epics row and sets
+// epic_id on the given stories.
+func TestStoreChannel_ProposeEpic_CreatesEpicAndGroupsStories(t *testing.T) {
+ store := &fakeChannelStore{
+ stories: map[string]*story.Story{
+ "story-1": {ID: "story-1", Name: "s1"},
+ "story-2": {ID: "story-2", Name: "s2"},
+ },
+ }
+ ch := newStoreChannel(store, "task-1")
+
+ epicID, err := ch.ProposeEpic(context.Background(), agentchannel.EpicProposal{
+ Name: "Checkout revamp",
+ Description: "Everything to relaunch checkout",
+ StoryIDs: []string{"story-1", "story-2"},
+ })
+ if err != nil {
+ t.Fatalf("ProposeEpic: %v", err)
+ }
+ if epicID == "" {
+ t.Fatal("expected non-empty epic ID")
+ }
+ if len(store.epics) != 1 {
+ t.Fatalf("expected 1 created epic, got %d", len(store.epics))
+ }
+ epic := store.epics[epicID]
+ if epic == nil {
+ t.Fatalf("epic %q not found in store", epicID)
+ }
+ if epic.Name != "Checkout revamp" || epic.Description != "Everything to relaunch checkout" {
+ t.Errorf("epic fields not propagated: %+v", epic)
+ }
+ if epic.DiscoverySource != "agent" {
+ t.Errorf("DiscoverySource: want agent, got %q", epic.DiscoverySource)
+ }
+ if store.stories["story-1"].EpicID != epicID || store.stories["story-2"].EpicID != epicID {
+ t.Errorf("stories not grouped under epic: %+v / %+v", store.stories["story-1"], store.stories["story-2"])
+ }
+ if len(store.updatedStories) != 2 {
+ t.Errorf("expected 2 UpdateStory calls, got %d", len(store.updatedStories))
+ }
+}
+
+// TestStoreChannel_ProposeEpic_ReusesExistingEpicByName proves (b): calling
+// ProposeEpic again with the same name reuses the existing epic rather than
+// creating a duplicate.
+func TestStoreChannel_ProposeEpic_ReusesExistingEpicByName(t *testing.T) {
+ store := &fakeChannelStore{
+ stories: map[string]*story.Story{
+ "story-1": {ID: "story-1", Name: "s1"},
+ "story-3": {ID: "story-3", Name: "s3"},
+ },
+ }
+ ch := newStoreChannel(store, "task-1")
+
+ firstID, err := ch.ProposeEpic(context.Background(), agentchannel.EpicProposal{
+ Name: "Checkout revamp",
+ StoryIDs: []string{"story-1"},
+ })
+ if err != nil {
+ t.Fatalf("first ProposeEpic: %v", err)
+ }
+
+ secondID, err := ch.ProposeEpic(context.Background(), agentchannel.EpicProposal{
+ Name: "Checkout revamp",
+ StoryIDs: []string{"story-3"},
+ })
+ if err != nil {
+ t.Fatalf("second ProposeEpic: %v", err)
+ }
+
+ if firstID != secondID {
+ t.Errorf("expected the same epic ID to be reused, got %q then %q", firstID, secondID)
+ }
+ if len(store.epics) != 1 {
+ t.Fatalf("expected exactly 1 epic (no duplicate), got %d", len(store.epics))
+ }
+ if store.stories["story-3"].EpicID != firstID {
+ t.Errorf("story-3 not grouped under the reused epic: %+v", store.stories["story-3"])
+ }
+}
+
+// TestStoreChannel_ProposeEpic_SkipsUnresolvedStoryIDs proves that a story ID
+// which doesn't resolve is skipped rather than failing the whole call.
+func TestStoreChannel_ProposeEpic_SkipsUnresolvedStoryIDs(t *testing.T) {
+ store := &fakeChannelStore{
+ stories: map[string]*story.Story{
+ "story-1": {ID: "story-1", Name: "s1"},
+ },
+ }
+ ch := newStoreChannel(store, "task-1")
+
+ epicID, err := ch.ProposeEpic(context.Background(), agentchannel.EpicProposal{
+ Name: "Checkout revamp",
+ StoryIDs: []string{"story-1", "does-not-exist"},
+ })
+ if err != nil {
+ t.Fatalf("ProposeEpic should not fail on one bad story ID: %v", err)
+ }
+ if store.stories["story-1"].EpicID != epicID {
+ t.Errorf("story-1 should still be grouped: %+v", store.stories["story-1"])
+ }
+ if len(store.updatedStories) != 1 {
+ t.Errorf("expected exactly 1 UpdateStory call (the bad ID skipped), got %d", len(store.updatedStories))
+ }
+}
+
+// TestStoreChannel_ProposeEpic_EmitsEpicProposedEvent proves (c): a
+// KindEpicProposed event is emitted, attached to the epic's own ID (not a
+// task ID), with a payload naming the epic and the stories actually grouped.
+func TestStoreChannel_ProposeEpic_EmitsEpicProposedEvent(t *testing.T) {
+ store := &fakeChannelStore{
+ stories: map[string]*story.Story{
+ "story-1": {ID: "story-1", Name: "s1"},
+ },
+ }
+ ch := newStoreChannel(store, "task-1")
+
+ epicID, err := ch.ProposeEpic(context.Background(), agentchannel.EpicProposal{
+ Name: "Checkout revamp",
+ StoryIDs: []string{"story-1", "missing"},
+ })
+ if err != nil {
+ t.Fatalf("ProposeEpic: %v", err)
+ }
+
+ if len(store.createdEvents) != 1 {
+ t.Fatalf("expected 1 event, got %d", len(store.createdEvents))
+ }
+ ev := store.createdEvents[0]
+ if ev.Kind != event.KindEpicProposed || ev.Actor != event.ActorAgent {
+ t.Errorf("got kind=%v actor=%v want epic_proposed/agent", ev.Kind, ev.Actor)
+ }
+ if ev.TaskID != epicID {
+ t.Errorf("event should be attached to the epic ID %q, got %q", epicID, ev.TaskID)
+ }
+ var payload struct {
+ EpicID string `json:"epic_id"`
+ Name string `json:"name"`
+ StoryIDs []string `json:"story_ids"`
+ }
+ if err := json.Unmarshal(ev.Payload, &payload); err != nil {
+ t.Fatalf("unmarshal payload: %v", err)
+ }
+ if payload.EpicID != epicID || payload.Name != "Checkout revamp" {
+ t.Errorf("payload epic fields: %+v", payload)
+ }
+ if len(payload.StoryIDs) != 1 || payload.StoryIDs[0] != "story-1" {
+ t.Errorf("payload.StoryIDs should only include the resolved story, got %+v", payload.StoryIDs)
+ }
+}
diff --git a/internal/executor/container_test.go b/internal/executor/container_test.go
index b06d153..84b3208 100644
--- a/internal/executor/container_test.go
+++ b/internal/executor/container_test.go
@@ -832,3 +832,6 @@ func (noopChannel) SpawnSubtask(_ context.Context, _ SubtaskSpec) (string, error
return "", nil
}
func (noopChannel) RecordProgress(_ context.Context, _ string) error { return nil }
+func (noopChannel) ProposeEpic(_ context.Context, _ EpicProposal) (string, error) {
+ return "", nil
+}
diff --git a/internal/executor/executor.go b/internal/executor/executor.go
index ed38d7d..3c41250 100644
--- a/internal/executor/executor.go
+++ b/internal/executor/executor.go
@@ -16,6 +16,7 @@ import (
"github.com/thepeterstone/claudomator/internal/retry"
"github.com/thepeterstone/claudomator/internal/role"
"github.com/thepeterstone/claudomator/internal/storage"
+ "github.com/thepeterstone/claudomator/internal/story"
"github.com/thepeterstone/claudomator/internal/task"
"github.com/google/uuid"
)
@@ -48,6 +49,12 @@ type Store interface {
// ListDependents returns tasks that directly depend on taskID. Used by
// cascadeFail to proactively cancel a failed task's downstream subtree.
ListDependents(taskID string) ([]*task.Task, error)
+ // CreateEpic, GetEpicByName, GetStory, and UpdateStory back
+ // storeChannel.ProposeEpic (Phase 7c) — see internal/executor/channel.go.
+ CreateEpic(e *story.Epic) error
+ GetEpicByName(name string) (*story.Epic, error)
+ GetStory(id string) (*story.Story, error)
+ UpdateStory(st *story.Story) error
}
// LogPather is an optional interface runners can implement to provide the log
diff --git a/internal/executor/executor_test.go b/internal/executor/executor_test.go
index e7e445e..7a7da48 100644
--- a/internal/executor/executor_test.go
+++ b/internal/executor/executor_test.go
@@ -14,6 +14,7 @@ import (
"github.com/thepeterstone/claudomator/internal/event"
"github.com/thepeterstone/claudomator/internal/storage"
+ "github.com/thepeterstone/claudomator/internal/story"
"github.com/thepeterstone/claudomator/internal/task"
)
@@ -1167,6 +1168,12 @@ func (m *minimalMockStore) GetActiveRoleConfig(_ string) (*storage.RoleConfigRow
return nil, sql.ErrNoRows
}
func (m *minimalMockStore) ListDependents(_ string) ([]*task.Task, error) { return nil, nil }
+func (m *minimalMockStore) CreateEpic(_ *story.Epic) error { return nil }
+func (m *minimalMockStore) GetEpicByName(_ string) (*story.Epic, error) {
+ return nil, sql.ErrNoRows
+}
+func (m *minimalMockStore) GetStory(_ string) (*story.Story, error) { return nil, sql.ErrNoRows }
+func (m *minimalMockStore) UpdateStory(_ *story.Story) error { return nil }
func (m *minimalMockStore) lastStateUpdate() (string, task.State, bool) {
m.mu.Lock()
diff --git a/internal/executor/nativerunner_test.go b/internal/executor/nativerunner_test.go
index b081b8e..afdb49d 100644
--- a/internal/executor/nativerunner_test.go
+++ b/internal/executor/nativerunner_test.go
@@ -19,6 +19,7 @@ import (
"github.com/thepeterstone/claudomator/internal/llm"
"github.com/thepeterstone/claudomator/internal/provider/openaicompat"
"github.com/thepeterstone/claudomator/internal/storage"
+ "github.com/thepeterstone/claudomator/internal/story"
"github.com/thepeterstone/claudomator/internal/task"
)
@@ -237,6 +238,50 @@ func TestNativeRunner_Run_ToolLoop_SpawnSubtask_RolePassthrough(t *testing.T) {
}
}
+// TestNativeRunner_Run_ToolLoop_ProposeEpic is an end-to-end-ish test (fake
+// LLM server, real agentloop.Loop/tools.go dispatch, real storeChannel)
+// proving a propose_epic tool call reaches AgentChannel.ProposeEpic
+// correctly through internal/agentloop/tools.go's dispatchTool, and from
+// there through storeChannel.ProposeEpic into a created epic with the given
+// stories grouped under it — mirroring the spawn_subtask role-passthrough
+// coverage above (Phase 6) for this phase's new tool (Phase 7c).
+func TestNativeRunner_Run_ToolLoop_ProposeEpic(t *testing.T) {
+ srv := fakeChatServer(t, []fakeTurn{
+ {toolCalls: []llm.ToolCall{toolCall("c1", "propose_epic", `{"name":"Checkout revamp","description":"relaunch","story_ids":["story-1","story-2"]}`)}},
+ {content: "finished"},
+ })
+ defer srv.Close()
+
+ r := newLocalRunner(t, srv)
+ tt := localTask()
+ store := &fakeChannelStore{
+ stories: map[string]*story.Story{
+ "story-1": {ID: "story-1", Name: "s1"},
+ "story-2": {ID: "story-2", Name: "s2"},
+ },
+ }
+ ch := newStoreChannel(store, tt.ID)
+ exec := &storage.Execution{ID: uuid.New().String(), TaskID: tt.ID}
+
+ if err := r.Run(context.Background(), tt, exec, ch); err != nil {
+ t.Fatalf("Run: %v", err)
+ }
+
+ if len(store.epics) != 1 {
+ t.Fatalf("expected 1 created epic, got %d", len(store.epics))
+ }
+ var epicID string
+ for id, e := range store.epics {
+ epicID = id
+ if e.Name != "Checkout revamp" || e.Description != "relaunch" {
+ t.Errorf("epic fields not propagated: %+v", e)
+ }
+ }
+ if store.stories["story-1"].EpicID != epicID || store.stories["story-2"].EpicID != epicID {
+ t.Errorf("stories not grouped under the proposed epic: %+v / %+v", store.stories["story-1"], store.stories["story-2"])
+ }
+}
+
func TestNativeRunner_Run_RecordProgress(t *testing.T) {
srv := fakeChatServer(t, []fakeTurn{
{toolCalls: []llm.ToolCall{toolCall("c1", "record_progress", `{"message":"halfway there"}`)}},
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)
+ }
+}
diff --git a/internal/storage/db.go b/internal/storage/db.go
index 8b563a6..51ea648 100644
--- a/internal/storage/db.go
+++ b/internal/storage/db.go
@@ -184,6 +184,11 @@ func (s *DB) migrate() error {
`CREATE INDEX IF NOT EXISTS idx_stories_status ON stories(status)`,
`CREATE INDEX IF NOT EXISTS idx_stories_epic_id ON stories(epic_id)`,
`CREATE INDEX IF NOT EXISTS idx_stories_root_task_id ON stories(root_task_id)`,
+ // needs_review (Phase 7c): flagged by internal/scheduler.Scheduler
+ // when it resumes a role-typed task past its ask_user-timeout with a
+ // system-authored fallback answer, so a human can find and double-check
+ // it later via GET /api/tasks?needs_review=true.
+ `ALTER TABLE tasks ADD COLUMN needs_review BOOLEAN NOT NULL DEFAULT 0`,
}
for _, m := range migrations {
if _, err := s.db.Exec(m); err != nil {
@@ -258,13 +263,13 @@ func (s *DB) CreateTask(t *task.Task) error {
// GetTask retrieves a task by ID.
func (s *DB) GetTask(id string) (*task.Task, error) {
- row := s.db.QueryRow(`SELECT id, name, description, elaboration_input, project, repository_url, config_json, priority, timeout_ns, retry_json, tags_json, depends_on_json, parent_task_id, state, created_at, updated_at, rejection_comment, question_json, summary, interactions_json FROM tasks WHERE id = ?`, id)
+ row := s.db.QueryRow(`SELECT id, name, description, elaboration_input, project, repository_url, config_json, priority, timeout_ns, retry_json, tags_json, depends_on_json, parent_task_id, state, created_at, updated_at, rejection_comment, question_json, summary, interactions_json, needs_review FROM tasks WHERE id = ?`, id)
return scanTask(row)
}
// ListTasks returns tasks matching the given filter.
func (s *DB) ListTasks(filter TaskFilter) ([]*task.Task, error) {
- query := `SELECT id, name, description, elaboration_input, project, repository_url, config_json, priority, timeout_ns, retry_json, tags_json, depends_on_json, parent_task_id, state, created_at, updated_at, rejection_comment, question_json, summary, interactions_json FROM tasks WHERE 1=1`
+ query := `SELECT id, name, description, elaboration_input, project, repository_url, config_json, priority, timeout_ns, retry_json, tags_json, depends_on_json, parent_task_id, state, created_at, updated_at, rejection_comment, question_json, summary, interactions_json, needs_review FROM tasks WHERE 1=1`
var args []interface{}
if filter.State != "" {
@@ -275,6 +280,10 @@ func (s *DB) ListTasks(filter TaskFilter) ([]*task.Task, error) {
query += " AND updated_at > ?"
args = append(args, filter.Since.UTC())
}
+ if filter.NeedsReview != nil {
+ query += " AND needs_review = ?"
+ args = append(args, *filter.NeedsReview)
+ }
query += " ORDER BY created_at DESC"
if filter.Limit > 0 {
query += " LIMIT ?"
@@ -300,7 +309,7 @@ func (s *DB) ListTasks(filter TaskFilter) ([]*task.Task, error) {
// ListSubtasks returns all tasks whose parent_task_id matches the given ID.
func (s *DB) ListSubtasks(parentID string) ([]*task.Task, error) {
- rows, err := s.db.Query(`SELECT id, name, description, elaboration_input, project, repository_url, config_json, priority, timeout_ns, retry_json, tags_json, depends_on_json, parent_task_id, state, created_at, updated_at, rejection_comment, question_json, summary, interactions_json FROM tasks WHERE parent_task_id = ? ORDER BY created_at ASC`, parentID)
+ rows, err := s.db.Query(`SELECT id, name, description, elaboration_input, project, repository_url, config_json, priority, timeout_ns, retry_json, tags_json, depends_on_json, parent_task_id, state, created_at, updated_at, rejection_comment, question_json, summary, interactions_json, needs_review FROM tasks WHERE parent_task_id = ? ORDER BY created_at ASC`, parentID)
if err != nil {
return nil, err
}
@@ -325,7 +334,7 @@ func (s *DB) ListSubtasks(parentID string) ([]*task.Task, error) {
// Only direct dependents are returned — callers that need the full
// transitive downstream subtree (e.g. cascade-cancellation) must recurse.
func (s *DB) ListDependents(taskID string) ([]*task.Task, error) {
- rows, err := s.db.Query(`SELECT id, name, description, elaboration_input, project, repository_url, config_json, priority, timeout_ns, retry_json, tags_json, depends_on_json, parent_task_id, state, created_at, updated_at, rejection_comment, question_json, summary, interactions_json FROM tasks`)
+ rows, err := s.db.Query(`SELECT id, name, description, elaboration_input, project, repository_url, config_json, priority, timeout_ns, retry_json, tags_json, depends_on_json, parent_task_id, state, created_at, updated_at, rejection_comment, question_json, summary, interactions_json, needs_review FROM tasks`)
if err != nil {
return nil, err
}
@@ -400,7 +409,7 @@ func (s *DB) ResetTaskForRetry(id string) (*task.Task, error) {
}
defer tx.Rollback() //nolint:errcheck
- t, err := scanTask(tx.QueryRow(`SELECT id, name, description, elaboration_input, project, repository_url, config_json, priority, timeout_ns, retry_json, tags_json, depends_on_json, parent_task_id, state, created_at, updated_at, rejection_comment, question_json, summary, interactions_json FROM tasks WHERE id = ?`, id))
+ t, err := scanTask(tx.QueryRow(`SELECT id, name, description, elaboration_input, project, repository_url, config_json, priority, timeout_ns, retry_json, tags_json, depends_on_json, parent_task_id, state, created_at, updated_at, rejection_comment, question_json, summary, interactions_json, needs_review FROM tasks WHERE id = ?`, id))
if err != nil {
if err == sql.ErrNoRows {
return nil, fmt.Errorf("task %q not found", id)
@@ -450,6 +459,18 @@ func (s *DB) UpdateTaskAgent(id string, agent task.AgentConfig) error {
return err
}
+// UpdateTaskNeedsReview sets tasks.needs_review. Used by
+// internal/scheduler.Scheduler to flag a task whose BLOCKED ask_user question
+// was resumed with a system-authored fallback answer after timing out,
+// rather than a real human answer, so a human can find it later via
+// GET /api/tasks?needs_review=true.
+func (s *DB) UpdateTaskNeedsReview(id string, needsReview bool) error {
+ now := time.Now().UTC()
+ _, err := s.db.Exec(`UPDATE tasks SET needs_review = ?, updated_at = ? WHERE id = ?`,
+ needsReview, now, id)
+ return err
+}
+
// RejectTask sets a task's state to PENDING and stores the rejection comment.
func (s *DB) RejectTask(id, comment string) error {
tx, err := s.db.Begin()
@@ -550,6 +571,9 @@ type TaskFilter struct {
State task.State
Limit int
Since time.Time
+ // NeedsReview, when non-nil, filters to tasks.needs_review == *NeedsReview.
+ // nil (the default) applies no filter. Backs GET /api/tasks?needs_review=true.
+ NeedsReview *bool
}
// GetMaxUpdatedAt returns the most recent updated_at timestamp across all tasks.
@@ -1156,7 +1180,7 @@ func scanTask(row scanner) (*task.Task, error) {
&t.ID, &t.Name, &t.Description, &elaborationInput, &project, &repositoryURL,
&configJSON, &priority, &timeoutNS, &retryJSON, &tagsJSON, &depsJSON,
&parentTaskID, &state, &t.CreatedAt, &t.UpdatedAt,
- &rejectionComment, &questionJSON, &summary, &interactionsJSON,
+ &rejectionComment, &questionJSON, &summary, &interactionsJSON, &t.NeedsReview,
)
t.ParentTaskID = parentTaskID.String
t.ElaborationInput = elaborationInput.String
diff --git a/internal/storage/epic.go b/internal/storage/epic.go
index 9f3863c..6d2b4d6 100644
--- a/internal/storage/epic.go
+++ b/internal/storage/epic.go
@@ -28,6 +28,17 @@ func (s *DB) GetEpic(id string) (*story.Epic, error) {
return scanEpic(row)
}
+// GetEpicByName retrieves an epic by exact name match (the earliest-created
+// one, if more than one somehow shares a name), or sql.ErrNoRows if none
+// exists. Used by internal/executor's ProposeEpic (Phase 7c) to decide
+// whether a discovery/planner agent's proposed epic name refers to an
+// existing epic or needs a new one — simplest reasonable matching, no fuzzy
+// dedup.
+func (s *DB) GetEpicByName(name string) (*story.Epic, error) {
+ row := s.db.QueryRow(`SELECT id, name, description, status, discovery_source, created_at, updated_at FROM epics WHERE name = ? ORDER BY created_at ASC LIMIT 1`, name)
+ return scanEpic(row)
+}
+
// ListEpics returns epics, optionally filtered by status. Pass an empty
// string to list all epics.
func (s *DB) ListEpics(status string) ([]*story.Epic, error) {
diff --git a/internal/task/task.go b/internal/task/task.go
index 811379e..464d05d 100644
--- a/internal/task/task.go
+++ b/internal/task/task.go
@@ -105,6 +105,12 @@ type Task struct {
Interactions []Interaction `yaml:"-" json:"interactions,omitempty"`
CreatedAt time.Time `yaml:"-" json:"created_at"`
UpdatedAt time.Time `yaml:"-" json:"updated_at"`
+ // NeedsReview is set by internal/scheduler.Scheduler when it resumes a
+ // BLOCKED role-typed task past its ask_user-timeout with a system-authored
+ // fallback answer rather than a real human answer — a flag for a human to
+ // find and double-check later (GET /api/tasks?needs_review=true), not a
+ // state-machine state.
+ NeedsReview bool `yaml:"-" json:"needs_review,omitempty"`
}
// Duration wraps time.Duration for YAML unmarshaling from strings like "30m".
generated by cgit v1.2.3 (git 2.39.1) at 2026-07-16 09:10:25 +0000