From 04b6e7eef473cb6eb69e345a4ea08243a8713077 Mon Sep 17 00:00:00 2001 From: Claude Sonnet 5 Date: Sat, 4 Jul 2026 04:39:28 +0000 Subject: 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 Claude-Session: https://claude.ai/code/session_01V1moSNCJRcP6kykA4tyUSs --- internal/agentchannel/agentchannel.go | 21 +++ internal/agentloop/tools.go | 38 ++++- internal/api/agentmcp_endpoint_test.go | 3 + internal/api/server.go | 9 ++ internal/api/server_test.go | 58 ++++++++ internal/cli/serve.go | 7 +- internal/config/config.go | 22 +++ internal/executor/agentmcp.go | 23 ++- internal/executor/agentmcp_test.go | 41 +++++- internal/executor/channel.go | 70 +++++++++ internal/executor/channel_test.go | 206 ++++++++++++++++++++++++++ internal/executor/container_test.go | 3 + internal/executor/executor.go | 7 + internal/executor/executor_test.go | 7 + internal/executor/nativerunner_test.go | 45 ++++++ internal/scheduler/scheduler.go | 261 +++++++++++++++++++++++++++++++-- internal/scheduler/scheduler_test.go | 253 ++++++++++++++++++++++++++++++++ internal/storage/db.go | 36 ++++- internal/storage/epic.go | 11 ++ internal/task/task.go | 6 + 20 files changed, 1093 insertions(+), 34 deletions(-) (limited to 'internal') 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 { @@ -66,6 +68,19 @@ func agentToolSpecs() []provider.ToolSpec { "required": []string{"message"}, }, }, + { + 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.", @@ -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". -- cgit v1.2.3