From dc97d7ae00bd25668ba495fba249ceade0c6b100 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 23 May 2026 08:07:50 +0000 Subject: feat(storage): emit state_change events on task transitions UpdateTaskState now writes a state_change event in the same transaction as the row update, so the event stream and task state never diverge. Adds UpdateTaskStateBy(id, newState, actor) for explicit actor attribution; UpdateTaskState delegates with ActorSystem (correct for the executor, which drives the state machine). RejectTask and ResetTaskForRetry also emit state_change events attributed to the user. API accept and cancel handlers now attribute their transitions to the user via UpdateTaskStateBy. The executor's Store interface is unchanged, so no test doubles needed updating. https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39 --- internal/api/server.go | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) (limited to 'internal/api') diff --git a/internal/api/server.go b/internal/api/server.go index 28cfe4a..ff3a111 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -11,6 +11,7 @@ import ( "time" "github.com/thepeterstone/claudomator/internal/config" + "github.com/thepeterstone/claudomator/internal/event" "github.com/thepeterstone/claudomator/internal/executor" "github.com/thepeterstone/claudomator/internal/llm" "github.com/thepeterstone/claudomator/internal/notify" @@ -285,7 +286,7 @@ func (s *Server) handleCancelTask(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusConflict, map[string]string{"error": "task cannot be cancelled from state " + string(tk.State)}) return } - if err := s.store.UpdateTaskState(taskID, task.StateCancelled); err != nil { + if err := s.store.UpdateTaskStateBy(taskID, task.StateCancelled, event.ActorUser); err != nil { writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "failed to cancel task"}) return } @@ -703,7 +704,7 @@ func (s *Server) handleAcceptTask(w http.ResponseWriter, r *http.Request) { }) return } - if err := s.store.UpdateTaskState(id, task.StateCompleted); err != nil { + if err := s.store.UpdateTaskStateBy(id, task.StateCompleted, event.ActorUser); err != nil { writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) return } -- cgit v1.2.3 From c7d95f3992d24f86ff71e5f3e18260a8ef8a09f0 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 24 May 2026 08:47:07 +0000 Subject: feat(executor): introduce AgentChannel seam for runner signals MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Defines AgentChannel — the normalized interface by which a runner reports agent-originated signals (AskUser, ReportSummary, SpawnSubtask, RecordProgress) — plus a default storeChannel implementation backed by storage. Runner.Run now takes an AgentChannel; the pool constructs one per execution. The file transport routes its post-exit summary detection through ch.ReportSummary (buffered onto the execution so the pool still applies its extract/synthesize fallbacks, no double-write). AskUser returns ErrAgentBlocked since write-and-exit cannot answer in-session; question persistence stays with the pool's BlockedError handling. SpawnSubtask and RecordProgress are implemented and tested, ready for the MCP transport in Phase 2 where the channel becomes fully load-bearing. Store gains CreateEvent so the channel can emit agent_message events. https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39 --- internal/api/server_test.go | 2 +- internal/executor/channel.go | 117 +++++++++++++++++++++++++++++++++++ internal/executor/channel_test.go | 118 ++++++++++++++++++++++++++++++++++++ internal/executor/claude.go | 6 +- internal/executor/claude_test.go | 8 +-- internal/executor/container.go | 12 ++-- internal/executor/container_test.go | 12 ++-- internal/executor/executor.go | 12 ++-- internal/executor/executor_test.go | 8 ++- internal/executor/gemini.go | 6 +- internal/executor/gemini_test.go | 14 ++--- internal/executor/local.go | 2 +- internal/executor/local_test.go | 6 +- 13 files changed, 282 insertions(+), 41 deletions(-) create mode 100644 internal/executor/channel.go create mode 100644 internal/executor/channel_test.go (limited to 'internal/api') diff --git a/internal/api/server_test.go b/internal/api/server_test.go index 2530d55..f902495 100644 --- a/internal/api/server_test.go +++ b/internal/api/server_test.go @@ -116,7 +116,7 @@ func (m *mockRunner) ExecLogDir(execID string) string { return filepath.Join(m.logDir, execID) } -func (m *mockRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution) error { +func (m *mockRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution, _ executor.AgentChannel) error { if e.ID == "" { e.ID = uuid.New().String() } diff --git a/internal/executor/channel.go b/internal/executor/channel.go new file mode 100644 index 0000000..76df94f --- /dev/null +++ b/internal/executor/channel.go @@ -0,0 +1,117 @@ +package executor + +import ( + "context" + "encoding/json" + "errors" + "time" + + "github.com/thepeterstone/claudomator/internal/event" + "github.com/thepeterstone/claudomator/internal/storage" + "github.com/thepeterstone/claudomator/internal/task" + "github.com/google/uuid" +) + +// ErrAgentBlocked is returned by AgentChannel.AskUser when the transport cannot +// deliver an answer within the current run — e.g. the file transport, which can +// only write-and-exit. Runners translate it into a *BlockedError so the task +// blocks and resumes later. Transports that can suspend the agent in-session +// (MCP, Phase 2) instead block and return the answer. +var ErrAgentBlocked = errors.New("agent blocked awaiting user input") + +// SubtaskSpec describes a child task an agent wants to spawn. +type SubtaskSpec struct { + Name string + Instructions string + Model string + MaxBudgetUSD float64 +} + +// 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 +// today, MCP/tool-use in Phase 2 — is the Runner's private concern. +type AgentChannel interface { + // AskUser records that the agent needs human input. Transports that can + // suspend the agent in-session return the answer; those that cannot return + // ErrAgentBlocked. + AskUser(ctx context.Context, questionJSON string) (answer string, err error) + // ReportSummary records the agent's final summary text. + ReportSummary(ctx context.Context, summary string) error + // SpawnSubtask creates a child task and returns its ID. + 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 +} + +// channelStore is the subset of storage the default channel needs. +type channelStore interface { + CreateTask(t *task.Task) error + CreateEvent(e *event.Event) error +} + +// storeChannel is the default AgentChannel backed by storage. Summary reports +// are buffered onto the execution record so the pool persists them once (with +// its extract/synthesize fallbacks); other signals are written immediately. +type storeChannel struct { + store channelStore + taskID string + exec *storage.Execution +} + +var _ AgentChannel = (*storeChannel)(nil) + +func newStoreChannel(store channelStore, taskID string, exec *storage.Execution) *storeChannel { + return &storeChannel{store: store, taskID: taskID, exec: exec} +} + +// AskUser on the default (file-transport) channel cannot deliver an answer +// in-session, so it always reports the run as blocked. Question persistence is +// owned by the pool's BlockedError handling. +func (c *storeChannel) AskUser(_ context.Context, _ string) (string, error) { + return "", ErrAgentBlocked +} + +// ReportSummary buffers the summary onto the execution record. The pool persists +// it (preferring this value over its extract/synthesize fallbacks). +func (c *storeChannel) ReportSummary(_ context.Context, summary string) error { + if c.exec != nil { + c.exec.Summary = summary + } + return nil +} + +func (c *storeChannel) SpawnSubtask(_ context.Context, spec SubtaskSpec) (string, error) { + now := time.Now().UTC() + child := &task.Task{ + ID: uuid.NewString(), + Name: spec.Name, + ParentTaskID: c.taskID, + Agent: task.AgentConfig{ + Type: "claude", + Model: spec.Model, + Instructions: spec.Instructions, + MaxBudgetUSD: spec.MaxBudgetUSD, + }, + Priority: task.PriorityNormal, + State: task.StatePending, + CreatedAt: now, + UpdatedAt: now, + } + if err := c.store.CreateTask(child); err != nil { + return "", err + } + return child.ID, nil +} + +func (c *storeChannel) RecordProgress(_ context.Context, message string) error { + payload, _ := json.Marshal(struct { + Message string `json:"message"` + }{Message: message}) + return c.store.CreateEvent(&event.Event{ + TaskID: c.taskID, + Kind: event.KindAgentMessage, + Actor: event.ActorAgent, + Payload: payload, + }) +} \ No newline at end of file diff --git a/internal/executor/channel_test.go b/internal/executor/channel_test.go new file mode 100644 index 0000000..822dd35 --- /dev/null +++ b/internal/executor/channel_test.go @@ -0,0 +1,118 @@ +package executor + +import ( + "context" + "errors" + "testing" + + "github.com/thepeterstone/claudomator/internal/event" + "github.com/thepeterstone/claudomator/internal/storage" + "github.com/thepeterstone/claudomator/internal/task" +) + +type fakeChannelStore struct { + createdTasks []*task.Task + createdEvents []*event.Event + createTaskErr error +} + +func (f *fakeChannelStore) CreateTask(t *task.Task) error { + if f.createTaskErr != nil { + return f.createTaskErr + } + f.createdTasks = append(f.createdTasks, t) + return nil +} + +func (f *fakeChannelStore) CreateEvent(e *event.Event) error { + f.createdEvents = append(f.createdEvents, e) + return nil +} + +func TestStoreChannel_AskUser_ReturnsBlocked(t *testing.T) { + ch := newStoreChannel(&fakeChannelStore{}, "task-1", &storage.Execution{}) + answer, err := ch.AskUser(context.Background(), `{"text":"q"}`) + if !errors.Is(err, ErrAgentBlocked) { + t.Errorf("expected ErrAgentBlocked, got %v", err) + } + if answer != "" { + t.Errorf("expected empty answer, got %q", answer) + } +} + +func TestStoreChannel_ReportSummary_BuffersOntoExec(t *testing.T) { + e := &storage.Execution{} + ch := newStoreChannel(&fakeChannelStore{}, "task-1", e) + if err := ch.ReportSummary(context.Background(), "did the thing"); err != nil { + t.Fatalf("ReportSummary: %v", err) + } + if e.Summary != "did the thing" { + t.Errorf("expected exec.Summary set, got %q", e.Summary) + } +} + +func TestStoreChannel_ReportSummary_NilExec(t *testing.T) { + ch := newStoreChannel(&fakeChannelStore{}, "task-1", nil) + if err := ch.ReportSummary(context.Background(), "x"); err != nil { + t.Errorf("expected nil err with nil exec, got %v", err) + } +} + +func TestStoreChannel_SpawnSubtask_CreatesChildWithParent(t *testing.T) { + store := &fakeChannelStore{} + ch := newStoreChannel(store, "parent-1", &storage.Execution{}) + id, err := ch.SpawnSubtask(context.Background(), SubtaskSpec{ + Name: "child", + Instructions: "do it", + Model: "sonnet", + MaxBudgetUSD: 1.5, + }) + if err != nil { + t.Fatalf("SpawnSubtask: %v", err) + } + if id == "" { + t.Error("expected non-empty subtask ID") + } + if len(store.createdTasks) != 1 { + t.Fatalf("expected 1 created task, got %d", len(store.createdTasks)) + } + ct := store.createdTasks[0] + if ct.ParentTaskID != "parent-1" { + t.Errorf("ParentTaskID: got %q want parent-1", ct.ParentTaskID) + } + if ct.ID != id { + t.Errorf("returned id %q != created task id %q", id, ct.ID) + } + if ct.Agent.Instructions != "do it" || ct.Agent.Model != "sonnet" || ct.Agent.MaxBudgetUSD != 1.5 { + t.Errorf("agent config not propagated: %+v", ct.Agent) + } + if ct.State != task.StatePending { + t.Errorf("expected PENDING child, got %s", ct.State) + } +} + +func TestStoreChannel_SpawnSubtask_PropagatesError(t *testing.T) { + store := &fakeChannelStore{createTaskErr: errors.New("boom")} + ch := newStoreChannel(store, "parent-1", &storage.Execution{}) + if _, err := ch.SpawnSubtask(context.Background(), SubtaskSpec{Name: "x"}); err == nil { + t.Error("expected error from CreateTask") + } +} + +func TestStoreChannel_RecordProgress_EmitsAgentMessage(t *testing.T) { + store := &fakeChannelStore{} + ch := newStoreChannel(store, "task-1", &storage.Execution{}) + if err := ch.RecordProgress(context.Background(), "halfway done"); err != nil { + t.Fatalf("RecordProgress: %v", err) + } + if len(store.createdEvents) != 1 { + t.Fatalf("expected 1 event, got %d", len(store.createdEvents)) + } + e := store.createdEvents[0] + if e.Kind != event.KindAgentMessage || e.Actor != event.ActorAgent { + t.Errorf("got kind=%v actor=%v want agent_message/agent", e.Kind, e.Actor) + } + if e.TaskID != "task-1" { + t.Errorf("TaskID: got %q want task-1", e.TaskID) + } +} diff --git a/internal/executor/claude.go b/internal/executor/claude.go index 3c87f26..0123754 100644 --- a/internal/executor/claude.go +++ b/internal/executor/claude.go @@ -51,7 +51,7 @@ func (r *ClaudeRunner) binaryPath() string { // project into a temp sandbox, runs the agent there, then merges committed // changes back to project_dir. On failure the sandbox is preserved and its // path is included in the error. -func (r *ClaudeRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution) error { +func (r *ClaudeRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution, ch AgentChannel) error { projectDir := t.Agent.ProjectDir // Validate project_dir exists when set. @@ -164,7 +164,7 @@ func (r *ClaudeRunner) Run(ctx context.Context, t *task.Task, e *storage.Executi // extract the text as the summary and fall through to normal completion. if isCompletionReport(questionJSON) { r.Logger.Info("treating question file as completion report", "taskID", e.TaskID) - e.Summary = extractQuestionText(questionJSON) + _ = ch.ReportSummary(ctx, extractQuestionText(questionJSON)) } else { // Preserve sandbox on BLOCKED — agent may have partial work and its // Claude session files are stored under the sandbox's project slug. @@ -177,7 +177,7 @@ func (r *ClaudeRunner) Run(ctx context.Context, t *task.Task, e *storage.Executi summaryFile := filepath.Join(logDir, "summary.txt") if summaryData, readErr := os.ReadFile(summaryFile); readErr == nil { os.Remove(summaryFile) // consumed - e.Summary = strings.TrimSpace(string(summaryData)) + _ = ch.ReportSummary(ctx, strings.TrimSpace(string(summaryData))) } // Merge sandbox back to project_dir and clean up. diff --git a/internal/executor/claude_test.go b/internal/executor/claude_test.go index c01e160..0e76260 100644 --- a/internal/executor/claude_test.go +++ b/internal/executor/claude_test.go @@ -259,7 +259,7 @@ func TestClaudeRunner_Run_ResumeSetsSessionIDFromResumeSession(t *testing.T) { } // Run completes successfully (binary is "true"). - _ = r.Run(context.Background(), tk, exec) + _ = r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID, exec)) // SessionID must be the original session (ResumeSessionID), not the new // exec's own ID. If it were exec.ID, a second blocked-then-resumed cycle @@ -284,7 +284,7 @@ func TestClaudeRunner_Run_InaccessibleWorkingDir_ReturnsError(t *testing.T) { } exec := &storage.Execution{ID: "test-exec"} - err := r.Run(context.Background(), tk, exec) + err := r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID, exec)) if err == nil { t.Fatal("expected error for inaccessible working_dir, got nil") @@ -732,7 +732,7 @@ func TestClaudeRunner_Run_ResumeUsesStoredSandboxDir(t *testing.T) { SandboxDir: sandboxDir, } - _ = r.Run(context.Background(), tk, exec) + _ = r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID, exec)) got, err := os.ReadFile(cwdFile) if err != nil { @@ -778,7 +778,7 @@ func TestClaudeRunner_Run_StaleSandboxDir_ClonesAfresh(t *testing.T) { SandboxDir: staleSandbox, } - if err := r.Run(context.Background(), tk, e); err != nil { + if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)); err != nil { t.Fatalf("Run with stale sandbox: %v", err) } diff --git a/internal/executor/container.go b/internal/executor/container.go index 61ac29c..4269ef1 100644 --- a/internal/executor/container.go +++ b/internal/executor/container.go @@ -100,7 +100,7 @@ func (r *ContainerRunner) ensureStoryBranch(ctx context.Context, remoteURL, bran return nil } -func (r *ContainerRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution) error { +func (r *ContainerRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution, ch AgentChannel) error { var err error repoURL := t.RepositoryURL if repoURL == "" { @@ -227,7 +227,7 @@ func (r *ContainerRunner) Run(ctx context.Context, t *task.Task, e *storage.Exec } // Run container (with auth retry on failure). - runErr := r.runContainer(ctx, t, e, workspace, agentHome, isResume, storyBranch) + runErr := r.runContainer(ctx, t, e, workspace, agentHome, isResume, storyBranch, ch) if runErr != nil && isAuthError(runErr) && r.CredentialSyncCmd != "" { r.Logger.Warn("auth failure detected, syncing credentials and retrying once", "taskID", t.ID) syncOut, syncErr := r.command(ctx, r.CredentialSyncCmd).CombinedOutput() @@ -241,7 +241,7 @@ func (r *ContainerRunner) Run(ctx context.Context, t *task.Task, e *storage.Exec if srcData, readErr := os.ReadFile(filepath.Join(r.ClaudeConfigDir, ".claude.json")); readErr == nil { _ = os.WriteFile(filepath.Join(agentHome, ".claude.json"), srcData, 0644) } - runErr = r.runContainer(ctx, t, e, workspace, agentHome, isResume, storyBranch) + runErr = r.runContainer(ctx, t, e, workspace, agentHome, isResume, storyBranch, ch) } if runErr == nil { @@ -257,7 +257,7 @@ func (r *ContainerRunner) Run(ctx context.Context, t *task.Task, e *storage.Exec // runContainer runs the docker container for the given task and handles log setup, // environment files, instructions, and post-execution git operations. -func (r *ContainerRunner) runContainer(ctx context.Context, t *task.Task, e *storage.Execution, workspace, agentHome string, isResume bool, storyBranch string) error { +func (r *ContainerRunner) runContainer(ctx context.Context, t *task.Task, e *storage.Execution, workspace, agentHome string, isResume bool, storyBranch string, ch AgentChannel) error { repoURL := t.RepositoryURL image := t.Agent.ContainerImage @@ -372,7 +372,7 @@ func (r *ContainerRunner) runContainer(ctx context.Context, t *task.Task, e *sto questionJSON := strings.TrimSpace(string(data)) if isCompletionReport(questionJSON) { r.Logger.Info("treating question file as completion report", "taskID", e.TaskID) - e.Summary = extractQuestionText(questionJSON) + _ = ch.ReportSummary(ctx, extractQuestionText(questionJSON)) } else { if e.SessionID == "" { r.Logger.Warn("missing session ID; resume will start fresh", "taskID", e.TaskID) @@ -389,7 +389,7 @@ func (r *ContainerRunner) runContainer(ctx context.Context, t *task.Task, e *sto summaryFile := filepath.Join(logDir, "summary.txt") if summaryData, readErr := os.ReadFile(summaryFile); readErr == nil { os.Remove(summaryFile) // consumed - e.Summary = strings.TrimSpace(string(summaryData)) + _ = ch.ReportSummary(ctx, strings.TrimSpace(string(summaryData))) } // 5. Post-execution: push changes if successful diff --git a/internal/executor/container_test.go b/internal/executor/container_test.go index f0b2a3a..5ee3a3c 100644 --- a/internal/executor/container_test.go +++ b/internal/executor/container_test.go @@ -139,7 +139,7 @@ func TestContainerRunner_Run_PreservesWorkspaceOnFailure(t *testing.T) { } exec := &storage.Execution{ID: "test-exec", TaskID: "test-task"} - err := runner.Run(context.Background(), tk, exec) + err := runner.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID, exec)) if err == nil { t.Fatal("expected error due to mocked docker failure") } @@ -378,7 +378,7 @@ func TestContainerRunner_MissingCredentials_FailsFast(t *testing.T) { } e := &storage.Execution{ID: "test-exec", TaskID: "test-missing-creds"} - err := runner.Run(context.Background(), tk, e) + err := runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)) if err == nil { t.Fatal("expected error due to missing credentials, got nil") } @@ -418,7 +418,7 @@ func TestContainerRunner_MissingSettings_FailsFast(t *testing.T) { } e := &storage.Execution{ID: "test-exec-2", TaskID: "test-missing-settings"} - err := runner.Run(context.Background(), tk, e) + err := runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)) if err == nil { t.Fatal("expected error due to missing settings, got nil") } @@ -504,7 +504,7 @@ func TestContainerRunner_AuthError_SyncsAndRetries(t *testing.T) { e := &storage.Execution{ID: "auth-retry-exec", TaskID: "auth-retry-test"} // Run — first attempt will fail with auth error, triggering sync+retry - runner.Run(context.Background(), tk, e) + runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)) // We don't check error strictly since second run may also fail (git push etc.) // What we care about is that docker was called twice and sync was called if callCount < 2 { @@ -550,7 +550,7 @@ func TestContainerRunner_ClonesStoryBranch(t *testing.T) { } e := &storage.Execution{ID: "exec-1", TaskID: "story-branch-test"} - runner.Run(context.Background(), tk, e) + runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)) os.RemoveAll(e.SandboxDir) // Assert git checkout was called with the story branch name. @@ -597,7 +597,7 @@ func TestContainerRunner_ClonesDefaultBranchWhenNoBranchName(t *testing.T) { } e := &storage.Execution{ID: "exec-2", TaskID: "no-branch-test"} - runner.Run(context.Background(), tk, e) + runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)) os.RemoveAll(e.SandboxDir) for _, a := range cloneArgs { diff --git a/internal/executor/executor.go b/internal/executor/executor.go index 09169bd..76e67b8 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -12,6 +12,7 @@ import ( "sync" "time" + "github.com/thepeterstone/claudomator/internal/event" "github.com/thepeterstone/claudomator/internal/llm" "github.com/thepeterstone/claudomator/internal/retry" "github.com/thepeterstone/claudomator/internal/storage" @@ -41,6 +42,7 @@ type Store interface { ListTasksByStory(storyID string) ([]*task.Task, error) UpdateStoryStatus(id string, status task.StoryState) error CreateTask(t *task.Task) error + CreateEvent(e *event.Event) error UpdateTaskCheckerReport(id, report string) error GetCheckerTask(checkedTaskID string) (*task.Task, error) } @@ -52,9 +54,11 @@ type LogPather interface { ExecLogDir(execID string) string } -// Runner executes a single task and returns the result. +// Runner executes a single task and returns the result. The AgentChannel is how +// the runner reports agent-originated signals (summary, questions, subtasks, +// progress) back to the system. type Runner interface { - Run(ctx context.Context, t *task.Task, exec *storage.Execution) error + Run(ctx context.Context, t *task.Task, exec *storage.Execution, ch AgentChannel) error } // workItem is an entry in the pool's internal work queue. @@ -352,7 +356,7 @@ func (p *Pool) executeResume(ctx context.Context, t *task.Task, exec *storage.Ex } } - err = runner.Run(ctx, t, exec) + err = runner.Run(ctx, t, exec, newStoreChannel(p.store, t.ID, exec)) exec.EndTime = time.Now().UTC() p.decActiveAgent(agentType, &cleaned) @@ -1073,7 +1077,7 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) { } // Run the task. - err = runner.Run(ctx, t, exec) + err = runner.Run(ctx, t, exec, newStoreChannel(p.store, t.ID, exec)) exec.EndTime = time.Now().UTC() p.decActiveAgent(agentType, &cleaned) diff --git a/internal/executor/executor_test.go b/internal/executor/executor_test.go index 9214872..d98efbf 100644 --- a/internal/executor/executor_test.go +++ b/internal/executor/executor_test.go @@ -12,6 +12,7 @@ import ( "testing" "time" + "github.com/thepeterstone/claudomator/internal/event" "github.com/thepeterstone/claudomator/internal/storage" "github.com/thepeterstone/claudomator/internal/task" ) @@ -63,7 +64,7 @@ type mockRunner struct { onRun func(*task.Task, *storage.Execution) error } -func (m *mockRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution) error { +func (m *mockRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution, _ AgentChannel) error { m.mu.Lock() m.calls++ cb := m.onRun @@ -391,11 +392,11 @@ func (m *logPatherMockRunner) ExecLogDir(execID string) string { return filepath.Join(m.logDir, execID) } -func (m *logPatherMockRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution) error { +func (m *logPatherMockRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution, ch AgentChannel) error { m.mu.Lock() m.capturedPath = e.StdoutPath m.mu.Unlock() - return m.mockRunner.Run(ctx, t, e) + return m.mockRunner.Run(ctx, t, e, ch) } // TestPool_Execute_LogPathsPreSetBeforeRun verifies that when the runner @@ -1194,6 +1195,7 @@ func (m *minimalMockStore) GetStory(_ string) (*task.Story, error) func (m *minimalMockStore) ListTasksByStory(_ string) ([]*task.Task, error) { return nil, nil } func (m *minimalMockStore) UpdateStoryStatus(_ string, _ task.StoryState) error { return nil } func (m *minimalMockStore) CreateTask(_ *task.Task) error { return nil } +func (m *minimalMockStore) CreateEvent(_ *event.Event) error { return nil } func (m *minimalMockStore) UpdateTaskCheckerReport(_ string, _ string) error { return nil } func (m *minimalMockStore) GetCheckerTask(_ string) (*task.Task, error) { return nil, nil } diff --git a/internal/executor/gemini.go b/internal/executor/gemini.go index 3abec05..d229361 100644 --- a/internal/executor/gemini.go +++ b/internal/executor/gemini.go @@ -49,7 +49,7 @@ func (r *GeminiRunner) binaryPath() string { // inspect partial work. If the agent writes a question file before exiting, // Run returns *BlockedError with SandboxDir populated so a resume execution // can pick up in the same directory. -func (r *GeminiRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution) error { +func (r *GeminiRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution, ch AgentChannel) error { projectDir := t.Agent.ProjectDir if projectDir != "" { @@ -136,7 +136,7 @@ func (r *GeminiRunner) Run(ctx context.Context, t *task.Task, e *storage.Executi questionJSON := strings.TrimSpace(string(data)) if isCompletionReport(questionJSON) { r.Logger.Info("treating question file as completion report", "taskID", e.TaskID) - e.Summary = extractQuestionText(questionJSON) + _ = ch.ReportSummary(ctx, extractQuestionText(questionJSON)) } else { // Preserve sandbox on BLOCKED so a resume can pick up in the same dir. return &BlockedError{QuestionJSON: questionJSON, SessionID: e.SessionID, SandboxDir: sandboxDir} @@ -147,7 +147,7 @@ func (r *GeminiRunner) Run(ctx context.Context, t *task.Task, e *storage.Executi summaryFile := filepath.Join(logDir, "summary.txt") if summaryData, readErr := os.ReadFile(summaryFile); readErr == nil { os.Remove(summaryFile) - e.Summary = strings.TrimSpace(string(summaryData)) + _ = ch.ReportSummary(ctx, strings.TrimSpace(string(summaryData))) } // Merge sandbox back to project_dir and clean up. diff --git a/internal/executor/gemini_test.go b/internal/executor/gemini_test.go index cd11ebc..c8f7422 100644 --- a/internal/executor/gemini_test.go +++ b/internal/executor/gemini_test.go @@ -125,7 +125,7 @@ func TestGeminiRunner_Run_InaccessibleProjectDir_ReturnsError(t *testing.T) { } exec := &storage.Execution{ID: "test-exec"} - err := r.Run(context.Background(), tk, exec) + err := r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID, exec)) if err == nil { t.Fatal("expected error for inaccessible project_dir, got nil") @@ -213,7 +213,7 @@ func TestGeminiRunner_Run_ProjectDir_RunsInSandbox(t *testing.T) { } e := &storage.Execution{ID: "sandbox-exec", TaskID: "task-1"} - if err := r.Run(context.Background(), tk, e); err != nil { + if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)); err != nil { t.Fatalf("Run: %v", err) } @@ -261,7 +261,7 @@ fi } e := &storage.Execution{ID: "blocked-gemini-exec", TaskID: "task-1"} - err := r.Run(context.Background(), tk, e) + err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)) var blocked *BlockedError if !errors.As(err, &blocked) { @@ -301,7 +301,7 @@ func TestGeminiRunner_Run_ExecError_PreservesSandbox(t *testing.T) { } e := &storage.Execution{ID: "err-gemini-exec", TaskID: "task-1"} - err := r.Run(context.Background(), tk, e) + err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)) if err == nil { t.Fatal("expected error from failing gemini exit") } @@ -352,7 +352,7 @@ func TestGeminiRunner_Run_ResumeUsesStoredSandboxDir(t *testing.T) { SandboxDir: sandboxDir, } - if err := r.Run(context.Background(), tk, e); err != nil { + if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)); err != nil { t.Fatalf("Run with preserved sandbox: %v", err) } @@ -400,7 +400,7 @@ func TestGeminiRunner_Run_StaleSandboxDir_ClonesAfresh(t *testing.T) { SandboxDir: staleSandbox, } - if err := r.Run(context.Background(), tk, e); err != nil { + if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)); err != nil { t.Fatalf("Run with stale sandbox: %v", err) } @@ -438,7 +438,7 @@ func TestGeminiRunner_Run_NoProjectDir_SkipsSandbox(t *testing.T) { } e := &storage.Execution{ID: "no-pd-gemini", TaskID: "task-nopd"} - if err := r.Run(context.Background(), tk, e); err != nil { + if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)); err != nil { t.Fatalf("Run without project_dir: %v", err) } if e.SandboxDir != "" { diff --git a/internal/executor/local.go b/internal/executor/local.go index 5d874c6..e3c9c2c 100644 --- a/internal/executor/local.go +++ b/internal/executor/local.go @@ -40,7 +40,7 @@ func (r *LocalRunner) ExecLogDir(execID string) string { // Run streams a chat completion to stdout.log. The response is wrapped in // stream-json envelopes line-by-line so downstream parsers (summary, // changestats) read it the same way they read Claude output. -func (r *LocalRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution) error { +func (r *LocalRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution, _ AgentChannel) error { if r.Client == nil { return fmt.Errorf("local runner: no LLM client configured") } diff --git a/internal/executor/local_test.go b/internal/executor/local_test.go index d8ab678..8aa8791 100644 --- a/internal/executor/local_test.go +++ b/internal/executor/local_test.go @@ -72,7 +72,7 @@ func TestLocalRunner_Run_WritesStreamJSON(t *testing.T) { } exec := &storage.Execution{ID: uuid.New().String(), TaskID: tt.ID} - if err := r.Run(context.Background(), tt, exec); err != nil { + if err := r.Run(context.Background(), tt, exec, newStoreChannel(nil, tt.ID, exec)); err != nil { t.Fatalf("Run: %v", err) } @@ -121,7 +121,7 @@ func TestLocalRunner_Run_NoClient_Errors(t *testing.T) { r := &LocalRunner{LogDir: t.TempDir()} tt := &task.Task{ID: "x", Agent: task.AgentConfig{Instructions: "hi"}} exec := &storage.Execution{ID: "exec-x"} - err := r.Run(context.Background(), tt, exec) + err := r.Run(context.Background(), tt, exec, newStoreChannel(nil, tt.ID, exec)) if err == nil || !strings.Contains(err.Error(), "no LLM client") { t.Errorf("expected 'no LLM client' error, got %v", err) } @@ -134,7 +134,7 @@ func TestLocalRunner_Run_EmptyInstructions_Errors(t *testing.T) { } tt := &task.Task{ID: "x", Agent: task.AgentConfig{}} exec := &storage.Execution{ID: "exec-x"} - err := r.Run(context.Background(), tt, exec) + err := r.Run(context.Background(), tt, exec, newStoreChannel(nil, tt.ID, exec)) if err == nil || !strings.Contains(err.Error(), "empty instructions") { t.Errorf("expected empty-instructions error, got %v", err) } -- cgit v1.2.3 From 952b7623ee9dceec15099043086622aa2aab4741 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 24 May 2026 09:45:16 +0000 Subject: feat(executor,api): wire agent MCP into ContainerRunner + mount /mcp (Phase 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ContainerRunner now mints a per-task MCP token (from an injected Registry), writes a claude mcp-config into the workspace pointing at the host agent MCP server over host.docker.internal with that bearer, and adds --mcp-config to the in-container claude invocation. The token is revoked when the run ends. After the run, a buffered ask_user (PendingQuestion on the channel) is converted into a BlockedError — the MCP path to BLOCKED — with the file-based question.json kept as a fallback for in-flight tasks started on the old wire. The API server mounts the StreamableHTTP MCP handler at POST/GET/DELETE /mcp when a registry is provided; serve.go constructs one Registry shared by the runners (mint) and the server (resolve). Minting is skipped for gemini agents. Tests: writeMCPConfig output shape, buildInnerCmd flag presence/absence, and an api-level end-to-end MCP tool call through NewServer/Handler proving the route is mounted (and absent without a registry). https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39 --- internal/api/agentmcp_endpoint_test.go | 104 +++++++++++++++++++++++++++++++++ internal/api/server.go | 13 ++++- internal/api/server_test.go | 4 +- internal/cli/serve.go | 8 ++- internal/executor/channel.go | 15 +++++ internal/executor/container.go | 63 +++++++++++++++++++- internal/executor/container_test.go | 58 ++++++++++++++++-- 7 files changed, 253 insertions(+), 12 deletions(-) create mode 100644 internal/api/agentmcp_endpoint_test.go (limited to 'internal/api') diff --git a/internal/api/agentmcp_endpoint_test.go b/internal/api/agentmcp_endpoint_test.go new file mode 100644 index 0000000..b2f77d3 --- /dev/null +++ b/internal/api/agentmcp_endpoint_test.go @@ -0,0 +1,104 @@ +package api + +import ( + "context" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + + "github.com/modelcontextprotocol/go-sdk/mcp" + "github.com/thepeterstone/claudomator/internal/executor" + "github.com/thepeterstone/claudomator/internal/storage" +) + +type fakeAgentChannel struct{ progress []string } + +func (f *fakeAgentChannel) AskUser(context.Context, string) (string, error) { + return "", executor.ErrAgentBlocked +} +func (f *fakeAgentChannel) ReportSummary(context.Context, string) error { return nil } +func (f *fakeAgentChannel) SpawnSubtask(context.Context, executor.SubtaskSpec) (string, error) { + return "sub", nil +} +func (f *fakeAgentChannel) RecordProgress(_ context.Context, m string) error { + f.progress = append(f.progress, m) + return nil +} + +type tokenRT struct { + token string + base http.RoundTripper +} + +func (b tokenRT) RoundTrip(r *http.Request) (*http.Response, error) { + r = r.Clone(r.Context()) + r.Header.Set("Authorization", "Bearer "+b.token) + return b.base.RoundTrip(r) +} + +func TestServer_MountsAgentMCPEndpoint(t *testing.T) { + store, err := storage.Open(filepath.Join(t.TempDir(), "test.db")) + if err != nil { + t.Fatal(err) + } + defer store.Close() + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) + pool := executor.NewPool(1, map[string]executor.Runner{}, store, logger) + + reg := executor.NewRegistry() + fc := &fakeAgentChannel{} + token, _ := reg.Mint(fc) + + srv := NewServer(store, pool, reg, logger, "claude", "gemini") + ts := httptest.NewServer(srv.Handler()) + defer ts.Close() + + client := mcp.NewClient(&mcp.Implementation{Name: "test", Version: "1"}, nil) + cs, err := client.Connect(context.Background(), &mcp.StreamableClientTransport{ + Endpoint: ts.URL + "/mcp", + HTTPClient: &http.Client{Transport: tokenRT{token: token, base: http.DefaultTransport}}, + }, nil) + if err != nil { + t.Fatalf("connect to mounted /mcp: %v", err) + } + defer cs.Close() + + if _, err := cs.CallTool(context.Background(), &mcp.CallToolParams{ + Name: "record_progress", + Arguments: map[string]any{"message": "via api server"}, + }); err != nil { + t.Fatalf("CallTool through api server: %v", err) + } + if len(fc.progress) != 1 || fc.progress[0] != "via api server" { + t.Errorf("tool call did not reach channel: %+v", fc.progress) + } +} + +func TestServer_NoRegistry_NoMCPRoute(t *testing.T) { + store, err := storage.Open(filepath.Join(t.TempDir(), "test.db")) + if err != nil { + t.Fatal(err) + } + defer store.Close() + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) + pool := executor.NewPool(1, map[string]executor.Runner{}, store, logger) + + // nil registry → /mcp must not be mounted. + srv := NewServer(store, pool, nil, logger, "claude", "gemini") + ts := httptest.NewServer(srv.Handler()) + defer ts.Close() + + resp, err := http.Post(ts.URL+"/mcp", "application/json", http.NoBody) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + // Unmounted: the catch-all "GET /" route rejects POST with 405. A mounted + // MCP handler would instead respond (e.g. 400 for the missing token). + if resp.StatusCode == http.StatusBadRequest { + t.Errorf("expected /mcp to be absent without a registry, got 400 (handler ran)") + } +} diff --git a/internal/api/server.go b/internal/api/server.go index ff3a111..1522b72 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -39,6 +39,7 @@ type Server struct { taskLogStore taskLogStore // injectable for tests; defaults to store questionStore questionStore // injectable for tests; defaults to store pool *executor.Pool + registry *executor.Registry // per-task agent MCP token registry; mounts /mcp when set hub *Hub logger *slog.Logger mux *http.ServeMux @@ -100,7 +101,7 @@ func (s *Server) SetLLM(c *llm.Client) { } -func NewServer(store *storage.DB, pool *executor.Pool, logger *slog.Logger, claudeBinPath, geminiBinPath string) *Server { +func NewServer(store *storage.DB, pool *executor.Pool, registry *executor.Registry, logger *slog.Logger, claudeBinPath, geminiBinPath string) *Server { wd, _ := os.Getwd() s := &Server{ ctx: context.Background(), @@ -109,6 +110,7 @@ func NewServer(store *storage.DB, pool *executor.Pool, logger *slog.Logger, clau taskLogStore: store, questionStore: store, pool: pool, + registry: registry, hub: NewHub(), logger: logger, mux: http.NewServeMux(), @@ -180,6 +182,15 @@ func (s *Server) routes() { s.mux.HandleFunc("GET /api/drops", s.handleListDrops) s.mux.HandleFunc("GET /api/drops/{filename}", s.handleGetDrop) s.mux.HandleFunc("POST /api/drops", s.handlePostDrop) + if s.registry != nil { + mcpHandler := executor.NewAgentMCPHandler(s.registry) + // The streamable HTTP transport uses POST (messages), GET (SSE stream), + // and DELETE (session end). Register them explicitly so the patterns are + // more specific than the "GET /" catch-all and don't conflict. + s.mux.Handle("POST /mcp", mcpHandler) + s.mux.Handle("GET /mcp", mcpHandler) + s.mux.Handle("DELETE /mcp", mcpHandler) + } s.mux.Handle("GET /", http.FileServerFS(webui.Files)) } diff --git a/internal/api/server_test.go b/internal/api/server_test.go index f902495..dd6eed5 100644 --- a/internal/api/server_test.go +++ b/internal/api/server_test.go @@ -98,7 +98,7 @@ func testServerWithRunner(t *testing.T, runner executor.Runner) (*Server, *stora "gemini": runner, } pool := executor.NewPool(2, runners, store, logger) - srv := NewServer(store, pool, logger, "claude", "gemini") + srv := NewServer(store, pool, nil, logger, "claude", "gemini") return srv, store } @@ -194,7 +194,7 @@ func testServerWithGeminiMockRunner(t *testing.T) (*Server, *storage.DB) { "gemini": mr, } pool := executor.NewPool(2, runners, store, logger) - srv := NewServer(store, pool, logger, "claude", "gemini") + srv := NewServer(store, pool, nil, logger, "claude", "gemini") return srv, store } diff --git a/internal/cli/serve.go b/internal/cli/serve.go index 459c35b..b23304b 100644 --- a/internal/cli/serve.go +++ b/internal/cli/serve.go @@ -79,6 +79,9 @@ func serve(addr string) error { claudeConfigDir := cfg.ClaudeConfigDir repoDir, _ := os.Getwd() + // Shared per-task agent MCP token registry: runners mint tokens; the API + // server mounts /mcp and resolves them. + agentRegistry := executor.NewRegistry() runners := map[string]executor.Runner{ // ContainerRunner: binaries are resolved via PATH inside the container image, // so ClaudeBinary/GeminiBinary are left empty (host paths would not exist inside). @@ -92,6 +95,7 @@ func serve(addr string) error { ClaudeConfigDir: claudeConfigDir, CredentialSyncCmd: filepath.Join(repoDir, "scripts", "sync-credentials"), Store: store, + Registry: agentRegistry, }, "gemini": &executor.ContainerRunner{ Image: cfg.GeminiImage, @@ -103,6 +107,7 @@ func serve(addr string) error { ClaudeConfigDir: claudeConfigDir, CredentialSyncCmd: filepath.Join(repoDir, "scripts", "sync-credentials"), Store: store, + Registry: agentRegistry, }, "container": &executor.ContainerRunner{ Image: "claudomator-agent:latest", @@ -114,6 +119,7 @@ func serve(addr string) error { ClaudeConfigDir: claudeConfigDir, CredentialSyncCmd: filepath.Join(repoDir, "scripts", "sync-credentials"), Store: store, + Registry: agentRegistry, }, } @@ -146,7 +152,7 @@ func serve(addr string) error { pool.RecoverStaleQueued(context.Background()) pool.RecoverStaleBlocked() - srv := api.NewServer(store, pool, logger, cfg.ClaudeBinaryPath, cfg.GeminiBinaryPath) + srv := api.NewServer(store, pool, agentRegistry, logger, cfg.ClaudeBinaryPath, cfg.GeminiBinaryPath) // Configure notifiers: combine webhook (if set) with web push. notifiers := []notify.Notifier{} diff --git a/internal/executor/channel.go b/internal/executor/channel.go index 541694b..8605ffa 100644 --- a/internal/executor/channel.go +++ b/internal/executor/channel.go @@ -50,6 +50,21 @@ type channelStore interface { CreateEvent(e *event.Event) error } +// pendingAsker is implemented by channels that buffer an ask_user call so the +// runner can convert it into a BlockedError after the agent subprocess exits. +type pendingAsker interface { + PendingQuestion() (questionJSON string, blocked bool) +} + +// channelPendingQuestion reports whether the agent asked a question via the +// channel during the run (the MCP transport's ask_user path). +func channelPendingQuestion(ch AgentChannel) (string, bool) { + if pa, ok := ch.(pendingAsker); ok { + return pa.PendingQuestion() + } + return "", false +} + // storeChannel is the default AgentChannel backed by storage. Summary and // question signals are buffered here under a mutex (they may arrive on an MCP // handler goroutine mid-run); the pool flushes the summary to the execution and diff --git a/internal/executor/container.go b/internal/executor/container.go index 4269ef1..f0f728c 100644 --- a/internal/executor/container.go +++ b/internal/executor/container.go @@ -2,6 +2,7 @@ package executor import ( "context" + "encoding/json" "errors" "fmt" "log/slog" @@ -29,6 +30,9 @@ type ContainerRunner struct { ClaudeConfigDir string // host path to ~/.claude; mounted into container for auth credentials CredentialSyncCmd string // optional path to sync-credentials script for auth-error auto-recovery Store Store // optional; used to look up stories and projects for story-aware cloning + // Registry mints the per-task MCP token; when set, the agent gets an + // mcp-config pointing at the host agent MCP server. + Registry *Registry // Command allows mocking exec.CommandContext for tests. Command func(ctx context.Context, name string, arg ...string) *exec.Cmd } @@ -307,8 +311,24 @@ func (r *ContainerRunner) runContainer(ctx context.Context, t *task.Task, e *sto return fmt.Errorf("writing instructions: %w", err) } + // Per-task MCP back-channel: mint a scoped token bound to this run's channel + // and write an mcp-config pointing the agent at the host agent MCP server. + mcpEnabled := false + if r.Registry != nil && t.Agent.Type != "gemini" { + token, mintErr := r.Registry.Mint(ch) + if mintErr != nil { + return fmt.Errorf("minting agent mcp token: %w", mintErr) + } + defer r.Registry.Revoke(token) + mcpURL := strings.TrimRight(strings.ReplaceAll(r.APIURL, "localhost", "host.docker.internal"), "/") + "/mcp" + if err := writeMCPConfig(workspace, mcpURL, token); err != nil { + return fmt.Errorf("writing mcp config: %w", err) + } + mcpEnabled = true + } + args := r.buildDockerArgs(workspace, agentHome, e.TaskID) - innerCmd := r.buildInnerCmd(t, e, isResume) + innerCmd := r.buildInnerCmd(t, e, isResume, mcpEnabled) fullArgs := append(args, image) fullArgs = append(fullArgs, innerCmd...) @@ -365,7 +385,17 @@ func (r *ContainerRunner) runContainer(ctx context.Context, t *task.Task, e *sto e.SessionID = sessionID } - // Check whether the agent left a question before exiting. + // MCP transport: if the agent called ask_user, the question is buffered on + // the channel. Block so the task resumes with the user's answer. + if q, blocked := channelPendingQuestion(ch); blocked { + if e.SessionID == "" { + r.Logger.Warn("missing session ID; resume will start fresh", "taskID", e.TaskID) + } + return &BlockedError{QuestionJSON: q, SessionID: e.SessionID, SandboxDir: workspace} + } + + // Check whether the agent left a question before exiting (file fallback for + // in-flight tasks started on the pre-MCP wire). questionFile := filepath.Join(logDir, "question.json") if data, readErr := os.ReadFile(questionFile); readErr == nil { os.Remove(questionFile) // consumed @@ -467,7 +497,30 @@ func (r *ContainerRunner) buildDockerArgs(workspace, claudeHome, taskID string) return args } -func (r *ContainerRunner) buildInnerCmd(t *task.Task, e *storage.Execution, isResume bool) []string { +// mcpConfigContainerPath is where the agent mcp-config is mounted in-container +// (the workspace is bind-mounted at /workspace). +const mcpConfigContainerPath = "/workspace/.claudomator-mcp.json" + +// writeMCPConfig writes a claude CLI mcp-config that registers the per-task +// agent MCP server over HTTP with a bearer token. +func writeMCPConfig(workspace, mcpURL, token string) error { + cfg := map[string]any{ + "mcpServers": map[string]any{ + "claudomator": map[string]any{ + "type": "http", + "url": mcpURL, + "headers": map[string]string{"Authorization": "Bearer " + token}, + }, + }, + } + data, err := json.Marshal(cfg) + if err != nil { + return err + } + return os.WriteFile(filepath.Join(workspace, ".claudomator-mcp.json"), data, 0600) +} + +func (r *ContainerRunner) buildInnerCmd(t *task.Task, e *storage.Execution, isResume, mcpEnabled bool) []string { // Claude CLI uses -p for prompt text. To pass a file, we use a shell to cat it. // We use a shell variable to capture the expansion to avoid quoting issues with instructions contents. // The outer single quotes around the sh -c argument prevent host-side expansion. @@ -491,6 +544,9 @@ func (r *ContainerRunner) buildInnerCmd(t *task.Task, e *storage.Execution, isRe if isResume && e.ResumeSessionID != "" { claudeCmd.WriteString(fmt.Sprintf(" --resume %s", e.ResumeSessionID)) } + if mcpEnabled { + claudeCmd.WriteString(" --mcp-config " + mcpConfigContainerPath) + } claudeCmd.WriteString(" --output-format stream-json --verbose --permission-mode bypassPermissions") return []string{"sh", "-c", claudeCmd.String()} @@ -501,6 +557,7 @@ func (r *ContainerRunner) buildInnerCmd(t *task.Task, e *storage.Execution, isRe var scaffoldPrefixes = []string{ ".claudomator-env", ".claudomator-instructions.txt", + ".claudomator-mcp.json", ".agent-home", } diff --git a/internal/executor/container_test.go b/internal/executor/container_test.go index 9cd80dc..86e95e2 100644 --- a/internal/executor/container_test.go +++ b/internal/executor/container_test.go @@ -2,6 +2,7 @@ package executor import ( "context" + "encoding/json" "fmt" "io" "log/slog" @@ -53,13 +54,44 @@ func TestContainerRunner_BuildDockerArgs(t *testing.T) { } } +func TestWriteMCPConfig(t *testing.T) { + dir := t.TempDir() + if err := writeMCPConfig(dir, "http://host.docker.internal:8484/mcp", "tok-abc"); err != nil { + t.Fatalf("writeMCPConfig: %v", err) + } + data, err := os.ReadFile(filepath.Join(dir, ".claudomator-mcp.json")) + if err != nil { + t.Fatalf("read config: %v", err) + } + var parsed struct { + MCPServers map[string]struct { + Type string `json:"type"` + URL string `json:"url"` + Headers map[string]string `json:"headers"` + } `json:"mcpServers"` + } + if err := json.Unmarshal(data, &parsed); err != nil { + t.Fatalf("config is not valid JSON: %v", err) + } + srv, ok := parsed.MCPServers["claudomator"] + if !ok { + t.Fatal("expected claudomator server entry") + } + if srv.Type != "http" || srv.URL != "http://host.docker.internal:8484/mcp" { + t.Errorf("unexpected server config: %+v", srv) + } + if srv.Headers["Authorization"] != "Bearer tok-abc" { + t.Errorf("expected bearer header, got %q", srv.Headers["Authorization"]) + } +} + func TestContainerRunner_BuildInnerCmd(t *testing.T) { runner := &ContainerRunner{} t.Run("claude-fresh", func(t *testing.T) { tk := &task.Task{Agent: task.AgentConfig{Type: "claude"}} exec := &storage.Execution{} - cmd := runner.buildInnerCmd(tk, exec, false) + cmd := runner.buildInnerCmd(tk, exec, false, false) cmdStr := strings.Join(cmd, " ") if strings.Contains(cmdStr, "--resume") { @@ -73,7 +105,7 @@ func TestContainerRunner_BuildInnerCmd(t *testing.T) { t.Run("claude-resume", func(t *testing.T) { tk := &task.Task{Agent: task.AgentConfig{Type: "claude"}} exec := &storage.Execution{ResumeSessionID: "orig-session-123"} - cmd := runner.buildInnerCmd(tk, exec, true) + cmd := runner.buildInnerCmd(tk, exec, true, false) cmdStr := strings.Join(cmd, " ") if !strings.Contains(cmdStr, "--resume orig-session-123") { @@ -81,10 +113,26 @@ func TestContainerRunner_BuildInnerCmd(t *testing.T) { } }) + t.Run("claude-mcp-enabled", func(t *testing.T) { + tk := &task.Task{Agent: task.AgentConfig{Type: "claude"}} + cmdStr := strings.Join(runner.buildInnerCmd(tk, &storage.Execution{}, false, true), " ") + if !strings.Contains(cmdStr, "--mcp-config "+mcpConfigContainerPath) { + t.Errorf("expected --mcp-config flag when MCP enabled, got %q", cmdStr) + } + }) + + t.Run("claude-mcp-disabled", func(t *testing.T) { + tk := &task.Task{Agent: task.AgentConfig{Type: "claude"}} + cmdStr := strings.Join(runner.buildInnerCmd(tk, &storage.Execution{}, false, false), " ") + if strings.Contains(cmdStr, "--mcp-config") { + t.Errorf("did not expect --mcp-config flag when MCP disabled, got %q", cmdStr) + } + }) + t.Run("gemini", func(t *testing.T) { tk := &task.Task{Agent: task.AgentConfig{Type: "gemini"}} exec := &storage.Execution{} - cmd := runner.buildInnerCmd(tk, exec, false) + cmd := runner.buildInnerCmd(tk, exec, false, false) cmdStr := strings.Join(cmd, " ") if !strings.Contains(cmdStr, "gemini -p \"$INST\"") { @@ -99,13 +147,13 @@ func TestContainerRunner_BuildInnerCmd(t *testing.T) { } tkClaude := &task.Task{Agent: task.AgentConfig{Type: "claude"}} - cmdClaude := runnerCustom.buildInnerCmd(tkClaude, &storage.Execution{}, false) + cmdClaude := runnerCustom.buildInnerCmd(tkClaude, &storage.Execution{}, false, false) if !strings.Contains(strings.Join(cmdClaude, " "), "/usr/bin/claude-v2 -p") { t.Errorf("expected custom claude binary, got %q", cmdClaude) } tkGemini := &task.Task{Agent: task.AgentConfig{Type: "gemini"}} - cmdGemini := runnerCustom.buildInnerCmd(tkGemini, &storage.Execution{}, false) + cmdGemini := runnerCustom.buildInnerCmd(tkGemini, &storage.Execution{}, false, false) if !strings.Contains(strings.Join(cmdGemini, " "), "/usr/local/bin/gemini-pro -p") { t.Errorf("expected custom gemini binary, got %q", cmdGemini) } -- cgit v1.2.3 From e766b4c3ef13181ec6adf90ec4abe9db0129fab1 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 25 May 2026 04:57:37 +0000 Subject: feat(api): chatbot-facing MCP server for task orchestration (Phase 3) Adds a chatbot-facing MCP server mounted at /chatbot/mcp, guarded by the shared API bearer token (new config api_token, wired through SetAPIToken). Tools: submit_task, list_tasks, get_task, get_events, answer_question, accept_task, reject_task, cancel_task. submit_task creates and immediately queues a task, resolving the repository URL from a named project when not given explicitly. To keep the REST API and the MCP surface from drifting, the accept/reject/ cancel/answer operations are extracted into a shared service layer (taskops.go) with typed errors mapped to REST status codes; the existing HTTP handlers now delegate to it. The endpoint is only served when a token is configured and presented as a bearer. https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39 --- internal/api/chatbotmcp.go | 218 ++++++++++++++++++++++++++++++++++ internal/api/chatbotmcp_test.go | 233 ++++++++++++++++++++++++++++++++++++ internal/api/server.go | 125 +++----------------- internal/api/taskops.go | 253 ++++++++++++++++++++++++++++++++++++++++ internal/cli/serve.go | 5 + internal/config/config.go | 1 + 6 files changed, 725 insertions(+), 110 deletions(-) create mode 100644 internal/api/chatbotmcp.go create mode 100644 internal/api/chatbotmcp_test.go create mode 100644 internal/api/taskops.go (limited to 'internal/api') diff --git a/internal/api/chatbotmcp.go b/internal/api/chatbotmcp.go new file mode 100644 index 0000000..e14b522 --- /dev/null +++ b/internal/api/chatbotmcp.go @@ -0,0 +1,218 @@ +package api + +import ( + "context" + "crypto/subtle" + "encoding/json" + "net/http" + "strings" + "time" + + "github.com/modelcontextprotocol/go-sdk/mcp" + "github.com/thepeterstone/claudomator/internal/event" + "github.com/thepeterstone/claudomator/internal/storage" + "github.com/thepeterstone/claudomator/internal/task" +) + +// The chatbot-facing MCP server lets a chatbot drive task creation and +// orchestration. Unlike the per-task agent server, it spans all tasks and is +// guarded by the single shared API bearer token (config api_token / SetAPIToken). + +type submitTaskMCPInput struct { + Instructions string `json:"instructions" jsonschema:"complete instructions for the agent that will run this task"` + Name string `json:"name,omitempty" jsonschema:"short human-readable task name; defaults to the first line of instructions"` + Project string `json:"project,omitempty" jsonschema:"name of a configured project; used to resolve the repository when repository_url is omitted"` + RepositoryURL string `json:"repository_url,omitempty" jsonschema:"git repository URL to work in; required unless a project resolves to one"` + Model string `json:"model,omitempty" jsonschema:"optional model override, e.g. sonnet or opus"` + MaxBudgetUSD float64 `json:"max_budget_usd,omitempty" jsonschema:"optional spend cap in USD"` + Timeout string `json:"timeout,omitempty" jsonschema:"optional duration like 15m or 1h"` + Tags []string `json:"tags,omitempty" jsonschema:"optional tags"` +} + +type listTasksMCPInput struct { + State string `json:"state,omitempty" jsonschema:"optional state filter, e.g. RUNNING, READY, BLOCKED, COMPLETED, FAILED"` + Limit int `json:"limit,omitempty" jsonschema:"optional max number of tasks to return"` +} + +type taskIDInput struct { + TaskID string `json:"task_id" jsonschema:"the task ID"` +} + +type getEventsMCPInput struct { + TaskID string `json:"task_id" jsonschema:"the task ID"` + SinceSeq int64 `json:"since_seq,omitempty" jsonschema:"return only events with seq greater than this value"` +} + +type answerQuestionMCPInput struct { + TaskID string `json:"task_id" jsonschema:"the task ID of the BLOCKED task"` + Answer string `json:"answer" jsonschema:"the answer to the agent's question"` +} + +type rejectTaskMCPInput struct { + TaskID string `json:"task_id" jsonschema:"the task ID"` + Comment string `json:"comment,omitempty" jsonschema:"optional reason the work was rejected, fed back to the agent on retry"` +} + +type compactTask struct { + ID string `json:"id"` + Name string `json:"name"` + State task.State `json:"state"` + Project string `json:"project,omitempty"` +} + +func mcpText(text string) *mcp.CallToolResult { + return &mcp.CallToolResult{Content: []mcp.Content{&mcp.TextContent{Text: text}}} +} + +func mcpJSON(v any) (*mcp.CallToolResult, any, error) { + b, err := json.MarshalIndent(v, "", " ") + if err != nil { + return nil, nil, err + } + return mcpText(string(b)), nil, nil +} + +// newChatbotServer builds the chatbot-facing MCP server bound to this Server's +// store and pool. +func (s *Server) newChatbotServer() *mcp.Server { + srv := mcp.NewServer(&mcp.Implementation{Name: "claudomator-chatbot", Version: "1"}, nil) + + mcp.AddTool(srv, &mcp.Tool{ + Name: "submit_task", + Description: "Create a task and immediately queue it for execution. Returns the new task ID and state.", + }, func(ctx context.Context, _ *mcp.CallToolRequest, in submitTaskMCPInput) (*mcp.CallToolResult, any, error) { + spec := submitTaskSpec{ + Name: in.Name, + Instructions: in.Instructions, + Project: in.Project, + RepositoryURL: in.RepositoryURL, + Model: in.Model, + MaxBudgetUSD: in.MaxBudgetUSD, + Tags: in.Tags, + } + if strings.TrimSpace(in.Timeout) != "" { + d, err := time.ParseDuration(in.Timeout) + if err != nil { + return nil, nil, badRequestf("invalid timeout %q: %v", in.Timeout, err) + } + spec.Timeout = d + } + t, err := s.submitTask(ctx, spec) + if err != nil { + return nil, nil, err + } + return mcpJSON(compactTask{ID: t.ID, Name: t.Name, State: t.State, Project: t.Project}) + }) + + mcp.AddTool(srv, &mcp.Tool{ + Name: "list_tasks", + Description: "List tasks, optionally filtered by state. Returns compact records; use get_task for full detail.", + }, func(_ context.Context, _ *mcp.CallToolRequest, in listTasksMCPInput) (*mcp.CallToolResult, any, error) { + filter := storage.TaskFilter{Limit: in.Limit} + if in.State != "" { + st := task.State(in.State) + if !validTaskStates[st] { + return nil, nil, badRequestf("invalid state %q", in.State) + } + filter.State = st + } + tasks, err := s.store.ListTasks(filter) + if err != nil { + return nil, nil, err + } + out := make([]compactTask, 0, len(tasks)) + for _, tk := range tasks { + out = append(out, compactTask{ID: tk.ID, Name: tk.Name, State: tk.State, Project: tk.Project}) + } + return mcpJSON(out) + }) + + mcp.AddTool(srv, &mcp.Tool{ + Name: "get_task", + Description: "Get one task with enriched detail (changestats, deployment status, error message).", + }, func(_ context.Context, _ *mcp.CallToolRequest, in taskIDInput) (*mcp.CallToolResult, any, error) { + tk, err := s.store.GetTask(in.TaskID) + if err != nil { + return nil, nil, errTaskNotFound + } + return mcpJSON(s.enrichTask(tk)) + }) + + mcp.AddTool(srv, &mcp.Tool{ + Name: "get_events", + Description: "Get the observability event stream for a task in seq order. Pass since_seq to page incrementally.", + }, func(_ context.Context, _ *mcp.CallToolRequest, in getEventsMCPInput) (*mcp.CallToolResult, any, error) { + if _, err := s.store.GetTask(in.TaskID); err != nil { + return nil, nil, errTaskNotFound + } + events, err := s.store.ListEvents(in.TaskID, in.SinceSeq) + if err != nil { + return nil, nil, err + } + if events == nil { + events = []*event.Event{} + } + return mcpJSON(events) + }) + + mcp.AddTool(srv, &mcp.Tool{ + Name: "answer_question", + Description: "Answer a BLOCKED task's clarification question. The task resumes from where the agent left off.", + }, func(ctx context.Context, _ *mcp.CallToolRequest, in answerQuestionMCPInput) (*mcp.CallToolResult, any, error) { + if err := s.answerTaskQuestion(ctx, in.TaskID, in.Answer); err != nil { + return nil, nil, err + } + return mcpText("Answer recorded; task queued for resume."), nil, nil + }) + + mcp.AddTool(srv, &mcp.Tool{ + Name: "accept_task", + Description: "Accept a READY task, marking it COMPLETED.", + }, func(ctx context.Context, _ *mcp.CallToolRequest, in taskIDInput) (*mcp.CallToolResult, any, error) { + if err := s.acceptTask(ctx, in.TaskID, event.ActorChatbot); err != nil { + return nil, nil, err + } + return mcpText("Task accepted."), nil, nil + }) + + mcp.AddTool(srv, &mcp.Tool{ + Name: "reject_task", + Description: "Reject a READY task, sending it back to PENDING with an optional comment for the next attempt.", + }, func(_ context.Context, _ *mcp.CallToolRequest, in rejectTaskMCPInput) (*mcp.CallToolResult, any, error) { + if err := s.rejectTask(in.TaskID, in.Comment); err != nil { + return nil, nil, err + } + return mcpText("Task rejected."), nil, nil + }) + + mcp.AddTool(srv, &mcp.Tool{ + Name: "cancel_task", + Description: "Cancel a RUNNING or QUEUED task.", + }, func(_ context.Context, _ *mcp.CallToolRequest, in taskIDInput) (*mcp.CallToolResult, any, error) { + msg, err := s.cancelTask(in.TaskID, event.ActorChatbot) + if err != nil { + return nil, nil, err + } + return mcpText(msg), nil, nil + }) + + return srv +} + +// chatbotMCPHandler returns the HTTP handler for the chatbot MCP server. It +// serves only when an API token is configured and the request presents it as a +// bearer credential; otherwise the underlying handler receives a nil server and +// rejects the request. +func (s *Server) chatbotMCPHandler() http.Handler { + srv := s.newChatbotServer() + return mcp.NewStreamableHTTPHandler(func(r *http.Request) *mcp.Server { + if s.apiToken == "" { + return nil + } + tok := strings.TrimSpace(strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")) + if subtle.ConstantTimeCompare([]byte(tok), []byte(s.apiToken)) != 1 { + return nil + } + return srv + }, nil) +} diff --git a/internal/api/chatbotmcp_test.go b/internal/api/chatbotmcp_test.go new file mode 100644 index 0000000..d13b0d7 --- /dev/null +++ b/internal/api/chatbotmcp_test.go @@ -0,0 +1,233 @@ +package api + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/modelcontextprotocol/go-sdk/mcp" + "github.com/thepeterstone/claudomator/internal/storage" + "github.com/thepeterstone/claudomator/internal/task" +) + +// chatbotClient starts the api server with the given token and returns a +// connected MCP client session pointed at /chatbot/mcp. +func chatbotClient(t *testing.T, srv *Server, token string) *mcp.ClientSession { + t.Helper() + ts := httptest.NewServer(srv.Handler()) + t.Cleanup(ts.Close) + + client := mcp.NewClient(&mcp.Implementation{Name: "test", Version: "1"}, nil) + cs, err := client.Connect(context.Background(), &mcp.StreamableClientTransport{ + Endpoint: ts.URL + "/chatbot/mcp", + HTTPClient: &http.Client{Transport: tokenRT{token: token, base: http.DefaultTransport}}, + }, nil) + if err != nil { + t.Fatalf("connect to /chatbot/mcp: %v", err) + } + t.Cleanup(func() { cs.Close() }) + return cs +} + +func callText(t *testing.T, cs *mcp.ClientSession, name string, args map[string]any) string { + t.Helper() + res, err := cs.CallTool(context.Background(), &mcp.CallToolParams{Name: name, Arguments: args}) + if err != nil { + t.Fatalf("CallTool %s transport error: %v", name, err) + } + if res.IsError { + t.Fatalf("CallTool %s returned tool error: %s", name, toolText(res)) + } + return toolText(res) +} + +func toolText(res *mcp.CallToolResult) string { + if len(res.Content) == 0 { + return "" + } + if tc, ok := res.Content[0].(*mcp.TextContent); ok { + return tc.Text + } + return "" +} + +func TestChatbotMCP_RequiresToken(t *testing.T) { + srv, _ := testServer(t) + srv.SetAPIToken("s3cret") + ts := httptest.NewServer(srv.Handler()) + defer ts.Close() + + client := mcp.NewClient(&mcp.Implementation{Name: "test", Version: "1"}, nil) + _, err := client.Connect(context.Background(), &mcp.StreamableClientTransport{ + Endpoint: ts.URL + "/chatbot/mcp", + HTTPClient: &http.Client{Transport: tokenRT{token: "wrong", base: http.DefaultTransport}}, + }, nil) + if err == nil { + t.Fatal("expected connect with wrong token to be rejected") + } +} + +func TestChatbotMCP_SubmitTask_CreatesAndQueues(t *testing.T) { + srv, store := testServer(t) + srv.SetAPIToken("s3cret") + cs := chatbotClient(t, srv, "s3cret") + + out := callText(t, cs, "submit_task", map[string]any{ + "instructions": "fix the flaky test in package foo", + "repository_url": "https://github.com/user/repo", + }) + var ct compactTask + if err := json.Unmarshal([]byte(out), &ct); err != nil { + t.Fatalf("unmarshal submit_task result %q: %v", out, err) + } + if ct.ID == "" { + t.Fatal("submit_task returned empty task ID") + } + if ct.Name != "fix the flaky test in package foo" { + t.Errorf("name defaulted wrong: %q", ct.Name) + } + if _, err := store.GetTask(ct.ID); err != nil { + t.Fatalf("submitted task not persisted: %v", err) + } +} + +func TestChatbotMCP_SubmitTask_RequiresRepo(t *testing.T) { + srv, _ := testServer(t) + srv.SetAPIToken("s3cret") + cs := chatbotClient(t, srv, "s3cret") + + res, err := cs.CallTool(context.Background(), &mcp.CallToolParams{ + Name: "submit_task", + Arguments: map[string]any{"instructions": "do a thing"}, + }) + if err != nil { + t.Fatalf("transport error: %v", err) + } + if !res.IsError { + t.Fatal("expected submit_task without repository_url to be a tool error") + } +} + +func TestChatbotMCP_GetAndListTasks(t *testing.T) { + srv, store := testServer(t) + srv.SetAPIToken("s3cret") + createTaskWithState(t, store, "ct-ready", task.StateReady) + cs := chatbotClient(t, srv, "s3cret") + + got := callText(t, cs, "get_task", map[string]any{"task_id": "ct-ready"}) + var view map[string]any + if err := json.Unmarshal([]byte(got), &view); err != nil { + t.Fatalf("unmarshal get_task: %v", err) + } + if view["id"] != "ct-ready" { + t.Errorf("get_task id: got %v", view["id"]) + } + + listed := callText(t, cs, "list_tasks", map[string]any{"state": "READY"}) + var tasks []compactTask + if err := json.Unmarshal([]byte(listed), &tasks); err != nil { + t.Fatalf("unmarshal list_tasks: %v", err) + } + if len(tasks) != 1 || tasks[0].ID != "ct-ready" { + t.Errorf("list_tasks returned %+v", tasks) + } +} + +func TestChatbotMCP_GetTask_NotFound(t *testing.T) { + srv, _ := testServer(t) + srv.SetAPIToken("s3cret") + cs := chatbotClient(t, srv, "s3cret") + + res, err := cs.CallTool(context.Background(), &mcp.CallToolParams{ + Name: "get_task", + Arguments: map[string]any{"task_id": "nope"}, + }) + if err != nil { + t.Fatalf("transport error: %v", err) + } + if !res.IsError { + t.Fatal("expected get_task on missing task to be a tool error") + } +} + +func TestChatbotMCP_GetEvents(t *testing.T) { + srv, store := testServer(t) + srv.SetAPIToken("s3cret") + createTaskWithState(t, store, "ct-events", task.StateReady) + cs := chatbotClient(t, srv, "s3cret") + + out := callText(t, cs, "get_events", map[string]any{"task_id": "ct-events"}) + var events []map[string]any + if err := json.Unmarshal([]byte(out), &events); err != nil { + t.Fatalf("unmarshal get_events %q: %v", out, err) + } + // State walk to READY emits state_change events. + if len(events) == 0 { + t.Error("expected at least one event for a task that walked to READY") + } +} + +func TestChatbotMCP_AcceptTask(t *testing.T) { + srv, store := testServer(t) + srv.SetAPIToken("s3cret") + createTaskWithState(t, store, "ct-accept", task.StateReady) + cs := chatbotClient(t, srv, "s3cret") + + callText(t, cs, "accept_task", map[string]any{"task_id": "ct-accept"}) + got, _ := store.GetTask("ct-accept") + if got.State != task.StateCompleted { + t.Errorf("after accept_task: want COMPLETED, got %s", got.State) + } +} + +func TestChatbotMCP_RejectTask(t *testing.T) { + srv, store := testServer(t) + srv.SetAPIToken("s3cret") + createTaskWithState(t, store, "ct-reject", task.StateReady) + cs := chatbotClient(t, srv, "s3cret") + + callText(t, cs, "reject_task", map[string]any{"task_id": "ct-reject", "comment": "needs work"}) + got, _ := store.GetTask("ct-reject") + if got.State != task.StatePending { + t.Errorf("after reject_task: want PENDING, got %s", got.State) + } + if got.RejectionComment != "needs work" { + t.Errorf("rejection comment: got %q", got.RejectionComment) + } +} + +func TestChatbotMCP_CancelTask(t *testing.T) { + srv, store := testServer(t) + srv.SetAPIToken("s3cret") + createTaskWithState(t, store, "ct-cancel", task.StateQueued) + cs := chatbotClient(t, srv, "s3cret") + + callText(t, cs, "cancel_task", map[string]any{"task_id": "ct-cancel"}) + got, _ := store.GetTask("ct-cancel") + if got.State != task.StateCancelled { + t.Errorf("after cancel_task: want CANCELLED, got %s", got.State) + } +} + +func TestChatbotMCP_AnswerQuestion(t *testing.T) { + srv, store := testServer(t) + srv.SetAPIToken("s3cret") + createTaskWithState(t, store, "ct-answer", task.StateBlocked) + if err := store.CreateExecution(&storage.Execution{ + ID: "ct-answer-exec", + TaskID: "ct-answer", + SessionID: "550e8400-e29b-41d4-a716-446655440099", + Status: "BLOCKED", + }); err != nil { + t.Fatalf("create execution: %v", err) + } + cs := chatbotClient(t, srv, "s3cret") + + callText(t, cs, "answer_question", map[string]any{"task_id": "ct-answer", "answer": "use main"}) + got, _ := store.GetTask("ct-answer") + if got.State != task.StateQueued && got.State != task.StateRunning && got.State != task.StateReady { + t.Errorf("after answer_question: want QUEUED/RUNNING/READY, got %s", got.State) + } +} diff --git a/internal/api/server.go b/internal/api/server.go index 1522b72..c9fadb9 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -191,6 +191,12 @@ func (s *Server) routes() { s.mux.Handle("GET /mcp", mcpHandler) s.mux.Handle("DELETE /mcp", mcpHandler) } + // Chatbot-facing MCP server. Always mounted; the handler refuses to serve + // unless a shared API token is configured and presented as a bearer. + chatbotHandler := s.chatbotMCPHandler() + s.mux.Handle("POST /chatbot/mcp", chatbotHandler) + s.mux.Handle("GET /chatbot/mcp", chatbotHandler) + s.mux.Handle("DELETE /chatbot/mcp", chatbotHandler) s.mux.Handle("GET /", http.FileServerFS(webui.Files)) } @@ -282,41 +288,16 @@ func (s *Server) handleDeleteTask(w http.ResponseWriter, r *http.Request) { func (s *Server) handleCancelTask(w http.ResponseWriter, r *http.Request) { taskID := r.PathValue("id") - tk, err := s.store.GetTask(taskID) + msg, err := s.cancelTask(taskID, event.ActorUser) if err != nil { - writeJSON(w, http.StatusNotFound, map[string]string{"error": "task not found"}) - return - } - // If the task is actively running in the pool, cancel it there. - if s.pool.Cancel(taskID) { - writeJSON(w, http.StatusOK, map[string]string{"message": "task cancellation requested", "task_id": taskID}) - return - } - // For non-running tasks (PENDING, QUEUED), transition directly to CANCELLED. - if !task.ValidTransition(tk.State, task.StateCancelled) { - writeJSON(w, http.StatusConflict, map[string]string{"error": "task cannot be cancelled from state " + string(tk.State)}) - return - } - if err := s.store.UpdateTaskStateBy(taskID, task.StateCancelled, event.ActorUser); err != nil { - writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "failed to cancel task"}) + writeOpError(w, err) return } - writeJSON(w, http.StatusOK, map[string]string{"message": "task cancelled", "task_id": taskID}) + writeJSON(w, http.StatusOK, map[string]string{"message": msg, "task_id": taskID}) } func (s *Server) handleAnswerQuestion(w http.ResponseWriter, r *http.Request) { taskID := r.PathValue("id") - - tk, err := s.questionStore.GetTask(taskID) - if err != nil { - writeJSON(w, http.StatusNotFound, map[string]string{"error": "task not found"}) - return - } - if tk.State != task.StateBlocked { - writeJSON(w, http.StatusConflict, map[string]string{"error": "task is not blocked"}) - return - } - var input struct { Answer string `json:"answer"` } @@ -324,61 +305,10 @@ func (s *Server) handleAnswerQuestion(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON: " + err.Error()}) return } - if input.Answer == "" { - writeJSON(w, http.StatusBadRequest, map[string]string{"error": "answer is required"}) - return - } - - // Look up the session ID from the most recent execution. - latest, err := s.questionStore.GetLatestExecution(taskID) - if err != nil || latest.SessionID == "" { - writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "no resumable session found"}) - return - } - - // Record the Q&A interaction before clearing the question. - if tk.QuestionJSON != "" { - var qData struct { - Text string `json:"text"` - Options []string `json:"options"` - } - if jsonErr := json.Unmarshal([]byte(tk.QuestionJSON), &qData); jsonErr == nil { - interaction := task.Interaction{ - QuestionText: qData.Text, - Options: qData.Options, - Answer: input.Answer, - AskedAt: tk.UpdatedAt, - } - if appendErr := s.questionStore.AppendTaskInteraction(taskID, interaction); appendErr != nil { - s.logger.Error("failed to append interaction", "taskID", taskID, "error", appendErr) - } - } - } - - // Clear the question and transition to QUEUED. - if err := s.questionStore.UpdateTaskQuestion(taskID, ""); err != nil { - writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "failed to clear question"}) - return - } - if err := s.questionStore.UpdateTaskState(taskID, task.StateQueued); err != nil { - writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "failed to queue task"}) + if err := s.answerTaskQuestion(s.ctx, taskID, input.Answer); err != nil { + writeOpError(w, err) return } - - // Submit a resume execution. Carry the sandbox path so the runner uses - // the same working directory where Claude stored its session files. - resumeExec := &storage.Execution{ - ID: uuid.New().String(), - TaskID: taskID, - ResumeSessionID: latest.SessionID, - ResumeAnswer: input.Answer, - SandboxDir: latest.SandboxDir, - } - if err := s.pool.SubmitResume(s.ctx, tk, resumeExec); err != nil { - writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": err.Error()}) - return - } - writeJSON(w, http.StatusOK, map[string]string{"message": "task queued for resume", "task_id": taskID}) } @@ -704,40 +634,15 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) { func (s *Server) handleAcceptTask(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") - t, err := s.store.GetTask(id) - if err != nil { - writeJSON(w, http.StatusNotFound, map[string]string{"error": "task not found"}) - return - } - if !task.ValidTransition(t.State, task.StateCompleted) { - writeJSON(w, http.StatusConflict, map[string]string{ - "error": fmt.Sprintf("task cannot be accepted from state %s", t.State), - }) - return - } - if err := s.store.UpdateTaskStateBy(id, task.StateCompleted, event.ActorUser); err != nil { - writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) + if err := s.acceptTask(r.Context(), id, event.ActorUser); err != nil { + writeOpError(w, err) return } - if t.StoryID != "" { - go s.pool.CheckStoryCompletion(r.Context(), t.StoryID) - } writeJSON(w, http.StatusOK, map[string]string{"message": "task accepted", "task_id": id}) } func (s *Server) handleRejectTask(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") - t, err := s.store.GetTask(id) - if err != nil { - writeJSON(w, http.StatusNotFound, map[string]string{"error": "task not found"}) - return - } - if !task.ValidTransition(t.State, task.StatePending) { - writeJSON(w, http.StatusConflict, map[string]string{ - "error": fmt.Sprintf("task cannot be rejected from state %s", t.State), - }) - return - } var input struct { Comment string `json:"comment"` } @@ -745,8 +650,8 @@ func (s *Server) handleRejectTask(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON: " + err.Error()}) return } - if err := s.store.RejectTask(id, input.Comment); err != nil { - writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) + if err := s.rejectTask(id, input.Comment); err != nil { + writeOpError(w, err) return } writeJSON(w, http.StatusOK, map[string]string{"message": "task rejected", "task_id": id}) diff --git a/internal/api/taskops.go b/internal/api/taskops.go new file mode 100644 index 0000000..881babe --- /dev/null +++ b/internal/api/taskops.go @@ -0,0 +1,253 @@ +package api + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "strings" + "time" + + "github.com/google/uuid" + "github.com/thepeterstone/claudomator/internal/event" + "github.com/thepeterstone/claudomator/internal/storage" + "github.com/thepeterstone/claudomator/internal/task" +) + +// Shared task-operation service layer. Both the REST handlers and the +// chatbot-facing MCP tools call these methods so the two surfaces cannot drift. +// Errors are typed so callers can map them to the right transport status. + +// errTaskNotFound is returned when a task ID does not resolve. REST maps it to +// 404; MCP surfaces it as a tool error. +var errTaskNotFound = errors.New("task not found") + +// conflictError signals an operation that is invalid for the task's current +// state (REST 409). +type conflictError struct{ msg string } + +func (e *conflictError) Error() string { return e.msg } + +func conflictf(format string, a ...any) error { return &conflictError{msg: fmt.Sprintf(format, a...)} } + +// badRequestError signals malformed or missing input (REST 400). +type badRequestError struct{ msg string } + +func (e *badRequestError) Error() string { return e.msg } + +func badRequestf(format string, a ...any) error { return &badRequestError{msg: fmt.Sprintf(format, a...)} } + +// submitTaskSpec describes a task a chatbot wants created and immediately run. +type submitTaskSpec struct { + Name string + Instructions string + Project string + RepositoryURL string + Model string + MaxBudgetUSD float64 + Timeout time.Duration + Tags []string +} + +// submitTask creates a task and queues it for execution in one step. The +// repository URL is resolved from the named project when not given explicitly, +// because task.Validate requires it before the executor's lazy resolution runs. +func (s *Server) submitTask(ctx context.Context, spec submitTaskSpec) (*task.Task, error) { + if strings.TrimSpace(spec.Instructions) == "" { + return nil, badRequestf("instructions are required") + } + + repoURL := spec.RepositoryURL + if repoURL == "" && spec.Project != "" { + if proj, err := s.store.GetProject(spec.Project); err == nil { + repoURL = proj.RemoteURL + } + } + if repoURL == "" { + return nil, badRequestf("repository_url is required (pass it directly or a project that resolves to one)") + } + + name := strings.TrimSpace(spec.Name) + if name == "" { + name = firstLine(spec.Instructions, 72) + } + + now := time.Now().UTC() + t := &task.Task{ + ID: uuid.New().String(), + Name: name, + Project: spec.Project, + RepositoryURL: repoURL, + Agent: task.AgentConfig{ + Type: "claude", + Model: spec.Model, + Instructions: spec.Instructions, + MaxBudgetUSD: spec.MaxBudgetUSD, + }, + Priority: task.PriorityNormal, + Tags: spec.Tags, + DependsOn: []string{}, + Retry: task.RetryConfig{MaxAttempts: 1, Backoff: "exponential"}, + State: task.StatePending, + CreatedAt: now, + UpdatedAt: now, + } + if spec.Timeout > 0 { + t.Timeout.Duration = spec.Timeout + } + if t.Tags == nil { + t.Tags = []string{} + } + + if err := task.Validate(t); err != nil { + return nil, badRequestf("%s", err.Error()) + } + if err := s.store.CreateTask(t); err != nil { + return nil, err + } + if err := s.store.UpdateTaskStateBy(t.ID, task.StateQueued, event.ActorChatbot); err != nil { + return nil, err + } + t.State = task.StateQueued + if err := s.pool.Submit(ctx, t); err != nil { + return nil, err + } + return t, nil +} + +// acceptTask transitions a READY task to COMPLETED. +func (s *Server) acceptTask(ctx context.Context, id string, actor event.Actor) error { + t, err := s.store.GetTask(id) + if err != nil { + return errTaskNotFound + } + if !task.ValidTransition(t.State, task.StateCompleted) { + return conflictf("task cannot be accepted from state %s", t.State) + } + if err := s.store.UpdateTaskStateBy(id, task.StateCompleted, actor); err != nil { + return err + } + if t.StoryID != "" { + go s.pool.CheckStoryCompletion(ctx, t.StoryID) + } + return nil +} + +// rejectTask transitions a READY task back to PENDING with an optional comment. +func (s *Server) rejectTask(id, comment string) error { + t, err := s.store.GetTask(id) + if err != nil { + return errTaskNotFound + } + if !task.ValidTransition(t.State, task.StatePending) { + return conflictf("task cannot be rejected from state %s", t.State) + } + return s.store.RejectTask(id, comment) +} + +// cancelTask cancels a running/queued task, returning a human-readable result. +func (s *Server) cancelTask(id string, actor event.Actor) (string, error) { + tk, err := s.store.GetTask(id) + if err != nil { + return "", errTaskNotFound + } + if s.pool.Cancel(id) { + return "task cancellation requested", nil + } + if !task.ValidTransition(tk.State, task.StateCancelled) { + return "", conflictf("task cannot be cancelled from state %s", tk.State) + } + if err := s.store.UpdateTaskStateBy(id, task.StateCancelled, actor); err != nil { + return "", err + } + return "task cancelled", nil +} + +// answerTaskQuestion answers a BLOCKED task's clarification and resumes it. +func (s *Server) answerTaskQuestion(ctx context.Context, id, answer string) error { + tk, err := s.questionStore.GetTask(id) + if err != nil { + return errTaskNotFound + } + if tk.State != task.StateBlocked { + return conflictf("task is not blocked") + } + if strings.TrimSpace(answer) == "" { + return badRequestf("answer is required") + } + + latest, err := s.questionStore.GetLatestExecution(id) + if err != nil || latest.SessionID == "" { + return fmt.Errorf("no resumable session found") + } + + if tk.QuestionJSON != "" { + var qData struct { + Text string `json:"text"` + Options []string `json:"options"` + } + if json.Unmarshal([]byte(tk.QuestionJSON), &qData) == nil { + if appendErr := s.questionStore.AppendTaskInteraction(id, task.Interaction{ + QuestionText: qData.Text, + Options: qData.Options, + Answer: answer, + AskedAt: tk.UpdatedAt, + }); appendErr != nil { + s.logger.Error("failed to append interaction", "taskID", id, "error", appendErr) + } + } + } + + if err := s.questionStore.UpdateTaskQuestion(id, ""); err != nil { + return err + } + if err := s.questionStore.UpdateTaskState(id, task.StateQueued); err != nil { + return err + } + + resumeExec := &storage.Execution{ + ID: uuid.New().String(), + TaskID: id, + ResumeSessionID: latest.SessionID, + ResumeAnswer: answer, + SandboxDir: latest.SandboxDir, + } + return s.pool.SubmitResume(ctx, tk, resumeExec) +} + +// writeOpError maps a service-layer error to the appropriate REST status. +func writeOpError(w http.ResponseWriter, err error) { + switch { + case errors.Is(err, errTaskNotFound): + writeJSON(w, http.StatusNotFound, map[string]string{"error": "task not found"}) + return + } + var ce *conflictError + if errors.As(err, &ce) { + writeJSON(w, http.StatusConflict, map[string]string{"error": ce.Error()}) + return + } + var be *badRequestError + if errors.As(err, &be) { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": be.Error()}) + return + } + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) +} + +// firstLine returns the first non-empty line of s, truncated to max runes. +func firstLine(s string, max int) string { + line := s + if i := strings.IndexByte(s, '\n'); i >= 0 { + line = s[:i] + } + line = strings.TrimSpace(line) + if r := []rune(line); len(r) > max { + return strings.TrimSpace(string(r[:max])) + } + if line == "" { + return "task" + } + return line +} diff --git a/internal/cli/serve.go b/internal/cli/serve.go index b23304b..e545763 100644 --- a/internal/cli/serve.go +++ b/internal/cli/serve.go @@ -182,6 +182,11 @@ func serve(addr string) error { } srv.SetGitHubWebhookConfig(cfg.WebhookSecret, cfg.Projects) + if cfg.APIToken != "" { + srv.SetAPIToken(cfg.APIToken) + logger.Info("API token configured; chatbot MCP endpoint enabled at /chatbot/mcp") + } + // Register scripts. wd, _ := os.Getwd() srv.SetScripts(api.ScriptRegistry{ diff --git a/internal/config/config.go b/internal/config/config.go index 25187cf..94f3ec7 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -62,6 +62,7 @@ type Config struct { WebhookURL string `toml:"webhook_url"` WorkspaceRoot string `toml:"workspace_root"` WebhookSecret string `toml:"webhook_secret"` + APIToken string `toml:"api_token"` // shared bearer for web UI + chatbot MCP; empty disables both auth and the chatbot MCP endpoint Projects []Project `toml:"projects"` VAPIDPublicKey string `toml:"vapid_public_key"` VAPIDPrivateKey string `toml:"vapid_private_key"` -- cgit v1.2.3 From 6890bafa65a309b540cbe1562c39df137f202369 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 25 May 2026 05:01:05 +0000 Subject: refactor(api): remove the natural-language elaborate REST endpoint (Phase 3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Chatbots now elaborate tasks in-conversation and submit them via the chatbot MCP server, so the POST /api/tasks/elaborate shell-out path is no longer needed. Removes the route, handleElaborateTask, and the task-elaborate-only helpers (buildElaboratePrompt, sanitizeElaboratedTask, readProjectContext, appendRawNarrative, elaborateWith{Claude,Local,Gemini}, and the elaboratedTask/ elaboratedAgent types) plus their dedicated tests. Shared helpers (extractJSON, claude/geminiBinaryPath, claude/geminiJSONResult, elaborateTimeout, the per-IP limiter) are retained — the story-elaborate endpoint and task-validate endpoint still use them. Stories themselves are untouched here; their removal stays in the Phase 8 cleanup pass. https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39 --- internal/api/elaborate.go | 337 ---------------------- internal/api/elaborate_local_test.go | 214 -------------- internal/api/elaborate_test.go | 527 ----------------------------------- internal/api/server.go | 1 - internal/api/server_test.go | 28 -- internal/api/validate_test.go | 21 ++ 6 files changed, 21 insertions(+), 1107 deletions(-) delete mode 100644 internal/api/elaborate_local_test.go delete mode 100644 internal/api/elaborate_test.go (limited to 'internal/api') diff --git a/internal/api/elaborate.go b/internal/api/elaborate.go index 8676b36..e8930a6 100644 --- a/internal/api/elaborate.go +++ b/internal/api/elaborate.go @@ -6,128 +6,13 @@ import ( "encoding/json" "fmt" "net/http" - "os" "os/exec" - "path/filepath" - "sort" "strings" "time" - - "github.com/thepeterstone/claudomator/internal/llm" ) const elaborateTimeout = 30 * time.Second -func buildElaboratePrompt(workDir string) string { - workDirLine := ` "project_dir": string — leave empty unless you have a specific reason to set it,` - if workDir != "" { - workDirLine = fmt.Sprintf(` "project_dir": string — use %q for tasks that operate on this codebase, empty string otherwise,`, workDir) - } - return `You are a task configuration assistant for Claudomator, an AI task runner that executes tasks by running Claude or Gemini as a subprocess. - -Your ONLY job is to convert any user request into a Claudomator task JSON object. You MUST always output valid JSON. Never ask clarifying questions. Never explain. Never refuse. Make reasonable assumptions and produce the JSON. - -Output ONLY a valid JSON object matching this schema (no markdown fences, no prose, no explanation): - -{ - "name": string — short imperative title (≤60 chars), - "description": string — 1-2 sentence summary, - "agent": { - "type": "claude" | "gemini", - "model": string — "sonnet" for claude, "gemini-2.5-flash-lite" for gemini, - "instructions": string — detailed, step-by-step instructions for the agent. Must end with a "## Acceptance Criteria" section listing measurable conditions that define success. For coding tasks, include TDD requirements (write failing tests first, then implement), -` + workDirLine + ` - "max_budget_usd": number — conservative estimate (0.25–5.00), - "allowed_tools": array — every tool the task genuinely needs. Include "Write" if creating files, "Edit" if modifying files, "Read" if reading files, "Bash" for shell/git/test commands, "Grep"/"Glob" for searching. - }, - "timeout": string — e.g. "15m", - "priority": string — "normal" | "high" | "low", - "tags": array — relevant lowercase tags -}` -} - -// elaboratedTask mirrors the task creation schema for elaboration responses. -type elaboratedTask struct { - Name string `json:"name"` - Description string `json:"description"` - Agent elaboratedAgent `json:"agent"` - Timeout string `json:"timeout"` - Priority string `json:"priority"` - Tags []string `json:"tags"` -} - -type elaboratedAgent struct { - Type string `json:"type"` - Model string `json:"model"` - Instructions string `json:"instructions"` - ProjectDir string `json:"project_dir"` - MaxBudgetUSD float64 `json:"max_budget_usd"` - AllowedTools []string `json:"allowed_tools"` -} - -// sanitizeElaboratedTask enforces tool completeness and dev practice compliance. -// It modifies t in place, inferring missing tools from instruction keywords and -// appending required sections when they are absent. -func sanitizeElaboratedTask(t *elaboratedTask) { - lower := strings.ToLower(t.Agent.Instructions) - - // Build current tool set. - toolSet := make(map[string]bool, len(t.Agent.AllowedTools)) - for _, tool := range t.Agent.AllowedTools { - toolSet[tool] = true - } - - // Infer missing tools from instruction keywords. - type rule struct { - tool string - keywords []string - } - rules := []rule{ - {"Write", []string{"create file", "write file", "new file", "write to", "save to", "output to", "generate file", "creates a file", "create a new file"}}, - {"Edit", []string{"edit", "modify", "refactor", "replace", "patch"}}, - {"Read", []string{"read", "inspect", "examine", "look at the file"}}, - {"Bash", []string{"run", "execute", "bash", "shell", "command", "build", "compile", "git", "install", "make"}}, - {"Grep", []string{"search for", "grep", "find in", "locate in"}}, - {"Glob", []string{"find file", "list file", "search file"}}, - } - for _, r := range rules { - if toolSet[r.tool] { - continue - } - for _, kw := range r.keywords { - if strings.Contains(lower, kw) { - toolSet[r.tool] = true - break - } - } - } - // Edit without Read is almost always wrong. - if toolSet["Edit"] && !toolSet["Read"] { - toolSet["Read"] = true - } - // Rebuild the list only when tools were added. - if len(toolSet) > len(t.Agent.AllowedTools) { - tools := make([]string, 0, len(toolSet)) - for tool := range toolSet { - tools = append(tools, tool) - } - sort.Strings(tools) - t.Agent.AllowedTools = tools - } - - // Append an acceptance criteria section when none is present. - if !strings.Contains(lower, "acceptance") && - !strings.Contains(lower, "done when") && - !strings.Contains(lower, "success criteria") { - t.Agent.Instructions += "\n\n## Acceptance Criteria\nBefore finishing, verify all stated goals are met, tests pass (if applicable), and no unintended side effects were introduced." - } - - // Append a TDD reminder for coding tasks that do not already mention tests. - if (toolSet["Edit"] || toolSet["Write"]) && !strings.Contains(lower, "test") { - t.Agent.Instructions += "\n\n## Dev Practices\nFollow TDD: write a failing test first, then implement the minimum code to make it pass. Commit all changes before finishing." - } -} - // claudeJSONResult is the top-level object returned by `claude --output-format json`. type claudeJSONResult struct { Result string `json:"result"` @@ -167,142 +52,6 @@ func (s *Server) geminiBinaryPath() string { return "gemini" } -func readProjectContext(workDir string) string { - if workDir == "" { - return "" - } - var sb strings.Builder - for _, filename := range []string{"CLAUDE.md", ".agent/worklog.md"} { - path := filepath.Join(workDir, filename) - if data, err := os.ReadFile(path); err == nil { - if sb.Len() > 0 { - sb.WriteString("\n\n") - } - sb.WriteString(fmt.Sprintf("--- %s ---\n%s", filename, string(data))) - } - } - return sb.String() -} - -func (s *Server) appendRawNarrative(workDir, prompt string) { - if workDir == "" { - return - } - docsDir := filepath.Join(workDir, "docs") - if err := os.MkdirAll(docsDir, 0755); err != nil { - s.logger.Error("elaborate: failed to create docs directory", "error", err, "path", docsDir) - return - } - path := filepath.Join(docsDir, "RAW_NARRATIVE.md") - f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644) - if err != nil { - s.logger.Error("elaborate: failed to open RAW_NARRATIVE.md", "error", err, "path", path) - return - } - defer f.Close() - - entry := fmt.Sprintf("\n--- %s ---\n%s\n", time.Now().Format(time.RFC3339), prompt) - if _, err := f.WriteString(entry); err != nil { - s.logger.Error("elaborate: failed to write to RAW_NARRATIVE.md", "error", err, "path", path) - } -} - -func (s *Server) elaborateWithClaude(ctx context.Context, workDir, fullPrompt string) (*elaboratedTask, error) { - cmd := exec.CommandContext(ctx, s.claudeBinaryPath(), - "-p", fullPrompt, - "--system-prompt", buildElaboratePrompt(workDir), - "--output-format", "json", - "--model", "haiku", - ) - - var stdout, stderr bytes.Buffer - cmd.Stdout = &stdout - cmd.Stderr = &stderr - - // Claude returns exit 1 if rate limited but still outputs JSON with is_error: true - err := cmd.Run() - - output := stdout.Bytes() - if len(output) == 0 { - if err != nil { - return nil, fmt.Errorf("claude failed: %w (stderr: %s)", err, stderr.String()) - } - return nil, fmt.Errorf("claude returned no output") - } - - var wrapper claudeJSONResult - if jerr := json.Unmarshal(output, &wrapper); jerr != nil { - return nil, fmt.Errorf("failed to parse claude JSON wrapper: %w (output: %s)", jerr, string(output)) - } - - if wrapper.IsError { - return nil, fmt.Errorf("claude error: %s", wrapper.Result) - } - - var result elaboratedTask - if jerr := json.Unmarshal([]byte(extractJSON(wrapper.Result)), &result); jerr != nil { - return nil, fmt.Errorf("failed to parse elaborated task JSON: %w (result: %s)", jerr, wrapper.Result) - } - - return &result, nil -} - -// elaborateWithLocal runs elaboration through an OpenAI-compatible local LLM. -// It uses the same prompt template as the Claude/Gemini paths and requests -// json_object response format so we can decode directly without the -// markdown-fence cleanup needed for the CLI paths. -func elaborateWithLocal(ctx context.Context, c *llm.Client, workDir, fullPrompt string) (*elaboratedTask, error) { - if c == nil { - return nil, fmt.Errorf("local llm: no client configured") - } - systemPrompt := buildElaboratePrompt(workDir) - resp, err := c.Chat(ctx, llm.ChatRequest{ - Messages: []llm.Message{ - {Role: "system", Content: systemPrompt}, - {Role: "user", Content: fullPrompt}, - }, - ResponseJSON: true, - }) - if err != nil { - return nil, fmt.Errorf("local llm: %w", err) - } - body := strings.TrimSpace(resp.Content) - var result elaboratedTask - if jerr := json.Unmarshal([]byte(extractJSON(body)), &result); jerr != nil { - return nil, fmt.Errorf("local llm: parse JSON: %w (response: %s)", jerr, body) - } - return &result, nil -} - -func (s *Server) elaborateWithGemini(ctx context.Context, workDir, fullPrompt string) (*elaboratedTask, error) { - combinedPrompt := fmt.Sprintf("%s\n\n%s", buildElaboratePrompt(workDir), fullPrompt) - cmd := exec.CommandContext(ctx, s.geminiBinaryPath(), - "-p", combinedPrompt, - "--output-format", "json", - "--model", "gemini-2.5-flash-lite", - ) - - var stdout, stderr bytes.Buffer - cmd.Stdout = &stdout - cmd.Stderr = &stderr - - if err := cmd.Run(); err != nil { - return nil, fmt.Errorf("gemini failed: %w (stderr: %s)", err, stderr.String()) - } - - var wrapper geminiJSONResult - if err := json.Unmarshal(stdout.Bytes(), &wrapper); err != nil { - return nil, fmt.Errorf("failed to parse gemini JSON wrapper: %w (output: %s)", err, stdout.String()) - } - - var result elaboratedTask - if err := json.Unmarshal([]byte(extractJSON(wrapper.Response)), &result); err != nil { - return nil, fmt.Errorf("failed to parse elaborated task JSON: %w (response: %s)", err, wrapper.Response) - } - - return &result, nil -} - // elaboratedStorySubtask is a leaf unit within a story task. type elaboratedStorySubtask struct { Name string `json:"name"` @@ -493,89 +242,3 @@ func (s *Server) handleElaborateStory(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, result) } - -func (s *Server) handleElaborateTask(w http.ResponseWriter, r *http.Request) { - if s.elaborateLimiter != nil && !s.elaborateLimiter.allow(realIP(r)) { - writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "rate limit exceeded"}) - return - } - - var input struct { - Prompt string `json:"prompt"` - ProjectID string `json:"project_id"` - // project_dir kept for backward compat; project_id takes precedence - ProjectDir string `json:"project_dir"` - } - if err := json.NewDecoder(r.Body).Decode(&input); err != nil { - writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON: " + err.Error()}) - return - } - if input.Prompt == "" { - writeJSON(w, http.StatusBadRequest, map[string]string{"error": "prompt is required"}) - return - } - - workDir := s.workDir - if input.ProjectID != "" { - if proj, err := s.store.GetProject(input.ProjectID); err == nil { - workDir = proj.LocalPath - } - } else if input.ProjectDir != "" { - workDir = input.ProjectDir - } - - if workDir != s.workDir { - go s.appendRawNarrative(workDir, input.Prompt) - } - - projectContext := readProjectContext(workDir) - fullPrompt := input.Prompt - if projectContext != "" { - fullPrompt = fmt.Sprintf("Project context from %s:\n%s\n\nUser request: %s", workDir, projectContext, input.Prompt) - } - - ctx, cancel := context.WithTimeout(r.Context(), elaborateTimeout) - defer cancel() - - var result *elaboratedTask - var err error - - // Try local LLM first when configured. Falls back to Claude → Gemini on - // hard failure of each prior attempt. - if s.llm != nil { - result, err = elaborateWithLocal(ctx, s.llm, workDir, fullPrompt) - if err != nil { - s.logger.Warn("elaborate: local llm failed, falling back to claude", "error", err) - result = nil - } - } - if result == nil { - result, err = s.elaborateWithClaude(ctx, workDir, fullPrompt) - if err != nil { - s.logger.Warn("elaborate: claude failed, falling back to gemini", "error", err) - result, err = s.elaborateWithGemini(ctx, workDir, fullPrompt) - if err != nil { - s.logger.Error("elaborate: gemini also failed", "error", err) - writeJSON(w, http.StatusBadGateway, map[string]string{ - "error": fmt.Sprintf("elaboration failed: %v", err), - }) - return - } - } - } - - if result.Name == "" || result.Agent.Instructions == "" { - writeJSON(w, http.StatusBadGateway, map[string]string{ - "error": "elaboration failed: missing required fields in response", - }) - return - } - - if result.Agent.Type == "" { - result.Agent.Type = "claude" - } - - sanitizeElaboratedTask(result) - - writeJSON(w, http.StatusOK, result) -} diff --git a/internal/api/elaborate_local_test.go b/internal/api/elaborate_local_test.go deleted file mode 100644 index 09a8f9e..0000000 --- a/internal/api/elaborate_local_test.go +++ /dev/null @@ -1,214 +0,0 @@ -package api - -import ( - "bytes" - "context" - "encoding/json" - "fmt" - "net/http" - "net/http/httptest" - "strings" - "sync/atomic" - "testing" - - "github.com/thepeterstone/claudomator/internal/llm" -) - -// fakeChatCompletionsServer returns an httptest server that responds to a -// /chat/completions POST with the given assistant content (which should be a -// JSON-encoded elaboratedTask). Returns the server and a counter of calls -// received so tests can assert dispatch ordering. -func fakeChatCompletionsServer(t *testing.T, assistantContent string) (*httptest.Server, *int32) { - t.Helper() - var calls int32 - srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - atomic.AddInt32(&calls, 1) - w.Header().Set("Content-Type", "application/json") - // The assistant content has to be JSON-encoded inside the wire format. - escaped, _ := json.Marshal(assistantContent) - fmt.Fprintf(w, `{ - "model":"local", - "choices":[{"message":{"role":"assistant","content":%s},"finish_reason":"stop"}], - "usage":{"prompt_tokens":10,"completion_tokens":50} - }`, string(escaped)) - })) - t.Cleanup(srv.Close) - return srv, &calls -} - -func TestElaborateWithLocal_ParsesValidResponse(t *testing.T) { - taskBody, _ := json.Marshal(elaboratedTask{ - Name: "Test elaborated task", - Description: "From local llm", - Agent: elaboratedAgent{ - Type: "claude", - Model: "sonnet", - Instructions: "Run go build.", - MaxBudgetUSD: 0.25, - AllowedTools: []string{"Bash"}, - }, - Timeout: "10m", - Priority: "normal", - Tags: []string{"build"}, - }) - srv, calls := fakeChatCompletionsServer(t, string(taskBody)) - - c := &llm.Client{Endpoint: srv.URL + "/v1", Model: "fake"} - result, err := elaborateWithLocal(context.Background(), c, "/some/dir", "build the project") - if err != nil { - t.Fatalf("elaborateWithLocal: %v", err) - } - if result.Name != "Test elaborated task" { - t.Errorf("Name: %q", result.Name) - } - if result.Agent.Instructions != "Run go build." { - t.Errorf("Instructions: %q", result.Agent.Instructions) - } - if got := atomic.LoadInt32(calls); got != 1 { - t.Errorf("expected 1 call, got %d", got) - } -} - -func TestElaborateWithLocal_NilClient(t *testing.T) { - _, err := elaborateWithLocal(context.Background(), nil, "", "p") - if err == nil || !strings.Contains(err.Error(), "no client") { - t.Errorf("expected nil-client error, got %v", err) - } -} - -func TestElaborateWithLocal_BadJSON(t *testing.T) { - srv, _ := fakeChatCompletionsServer(t, "this is not JSON at all") - c := &llm.Client{Endpoint: srv.URL + "/v1", Model: "fake"} - _, err := elaborateWithLocal(context.Background(), c, "", "p") - if err == nil || !strings.Contains(err.Error(), "parse JSON") { - t.Errorf("expected parse error, got %v", err) - } -} - -// TestElaborateTask_LocalLLMPreferred verifies the dispatcher uses local LLM -// when SetLLM is configured, and does not invoke claude. -func TestElaborateTask_LocalLLMPreferred(t *testing.T) { - srv, _ := testServer(t) - - taskBody, _ := json.Marshal(elaboratedTask{ - Name: "Local-elaborated", - Description: "From local", - Agent: elaboratedAgent{ - Type: "claude", - Model: "sonnet", - Instructions: "Do work. Tests pass when complete.", - MaxBudgetUSD: 0.25, - AllowedTools: []string{"Bash"}, - }, - Timeout: "10m", - Priority: "normal", - }) - llmSrv, _ := fakeChatCompletionsServer(t, string(taskBody)) - srv.SetLLM(&llm.Client{Endpoint: llmSrv.URL + "/v1", Model: "fake"}) - // Point Claude binary at a path that would fail if called. - srv.elaborateCmdPath = "/nonexistent/claude-should-not-run" - - body := `{"prompt":"do work"}` - req := httptest.NewRequest("POST", "/api/tasks/elaborate", bytes.NewBufferString(body)) - req.Header.Set("Content-Type", "application/json") - 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 got elaboratedTask - if err := json.NewDecoder(w.Body).Decode(&got); err != nil { - t.Fatalf("decode response: %v", err) - } - if got.Name != "Local-elaborated" { - t.Errorf("Name: want Local-elaborated got %q", got.Name) - } -} - -// TestElaborateTask_LocalFails_FallsBackToClaude verifies the dispatcher -// falls back to the Claude path when the local LLM returns an error. -func TestElaborateTask_LocalFails_FallsBackToClaude(t *testing.T) { - srv, _ := testServer(t) - - // Local LLM server that always 500s. - failSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - http.Error(w, "boom", http.StatusInternalServerError) - })) - t.Cleanup(failSrv.Close) - srv.SetLLM(&llm.Client{Endpoint: failSrv.URL + "/v1", Model: "fake"}) - - // Configure a working fake Claude binary. - taskBody, _ := json.Marshal(elaboratedTask{ - Name: "Claude-fallback", - Description: "From claude after local failed", - Agent: elaboratedAgent{ - Type: "claude", - Model: "sonnet", - Instructions: "Run tests.", - MaxBudgetUSD: 0.25, - AllowedTools: []string{"Bash"}, - }, - Timeout: "10m", - Priority: "normal", - }) - wrapper, _ := json.Marshal(map[string]string{"result": string(taskBody)}) - srv.elaborateCmdPath = createFakeClaude(t, string(wrapper), 0) - - body := `{"prompt":"run tests"}` - req := httptest.NewRequest("POST", "/api/tasks/elaborate", bytes.NewBufferString(body)) - req.Header.Set("Content-Type", "application/json") - 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 got elaboratedTask - if err := json.NewDecoder(w.Body).Decode(&got); err != nil { - t.Fatalf("decode response: %v", err) - } - if got.Name != "Claude-fallback" { - t.Errorf("Name: want Claude-fallback (fallback path) got %q", got.Name) - } -} - -// TestElaborateTask_NoLocalLLM_UsesClaude verifies that when SetLLM is not -// called, behavior is unchanged (Claude path still primary). -func TestElaborateTask_NoLocalLLM_UsesClaude(t *testing.T) { - srv, _ := testServer(t) - - taskBody, _ := json.Marshal(elaboratedTask{ - Name: "Claude-only", - Description: "no local llm configured", - Agent: elaboratedAgent{ - Type: "claude", - Model: "sonnet", - Instructions: "Do work.", - MaxBudgetUSD: 0.25, - AllowedTools: []string{"Bash"}, - }, - Timeout: "10m", - Priority: "normal", - }) - wrapper, _ := json.Marshal(map[string]string{"result": string(taskBody)}) - srv.elaborateCmdPath = createFakeClaude(t, string(wrapper), 0) - - body := `{"prompt":"do work"}` - req := httptest.NewRequest("POST", "/api/tasks/elaborate", bytes.NewBufferString(body)) - req.Header.Set("Content-Type", "application/json") - 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 got elaboratedTask - if err := json.NewDecoder(w.Body).Decode(&got); err != nil { - t.Fatalf("decode response: %v", err) - } - if got.Name != "Claude-only" { - t.Errorf("Name: %q", got.Name) - } -} - diff --git a/internal/api/elaborate_test.go b/internal/api/elaborate_test.go deleted file mode 100644 index 32cec3c..0000000 --- a/internal/api/elaborate_test.go +++ /dev/null @@ -1,527 +0,0 @@ -package api - -import ( - "bytes" - "encoding/json" - "fmt" - "net/http" - "net/http/httptest" - "os" - "path/filepath" - "strings" - "testing" - "time" -) - -// createFakeClaude writes a shell script to a temp dir that prints output and exits with the -// given code. Returns the script path. Used to mock the claude binary in elaborate tests. -func createFakeClaude(t *testing.T, output string, exitCode int) string { - t.Helper() - dir := t.TempDir() - outputFile := filepath.Join(dir, "output.json") - if err := os.WriteFile(outputFile, []byte(output), 0600); err != nil { - t.Fatal(err) - } - script := filepath.Join(dir, "claude") - content := fmt.Sprintf("#!/bin/sh\ncat %q\nexit %d\n", outputFile, exitCode) - if err := os.WriteFile(script, []byte(content), 0755); err != nil { - t.Fatal(err) - } - return script -} - -// hasTool is a test helper that reports whether name is in the tools slice. -func hasTool(tools []string, name string) bool { - for _, t := range tools { - if t == name { - return true - } - } - return false -} - -// --- sanitizeElaboratedTask unit tests --- - -func TestSanitize_AddsWriteWhenInstructionsMentionFileCreation(t *testing.T) { - task := &elaboratedTask{ - Agent: elaboratedAgent{ - Instructions: "Create a new file called output.txt with the results.", - AllowedTools: []string{"Bash"}, - }, - } - sanitizeElaboratedTask(task) - if !hasTool(task.Agent.AllowedTools, "Write") { - t.Errorf("expected Write in allowed_tools, got %v", task.Agent.AllowedTools) - } -} - -func TestSanitize_AddsReadWhenEditIsPresent(t *testing.T) { - task := &elaboratedTask{ - Agent: elaboratedAgent{ - Instructions: "Modify the configuration file.", - AllowedTools: []string{"Edit"}, - }, - } - sanitizeElaboratedTask(task) - if !hasTool(task.Agent.AllowedTools, "Read") { - t.Errorf("expected Read added alongside Edit, got %v", task.Agent.AllowedTools) - } -} - -func TestSanitize_NoDuplicateTools(t *testing.T) { - task := &elaboratedTask{ - Agent: elaboratedAgent{ - Instructions: "Run go test ./...", - AllowedTools: []string{"Bash"}, - }, - } - sanitizeElaboratedTask(task) - count := 0 - for _, tool := range task.Agent.AllowedTools { - if tool == "Bash" { - count++ - } - } - if count != 1 { - t.Errorf("Bash duplicated in allowed_tools: %v", task.Agent.AllowedTools) - } -} - -func TestSanitize_AddsAcceptanceCriteriaWhenMissing(t *testing.T) { - task := &elaboratedTask{ - Agent: elaboratedAgent{ - Instructions: "Do something useful with the codebase.", - AllowedTools: []string{"Bash"}, - }, - } - sanitizeElaboratedTask(task) - lower := strings.ToLower(task.Agent.Instructions) - if !strings.Contains(lower, "acceptance") && !strings.Contains(lower, "done when") { - t.Error("expected acceptance criteria section appended to instructions") - } -} - -func TestSanitize_NoopWhenAcceptanceCriteriaAlreadyPresent(t *testing.T) { - original := "Do something.\n\n## Acceptance Criteria\n- All tests pass." - task := &elaboratedTask{ - Agent: elaboratedAgent{ - Instructions: original, - AllowedTools: []string{"Bash"}, - }, - } - sanitizeElaboratedTask(task) - if task.Agent.Instructions != original { - t.Errorf("instructions were modified when acceptance criteria were already present") - } -} - -func TestSanitize_AddsTDDReminderForCodingTaskWithoutTestMention(t *testing.T) { - task := &elaboratedTask{ - Agent: elaboratedAgent{ - Instructions: "## Acceptance Criteria\nFix the bug.\n\nModify the handler to return 404 instead of 500.", - AllowedTools: []string{"Edit", "Read"}, - }, - } - sanitizeElaboratedTask(task) - lower := strings.ToLower(task.Agent.Instructions) - if !strings.Contains(lower, "tdd") && !strings.Contains(lower, "test") { - t.Error("expected TDD reminder for coding task without test mention") - } -} - -func TestSanitize_NoTDDReminderWhenTestsAlreadyMentioned(t *testing.T) { - original := "## Acceptance Criteria\nAll tests pass.\n\nEdit the file and run go test ./... to verify." - task := &elaboratedTask{ - Agent: elaboratedAgent{ - Instructions: original, - AllowedTools: []string{"Edit", "Read", "Bash"}, - }, - } - before := task.Agent.Instructions - sanitizeElaboratedTask(task) - // Should NOT add a second TDD block since tests are already mentioned. - // Count occurrences of "tdd" / "test" — just verify no double-append. - if strings.Count(strings.ToLower(task.Agent.Instructions), "tdd") > 1 { - t.Errorf("TDD block added twice; instructions:\n%s", task.Agent.Instructions) - } - _ = before -} - -func TestElaboratePrompt_RequiresAcceptanceCriteria(t *testing.T) { - prompt := buildElaboratePrompt("") - lower := strings.ToLower(prompt) - if !strings.Contains(lower, "acceptance criteria") { - t.Error("elaborate prompt should instruct the model to include acceptance criteria") - } -} - -func TestElaboratePrompt_RequiresAllRelevantTools(t *testing.T) { - prompt := buildElaboratePrompt("") - // Prompt must remind the model to include file-creating tools when needed. - if !strings.Contains(prompt, "Write") { - t.Error("elaborate prompt should mention the Write tool so models know to include it") - } -} - -func TestElaborateTask_SanitizationAppliedToResponse(t *testing.T) { - srv, _ := testServer(t) - - // Elaborator returns a task that needs Write (instructions say "create file") - // but does NOT include it in allowed_tools. - task := elaboratedTask{ - Name: "Generate report", - Description: "Creates a report file.", - Agent: elaboratedAgent{ - Type: "claude", - Model: "sonnet", - Instructions: "Create a new file called report.md with the analysis results.\n\n## Acceptance Criteria\n- report.md exists.", - MaxBudgetUSD: 0.5, - AllowedTools: []string{"Bash"}, // Write intentionally missing - }, - Timeout: "15m", - Priority: "normal", - Tags: []string{"report"}, - } - taskJSON, _ := json.Marshal(task) - wrapper := map[string]string{"result": string(taskJSON)} - wrapperJSON, _ := json.Marshal(wrapper) - - srv.elaborateCmdPath = createFakeClaude(t, string(wrapperJSON), 0) - - body := `{"prompt":"generate a report"}` - req := httptest.NewRequest("POST", "/api/tasks/elaborate", bytes.NewBufferString(body)) - req.Header.Set("Content-Type", "application/json") - 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 result elaboratedTask - if err := json.NewDecoder(w.Body).Decode(&result); err != nil { - t.Fatalf("failed to decode response: %v", err) - } - if !hasTool(result.Agent.AllowedTools, "Write") { - t.Errorf("expected Write in sanitized allowed_tools, got %v", result.Agent.AllowedTools) - } -} - -func TestElaboratePrompt_ContainsWorkDir(t *testing.T) { - prompt := buildElaboratePrompt("/some/custom/path") - if !strings.Contains(prompt, "/some/custom/path") { - t.Error("prompt should contain the provided workDir") - } - if strings.Contains(prompt, "/root/workspace/claudomator") { - t.Error("prompt should not hardcode /root/workspace/claudomator") - } -} - -func TestElaboratePrompt_EmptyWorkDir(t *testing.T) { - prompt := buildElaboratePrompt("") - if strings.Contains(prompt, "/root") { - t.Error("prompt should not reference /root when workDir is empty") - } -} - -func TestElaborateTask_Success(t *testing.T) { - srv, _ := testServer(t) - - // Build fake Claude output: {"result": ""} - task := elaboratedTask{ - Name: "Run Go tests with race detector", - Description: "Runs the Go test suite with -race flag and checks coverage.", - Agent: elaboratedAgent{ - Type: "claude", - Model: "sonnet", - Instructions: "Run go test -race ./... and report results.", - ProjectDir: "", - MaxBudgetUSD: 0.5, - AllowedTools: []string{"Bash"}, - }, - Timeout: "15m", - Priority: "normal", - Tags: []string{"testing", "ci"}, - } - taskJSON, err := json.Marshal(task) - if err != nil { - t.Fatal(err) - } - wrapper := map[string]string{"result": string(taskJSON)} - wrapperJSON, err := json.Marshal(wrapper) - if err != nil { - t.Fatal(err) - } - - srv.elaborateCmdPath = createFakeClaude(t, string(wrapperJSON), 0) - - body := `{"prompt":"run the Go test suite with race detector and fail if coverage < 80%"}` - req := httptest.NewRequest("POST", "/api/tasks/elaborate", bytes.NewBufferString(body)) - req.Header.Set("Content-Type", "application/json") - 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 result elaboratedTask - if err := json.NewDecoder(w.Body).Decode(&result); err != nil { - t.Fatalf("failed to decode response: %v", err) - } - if result.Name == "" { - t.Error("expected non-empty name") - } - if result.Agent.Instructions == "" { - t.Error("expected non-empty instructions") - } -} - -func TestElaborateTask_EmptyPrompt(t *testing.T) { - srv, _ := testServer(t) - - body := `{"prompt":""}` - req := httptest.NewRequest("POST", "/api/tasks/elaborate", bytes.NewBufferString(body)) - req.Header.Set("Content-Type", "application/json") - w := httptest.NewRecorder() - - srv.Handler().ServeHTTP(w, req) - - if w.Code != http.StatusBadRequest { - t.Fatalf("status: want 400, got %d; body: %s", w.Code, w.Body.String()) - } - - var resp map[string]string - json.NewDecoder(w.Body).Decode(&resp) - if resp["error"] == "" { - t.Error("expected error message in response") - } -} - -func TestElaborateTask_MarkdownFencedJSON(t *testing.T) { - srv, _ := testServer(t) - - // Build a valid task JSON but wrap it in markdown fences as haiku sometimes does. - task := elaboratedTask{ - Name: "Test task", - Description: "Does something.", - Agent: elaboratedAgent{ - Type: "claude", - Model: "sonnet", - Instructions: "Do the thing.", - MaxBudgetUSD: 0.5, - AllowedTools: []string{"Bash"}, - }, - Timeout: "15m", - Priority: "normal", - Tags: []string{"test"}, - } - taskJSON, _ := json.Marshal(task) - fenced := "```json\n" + string(taskJSON) + "\n```" - wrapper := map[string]string{"result": fenced} - wrapperJSON, _ := json.Marshal(wrapper) - - srv.elaborateCmdPath = createFakeClaude(t, string(wrapperJSON), 0) - - body := `{"prompt":"do something"}` - req := httptest.NewRequest("POST", "/api/tasks/elaborate", bytes.NewBufferString(body)) - req.Header.Set("Content-Type", "application/json") - 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 result elaboratedTask - if err := json.NewDecoder(w.Body).Decode(&result); err != nil { - t.Fatalf("failed to decode response: %v", err) - } - if result.Name != task.Name { - t.Errorf("name: want %q, got %q", task.Name, result.Name) - } -} - -func TestElaborateTask_InvalidJSONFromClaude(t *testing.T) { - srv, _ := testServer(t) - - // Fake Claude returns something that is not valid JSON. - srv.elaborateCmdPath = createFakeClaude(t, "not valid json at all", 0) - // Ensure Gemini fallback also fails so we get the expected 502. - srv.geminiBinPath = "/nonexistent/gemini" - - body := `{"prompt":"do something"}` - req := httptest.NewRequest("POST", "/api/tasks/elaborate", bytes.NewBufferString(body)) - req.Header.Set("Content-Type", "application/json") - w := httptest.NewRecorder() - - srv.Handler().ServeHTTP(w, req) - - if w.Code != http.StatusBadGateway { - t.Fatalf("status: want 502, got %d; body: %s", w.Code, w.Body.String()) - } - - var resp map[string]string - json.NewDecoder(w.Body).Decode(&resp) - if resp["error"] == "" { - t.Error("expected error message in response") - } -} - -func createFakeClaudeCapturingArgs(t *testing.T, output string, exitCode int, argsFile string) string { - t.Helper() - dir := t.TempDir() - outputFile := filepath.Join(dir, "output.json") - if err := os.WriteFile(outputFile, []byte(output), 0600); err != nil { - t.Fatal(err) - } - script := filepath.Join(dir, "claude") - // Use printf to handle arguments safely - content := fmt.Sprintf("#!/bin/sh\nprintf \"%%s\\n\" \"$@\" > %q\ncat %q\nexit %d\n", argsFile, outputFile, exitCode) - if err := os.WriteFile(script, []byte(content), 0755); err != nil { - t.Fatal(err) - } - return script -} - -func TestElaborateTask_WithProjectContext(t *testing.T) { - srv, _ := testServer(t) - - // Create a temporary workspace with CLAUDE.md and .agent/worklog.md - workDir := t.TempDir() - claudeContent := "Claude context info" - sessionContent := "Session state info" - if err := os.WriteFile(filepath.Join(workDir, "CLAUDE.md"), []byte(claudeContent), 0600); err != nil { - t.Fatal(err) - } - if err := os.MkdirAll(filepath.Join(workDir, ".agent"), 0700); err != nil { - t.Fatal(err) - } - if err := os.WriteFile(filepath.Join(workDir, ".agent", "worklog.md"), []byte(sessionContent), 0600); err != nil { - t.Fatal(err) - } - - // Capture arguments passed to claude - argsFile := filepath.Join(t.TempDir(), "args.txt") - - task := elaboratedTask{ - Name: "Task with context", - Agent: elaboratedAgent{ - Instructions: "Instructions", - }, - } - taskJSON, _ := json.Marshal(task) - wrapper := map[string]string{"result": string(taskJSON)} - wrapperJSON, _ := json.Marshal(wrapper) - - // Modified createFakeClaude to capture arguments - srv.elaborateCmdPath = createFakeClaudeCapturingArgs(t, string(wrapperJSON), 0, argsFile) - - body := fmt.Sprintf(`{"prompt":"do something", "project_dir":"%s"}`, workDir) - req := httptest.NewRequest("POST", "/api/tasks/elaborate", bytes.NewBufferString(body)) - req.Header.Set("Content-Type", "application/json") - 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()) - } - - // Check if captured arguments contain the context - capturedArgs, err := os.ReadFile(argsFile) - if err != nil { - t.Fatal(err) - } - argsStr := string(capturedArgs) - if !strings.Contains(argsStr, claudeContent) { - t.Errorf("expected arguments to contain CLAUDE.md content, got %s", argsStr) - } - if !strings.Contains(argsStr, sessionContent) { - t.Errorf("expected arguments to contain .agent/worklog.md content, got %s", argsStr) - } -} - -func TestElaborateTask_NoRawNarrativeWithoutExplicitProjectDir(t *testing.T) { - srv, _ := testServer(t) - // Point workDir at a temp dir so any accidental write is detectable. - srv.workDir = t.TempDir() - - task := elaboratedTask{ - Name: "Task", - Agent: elaboratedAgent{Instructions: "Instructions"}, - } - taskJSON, _ := json.Marshal(task) - wrapper := map[string]string{"result": string(taskJSON)} - wrapperJSON, _ := json.Marshal(wrapper) - srv.elaborateCmdPath = createFakeClaude(t, string(wrapperJSON), 0) - - // No project_dir in request body. - body := `{"prompt":"do something"}` - req := httptest.NewRequest("POST", "/api/tasks/elaborate", bytes.NewBufferString(body)) - req.Header.Set("Content-Type", "application/json") - w := httptest.NewRecorder() - srv.Handler().ServeHTTP(w, req) - - if w.Code != http.StatusOK { - t.Fatalf("status: want 200, got %d", w.Code) - } - - time.Sleep(30 * time.Millisecond) // let goroutine run if it was incorrectly triggered - narrativePath := filepath.Join(srv.workDir, "docs", "RAW_NARRATIVE.md") - if _, err := os.Stat(narrativePath); err == nil { - t.Errorf("RAW_NARRATIVE.md should NOT be written when project_dir is not provided by the user") - } -} - -func TestElaborateTask_AppendsRawNarrative(t *testing.T) { - srv, _ := testServer(t) - - workDir := t.TempDir() - prompt := "this is my raw request" - - task := elaboratedTask{ - Name: "Task", - Agent: elaboratedAgent{ - Instructions: "Instructions", - }, - } - taskJSON, _ := json.Marshal(task) - wrapper := map[string]string{"result": string(taskJSON)} - wrapperJSON, _ := json.Marshal(wrapper) - - srv.elaborateCmdPath = createFakeClaude(t, string(wrapperJSON), 0) - - body := fmt.Sprintf(`{"prompt":"%s", "project_dir":"%s"}`, prompt, workDir) - req := httptest.NewRequest("POST", "/api/tasks/elaborate", bytes.NewBufferString(body)) - req.Header.Set("Content-Type", "application/json") - 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()) - } - - // It runs in a goroutine, so wait a bit - path := filepath.Join(workDir, "docs", "RAW_NARRATIVE.md") - var data []byte - var err error - for i := 0; i < 10; i++ { - data, err = os.ReadFile(path) - if err == nil { - break - } - time.Sleep(10 * time.Millisecond) - } - - if err != nil { - t.Fatalf("failed to read RAW_NARRATIVE.md: %v", err) - } - if !strings.Contains(string(data), prompt) { - t.Errorf("expected RAW_NARRATIVE.md to contain prompt, got %s", string(data)) - } -} diff --git a/internal/api/server.go b/internal/api/server.go index c9fadb9..6564280 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -133,7 +133,6 @@ func (s *Server) StartHub() { } func (s *Server) routes() { - s.mux.HandleFunc("POST /api/tasks/elaborate", s.handleElaborateTask) s.mux.HandleFunc("POST /api/tasks/validate", s.handleValidateTask) s.mux.HandleFunc("POST /api/tasks", s.handleCreateTask) s.mux.HandleFunc("GET /api/tasks", s.handleListTasks) diff --git a/internal/api/server_test.go b/internal/api/server_test.go index dd6eed5..43d91b0 100644 --- a/internal/api/server_test.go +++ b/internal/api/server_test.go @@ -1228,34 +1228,6 @@ func TestServer_AnswerQuestion_UpdateStateFails_Returns500(t *testing.T) { } } -func TestRateLimit_ElaborateRejectsExcess(t *testing.T) { - srv, _ := testServer(t) - // Use burst-1 and rate-0 so the second request from the same IP is rejected. - srv.elaborateLimiter = newIPRateLimiter(0, 1) - - makeReq := func(remoteAddr string) int { - req := httptest.NewRequest("POST", "/api/tasks/elaborate", bytes.NewBufferString(`{"description":"x"}`)) - req.Header.Set("Content-Type", "application/json") - req.RemoteAddr = remoteAddr - w := httptest.NewRecorder() - srv.Handler().ServeHTTP(w, req) - return w.Code - } - - // First request from IP A: limiter allows it (non-429). - if code := makeReq("192.0.2.1:1234"); code == http.StatusTooManyRequests { - t.Errorf("first request should not be rate limited, got 429") - } - // Second request from IP A: bucket exhausted, must be 429. - if code := makeReq("192.0.2.1:1234"); code != http.StatusTooManyRequests { - t.Errorf("second request from same IP should be 429, got %d", code) - } - // First request from IP B: separate bucket, not limited. - if code := makeReq("192.0.2.2:1234"); code == http.StatusTooManyRequests { - t.Errorf("first request from different IP should not be rate limited, got 429") - } -} - func TestListWorkspaces_RequiresAuth(t *testing.T) { srv, _ := testServer(t) srv.SetAPIToken("secret-token") diff --git a/internal/api/validate_test.go b/internal/api/validate_test.go index c3d7b1f..60fec14 100644 --- a/internal/api/validate_test.go +++ b/internal/api/validate_test.go @@ -3,11 +3,32 @@ package api import ( "bytes" "encoding/json" + "fmt" "net/http" "net/http/httptest" + "os" + "path/filepath" "testing" ) +// createFakeClaude writes a stub `claude` script that prints the given output +// and exits with the given code, returning its path for use as a binary +// override in tests. +func createFakeClaude(t *testing.T, output string, exitCode int) string { + t.Helper() + dir := t.TempDir() + outputFile := filepath.Join(dir, "output.json") + if err := os.WriteFile(outputFile, []byte(output), 0600); err != nil { + t.Fatal(err) + } + script := filepath.Join(dir, "claude") + content := fmt.Sprintf("#!/bin/sh\ncat %q\nexit %d\n", outputFile, exitCode) + if err := os.WriteFile(script, []byte(content), 0755); err != nil { + t.Fatal(err) + } + return script +} + func TestValidateTask_Success(t *testing.T) { srv, _ := testServer(t) -- cgit v1.2.3 From b44c8277465c3b4e03fe4de4dd93850413023b69 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 25 May 2026 05:05:27 +0000 Subject: feat(api,web): event-stream timeline (Phase 3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds GET /api/tasks/{id}/events (seq-ordered, ?since_seq for incremental polling) as the timeline data source, and a per-task Timeline section in the web UI that fetches and renders the observability event stream. New pure, unit-tested JS helpers: formatEventText, renderEventTimeline, fetchTaskEvents. The timeline load is async and guarded on a global fetch so the synchronous renderTaskPanel path (and its DOM-mock unit tests) is unaffected. Verified via Go endpoint tests and node --test (272 JS tests pass). Note: the live in-browser panel render was not exercised in this environment — only unit-level coverage. https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39 --- internal/api/events.go | 38 +++++++++++++ internal/api/events_test.go | 94 ++++++++++++++++++++++++++++++++ internal/api/server.go | 1 + web/app.js | 98 ++++++++++++++++++++++++++++++++++ web/style.css | 45 ++++++++++++++++ web/test/event-timeline.test.mjs | 112 +++++++++++++++++++++++++++++++++++++++ 6 files changed, 388 insertions(+) create mode 100644 internal/api/events.go create mode 100644 internal/api/events_test.go create mode 100644 web/test/event-timeline.test.mjs (limited to 'internal/api') diff --git a/internal/api/events.go b/internal/api/events.go new file mode 100644 index 0000000..eefe064 --- /dev/null +++ b/internal/api/events.go @@ -0,0 +1,38 @@ +package api + +import ( + "net/http" + "strconv" + + "github.com/thepeterstone/claudomator/internal/event" +) + +// handleListTaskEvents returns a task's observability event stream in seq order. +// Pass ?since_seq=N to fetch only events newer than seq N (incremental polling). +func (s *Server) handleListTaskEvents(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + if _, err := s.store.GetTask(id); err != nil { + writeJSON(w, http.StatusNotFound, map[string]string{"error": "task not found"}) + return + } + + var sinceSeq int64 + if v := r.URL.Query().Get("since_seq"); v != "" { + n, err := strconv.ParseInt(v, 10, 64) + if err != nil || n < 0 { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid since_seq"}) + return + } + sinceSeq = n + } + + events, err := s.store.ListEvents(id, sinceSeq) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) + return + } + if events == nil { + events = []*event.Event{} + } + writeJSON(w, http.StatusOK, events) +} diff --git a/internal/api/events_test.go b/internal/api/events_test.go new file mode 100644 index 0000000..456b6d8 --- /dev/null +++ b/internal/api/events_test.go @@ -0,0 +1,94 @@ +package api + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strconv" + "testing" + + "github.com/thepeterstone/claudomator/internal/event" + "github.com/thepeterstone/claudomator/internal/task" +) + +func TestListTaskEvents_ReturnsStreamInSeqOrder(t *testing.T) { + srv, store := testServer(t) + createTaskWithState(t, store, "ev-task", task.StateReady) + + req := httptest.NewRequest("GET", "/api/tasks/ev-task/events", 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 events []*event.Event + if err := json.Unmarshal(w.Body.Bytes(), &events); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if len(events) == 0 { + t.Fatal("expected state-change events from the walk to READY") + } + for i := 1; i < len(events); i++ { + if events[i].Seq <= events[i-1].Seq { + t.Errorf("events not in ascending seq order: %d then %d", events[i-1].Seq, events[i].Seq) + } + } +} + +func TestListTaskEvents_SinceSeqFilters(t *testing.T) { + srv, store := testServer(t) + createTaskWithState(t, store, "ev-since", task.StateReady) + + all := getEvents(t, srv, "ev-since", "") + if len(all) < 2 { + t.Skipf("need at least 2 events to test since_seq, got %d", len(all)) + } + firstSeq := all[0].Seq + rest := getEvents(t, srv, "ev-since", "?since_seq="+itoa(firstSeq)) + for _, e := range rest { + if e.Seq <= firstSeq { + t.Errorf("since_seq=%d returned event with seq %d", firstSeq, e.Seq) + } + } +} + +func TestListTaskEvents_NotFound(t *testing.T) { + srv, _ := testServer(t) + req := httptest.NewRequest("GET", "/api/tasks/missing/events", nil) + w := httptest.NewRecorder() + srv.Handler().ServeHTTP(w, req) + if w.Code != http.StatusNotFound { + t.Errorf("want 404, got %d", w.Code) + } +} + +func TestListTaskEvents_InvalidSinceSeq(t *testing.T) { + srv, store := testServer(t) + createTaskWithState(t, store, "ev-bad", task.StateReady) + req := httptest.NewRequest("GET", "/api/tasks/ev-bad/events?since_seq=abc", nil) + w := httptest.NewRecorder() + srv.Handler().ServeHTTP(w, req) + if w.Code != http.StatusBadRequest { + t.Errorf("want 400, got %d", w.Code) + } +} + +func getEvents(t *testing.T, srv *Server, taskID, query string) []*event.Event { + t.Helper() + req := httptest.NewRequest("GET", "/api/tasks/"+taskID+"/events"+query, nil) + w := httptest.NewRecorder() + srv.Handler().ServeHTTP(w, req) + if w.Code != http.StatusOK { + t.Fatalf("getEvents %s%s: status %d", taskID, query, w.Code) + } + var events []*event.Event + if err := json.Unmarshal(w.Body.Bytes(), &events); err != nil { + t.Fatalf("getEvents unmarshal: %v", err) + } + return events +} + +func itoa(n int64) string { + return strconv.FormatInt(n, 10) +} diff --git a/internal/api/server.go b/internal/api/server.go index 6564280..a4b7ea1 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -144,6 +144,7 @@ func (s *Server) routes() { s.mux.HandleFunc("DELETE /api/tasks/{id}", s.handleDeleteTask) s.mux.HandleFunc("GET /api/tasks/{id}/subtasks", s.handleListSubtasks) s.mux.HandleFunc("GET /api/tasks/{id}/executions", s.handleListExecutions) + s.mux.HandleFunc("GET /api/tasks/{id}/events", s.handleListTaskEvents) s.mux.HandleFunc("GET /api/executions", s.handleListRecentExecutions) s.mux.HandleFunc("GET /api/stats", s.handleGetDashboardStats) s.mux.HandleFunc("GET /api/agents/status", s.handleGetAgentStatus) diff --git a/web/app.js b/web/app.js index ff8e381..09b545b 100644 --- a/web/app.js +++ b/web/app.js @@ -115,6 +115,89 @@ export function renderDeploymentBadge(status, doc = (typeof document !== 'undefi return span; } +// ── Event timeline ────────────────────────────────────────────────────────── +// The observability event stream (GET /api/tasks/{id}/events) replaces the +// ad-hoc question/summary panels. formatEventText turns one event into a human +// line; renderEventTimeline builds the list element. + +// formatEventText returns a one-line human description of a task event. +export function formatEventText(event) { + if (!event) return ''; + const p = event.payload || {}; + switch (event.kind) { + case 'state_change': + return `State: ${p.from || '?'} → ${p.to || '?'}`; + case 'summary': + return `Summary: ${p.text || p.summary || ''}`; + case 'clarification_request': + return `Question: ${p.text || ''}`; + case 'clarification_answer': + return `Answer: ${p.answer || ''}`; + case 'subtask_spawned': + return `Spawned subtask: ${p.name || p.subtask_id || ''}`; + case 'commit_made': + return `Commit: ${p.message || p.hash || ''}`; + case 'cost_report': + return `Cost: $${p.cost_usd != null ? p.cost_usd : 0}`; + case 'agent_message': + case 'human_message': + return p.text || p.message || ''; + case 'execution_started': + return 'Execution started'; + case 'execution_ended': + return p.status ? `Execution ended (${p.status})` : 'Execution ended'; + default: + return event.kind || ''; + } +} + +// renderEventTimeline builds a
    from an events array. +// Accepts an optional doc parameter for testability (defaults to document). +export function renderEventTimeline(events, doc = (typeof document !== 'undefined' ? document : null)) { + if (doc == null) return null; + const ul = doc.createElement('ul'); + ul.className = 'event-timeline'; + if (!events || events.length === 0) { + const empty = doc.createElement('li'); + empty.className = 'event-timeline__empty'; + empty.textContent = 'No events yet.'; + ul.appendChild(empty); + return ul; + } + for (const event of events) { + const li = doc.createElement('li'); + li.className = `event-timeline__item event-timeline__item--${event.kind || 'unknown'}`; + + const actor = doc.createElement('span'); + actor.className = 'event-timeline__actor'; + actor.textContent = event.actor || ''; + li.appendChild(actor); + + const text = doc.createElement('span'); + text.className = 'event-timeline__text'; + text.textContent = formatEventText(event); + li.appendChild(text); + + ul.appendChild(li); + } + return ul; +} + +// fetchTaskEvents loads a task's event stream. fetchImpl defaults to the global +// fetch; pass a stub in tests. Returns [] on any error so the UI degrades +// gracefully. +export async function fetchTaskEvents(taskId, fetchImpl = (typeof fetch !== 'undefined' ? fetch : null)) { + if (!fetchImpl) return []; + try { + const resp = await fetchImpl(`/api/tasks/${taskId}/events`); + if (!resp.ok) return []; + const data = await resp.json(); + return Array.isArray(data) ? data : []; + } catch { + return []; + } +} + function truncateToWordBoundary(text, maxLen = 120) { if (!text || text.length <= maxLen) return text; const cut = text.lastIndexOf(' ', maxLen); @@ -2055,6 +2138,21 @@ export function renderTaskPanel(task, executions) { execSection.appendChild(list); } content.appendChild(execSection); + + // ── Timeline ── + // Observability event stream, loaded asynchronously. Guarded so the sync + // panel render (and its unit tests, which have no fetch) is unaffected. + const timelineSection = makeSection('Timeline'); + const timelineContainer = document.createElement('div'); + timelineContainer.className = 'event-timeline-container'; + timelineSection.appendChild(timelineContainer); + content.appendChild(timelineSection); + if (typeof fetch !== 'undefined') { + fetchTaskEvents(task.id).then(events => { + timelineContainer.innerHTML = ''; + timelineContainer.appendChild(renderEventTimeline(events)); + }); + } } async function handleViewLogs(execId) { diff --git a/web/style.css b/web/style.css index d3b01d0..f4a9d91 100644 --- a/web/style.css +++ b/web/style.css @@ -1989,3 +1989,48 @@ dialog label select:focus { opacity: 0.85; padding: 0.1rem 0; } + +/* Event timeline (task observability stream) */ +.event-timeline { + list-style: none; + margin: 0; + padding: 0; + display: flex; + flex-direction: column; + gap: 6px; +} +.event-timeline__item { + display: flex; + align-items: baseline; + gap: 10px; + padding: 6px 10px; + background: var(--bg); + border: 1px solid var(--border); + border-left: 3px solid var(--border); + border-radius: 6px; + font-size: 0.85rem; +} +.event-timeline__item--state_change { border-left-color: var(--state-queued); } +.event-timeline__item--clarification_request { border-left-color: var(--state-blocked); } +.event-timeline__item--clarification_answer { border-left-color: var(--state-blocked); } +.event-timeline__item--summary { border-left-color: var(--state-completed); } +.event-timeline__item--subtask_spawned { border-left-color: var(--accent); } +.event-timeline__item--execution_started, +.event-timeline__item--execution_ended { border-left-color: var(--state-running); } +.event-timeline__actor { + flex: 0 0 auto; + text-transform: uppercase; + font-size: 0.65rem; + letter-spacing: 0.04em; + color: var(--text-muted); + min-width: 56px; +} +.event-timeline__text { + color: var(--text); + word-break: break-word; +} +.event-timeline__empty { + color: var(--text-muted); + font-size: 0.85rem; + padding: 6px 10px; +} diff --git a/web/test/event-timeline.test.mjs b/web/test/event-timeline.test.mjs new file mode 100644 index 0000000..20ef61a --- /dev/null +++ b/web/test/event-timeline.test.mjs @@ -0,0 +1,112 @@ +// event-timeline.test.mjs — Unit tests for the event timeline component. +// +// Run with: node --test web/test/event-timeline.test.mjs + +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { formatEventText, renderEventTimeline, fetchTaskEvents } from '../app.js'; + +function makeDoc() { + return { + createElement(tag) { + return { + tag, + className: '', + textContent: '', + children: [], + appendChild(child) { this.children.push(child); return child; }, + }; + }, + }; +} + +describe('formatEventText', () => { + it('formats state_change with from/to', () => { + assert.equal( + formatEventText({ kind: 'state_change', payload: { from: 'QUEUED', to: 'RUNNING' } }), + 'State: QUEUED → RUNNING', + ); + }); + + it('formats summary text', () => { + assert.equal( + formatEventText({ kind: 'summary', payload: { text: 'did the thing' } }), + 'Summary: did the thing', + ); + }); + + it('formats clarification_request and answer', () => { + assert.equal( + formatEventText({ kind: 'clarification_request', payload: { text: 'which branch?' } }), + 'Question: which branch?', + ); + assert.equal( + formatEventText({ kind: 'clarification_answer', payload: { answer: 'main' } }), + 'Answer: main', + ); + }); + + it('formats subtask_spawned by name', () => { + assert.equal( + formatEventText({ kind: 'subtask_spawned', payload: { name: 'write tests', subtask_id: 'x' } }), + 'Spawned subtask: write tests', + ); + }); + + it('falls back to kind for unknown events', () => { + assert.equal(formatEventText({ kind: 'mystery', payload: {} }), 'mystery'); + }); + + it('handles null/empty defensively', () => { + assert.equal(formatEventText(null), ''); + assert.equal(formatEventText({ kind: 'agent_message', payload: { message: 'hi' } }), 'hi'); + }); +}); + +describe('renderEventTimeline', () => { + it('renders one item per event with actor and text', () => { + const events = [ + { kind: 'state_change', actor: 'system', payload: { from: 'PENDING', to: 'QUEUED' } }, + { kind: 'summary', actor: 'agent', payload: { text: 'done' } }, + ]; + const ul = renderEventTimeline(events, makeDoc()); + assert.equal(ul.className, 'event-timeline'); + assert.equal(ul.children.length, 2); + // each item has [actor, text] + assert.equal(ul.children[0].children[0].textContent, 'system'); + assert.equal(ul.children[0].children[1].textContent, 'State: PENDING → QUEUED'); + assert.equal(ul.children[1].children[1].textContent, 'Summary: done'); + }); + + it('renders an empty-state item for no events', () => { + const ul = renderEventTimeline([], makeDoc()); + assert.equal(ul.children.length, 1); + assert.equal(ul.children[0].className, 'event-timeline__empty'); + }); + + it('returns null when doc is null', () => { + assert.equal(renderEventTimeline([], null), null); + }); +}); + +describe('fetchTaskEvents', () => { + it('returns parsed array on success', async () => { + const stub = async () => ({ ok: true, json: async () => [{ kind: 'summary', payload: {} }] }); + const events = await fetchTaskEvents('t1', stub); + assert.equal(events.length, 1); + }); + + it('returns [] on non-ok response', async () => { + const stub = async () => ({ ok: false, json: async () => ({}) }); + assert.deepEqual(await fetchTaskEvents('t1', stub), []); + }); + + it('returns [] when fetch throws', async () => { + const stub = async () => { throw new Error('network'); }; + assert.deepEqual(await fetchTaskEvents('t1', stub), []); + }); + + it('returns [] when no fetch implementation is available', async () => { + assert.deepEqual(await fetchTaskEvents('t1', null), []); + }); +}); -- cgit v1.2.3 From ab4b364954af08fa602388495ca425eaef0abf74 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 26 May 2026 20:34:09 +0000 Subject: feat(api,web): budget headroom endpoint + UI chips (Phase 6) Adds GET /api/budget returning per-provider rolling-window headroom (empty when budget gating is unconfigured), wired from serve.go via SetBudget. The web UI polls it and renders a chip per limited provider in the header, flagging any under 20% remaining. New pure JS helpers formatBudgetHeadroom/ renderBudgetHeadroom are unit-tested; the endpoint is covered by Go handler tests (empty/reports/500). UI render not browser-verified in this environment. Completes Phase 6: spend accounting + dispatcher gating + observability surface. https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39 --- internal/api/budget.go | 36 ++++++++++++++++++++++++ internal/api/budget_test.go | 68 +++++++++++++++++++++++++++++++++++++++++++++ internal/api/server.go | 2 ++ internal/cli/serve.go | 3 ++ web/app.js | 48 ++++++++++++++++++++++++++++++++ web/index.html | 1 + web/style.css | 21 ++++++++++++++ web/test/budget.test.mjs | 58 ++++++++++++++++++++++++++++++++++++++ 8 files changed, 237 insertions(+) create mode 100644 internal/api/budget.go create mode 100644 internal/api/budget_test.go create mode 100644 web/test/budget.test.mjs (limited to 'internal/api') diff --git a/internal/api/budget.go b/internal/api/budget.go new file mode 100644 index 0000000..eee5d09 --- /dev/null +++ b/internal/api/budget.go @@ -0,0 +1,36 @@ +package api + +import ( + "net/http" + + "github.com/thepeterstone/claudomator/internal/budget" +) + +// budgetReporter exposes per-provider spend headroom. Satisfied by +// *budget.Accountant; an interface keeps the handler testable. +type budgetReporter interface { + All() ([]budget.Headroom, error) +} + +// SetBudget wires the budget accountant so GET /api/budget can report headroom. +func (s *Server) SetBudget(b budgetReporter) { + s.budget = b +} + +// handleGetBudget returns per-provider rolling-window spend headroom. When no +// budget is configured it returns an empty list so the UI simply shows nothing. +func (s *Server) handleGetBudget(w http.ResponseWriter, _ *http.Request) { + if s.budget == nil { + writeJSON(w, http.StatusOK, []budget.Headroom{}) + return + } + hs, err := s.budget.All() + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) + return + } + if hs == nil { + hs = []budget.Headroom{} + } + writeJSON(w, http.StatusOK, hs) +} diff --git a/internal/api/budget_test.go b/internal/api/budget_test.go new file mode 100644 index 0000000..d59836d --- /dev/null +++ b/internal/api/budget_test.go @@ -0,0 +1,68 @@ +package api + +import ( + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "testing" + + "github.com/thepeterstone/claudomator/internal/budget" +) + +type fakeBudget struct { + hs []budget.Headroom + err error +} + +func (f fakeBudget) All() ([]budget.Headroom, error) { return f.hs, f.err } + +func TestHandleGetBudget_NoBudgetConfigured_ReturnsEmpty(t *testing.T) { + srv, _ := testServer(t) + req := httptest.NewRequest("GET", "/api/budget", nil) + w := httptest.NewRecorder() + srv.Handler().ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status: want 200, got %d", w.Code) + } + var hs []budget.Headroom + if err := json.Unmarshal(w.Body.Bytes(), &hs); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if len(hs) != 0 { + t.Errorf("want empty list, got %+v", hs) + } +} + +func TestHandleGetBudget_ReportsHeadroom(t *testing.T) { + srv, _ := testServer(t) + srv.SetBudget(fakeBudget{hs: []budget.Headroom{ + {Provider: "claude", Limited: true, Limit: 10, Spent: 4, Remaining: 6, Fraction: 0.6}, + }}) + req := httptest.NewRequest("GET", "/api/budget", nil) + w := httptest.NewRecorder() + srv.Handler().ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status: want 200, got %d", w.Code) + } + var hs []budget.Headroom + if err := json.Unmarshal(w.Body.Bytes(), &hs); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if len(hs) != 1 || hs[0].Provider != "claude" || hs[0].Remaining != 6 { + t.Errorf("unexpected headroom: %+v", hs) + } +} + +func TestHandleGetBudget_ErrorReturns500(t *testing.T) { + srv, _ := testServer(t) + srv.SetBudget(fakeBudget{err: errors.New("db down")}) + req := httptest.NewRequest("GET", "/api/budget", nil) + w := httptest.NewRecorder() + srv.Handler().ServeHTTP(w, req) + if w.Code != http.StatusInternalServerError { + t.Errorf("want 500, got %d", w.Code) + } +} diff --git a/internal/api/server.go b/internal/api/server.go index a4b7ea1..ffa8cd4 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -61,6 +61,7 @@ type Server struct { pushStore pushSubscriptionStore dropsDir string llm *llm.Client + budget budgetReporter // optional; per-provider spend headroom for GET /api/budget } // SetAPIToken configures a bearer token that must be supplied to access the API. @@ -147,6 +148,7 @@ func (s *Server) routes() { s.mux.HandleFunc("GET /api/tasks/{id}/events", s.handleListTaskEvents) s.mux.HandleFunc("GET /api/executions", s.handleListRecentExecutions) s.mux.HandleFunc("GET /api/stats", s.handleGetDashboardStats) + s.mux.HandleFunc("GET /api/budget", s.handleGetBudget) s.mux.HandleFunc("GET /api/agents/status", s.handleGetAgentStatus) s.mux.HandleFunc("GET /api/executions/{id}", s.handleGetExecution) s.mux.HandleFunc("GET /api/executions/{id}/log", s.handleGetExecutionLog) diff --git a/internal/cli/serve.go b/internal/cli/serve.go index 55fdaf5..7afa678 100644 --- a/internal/cli/serve.go +++ b/internal/cli/serve.go @@ -171,6 +171,9 @@ func serve(addr string) error { pool.RecoverStaleBlocked() srv := api.NewServer(store, pool, agentRegistry, logger, cfg.ClaudeBinaryPath, cfg.GeminiBinaryPath) + if accountant != nil { + srv.SetBudget(accountant) + } // Configure notifiers: combine webhook (if set) with web push. notifiers := []notify.Notifier{} diff --git a/web/app.js b/web/app.js index 8a2662f..7f96e09 100644 --- a/web/app.js +++ b/web/app.js @@ -198,6 +198,53 @@ export async function fetchTaskEvents(taskId, fetchImpl = (typeof fetch !== 'und } } +// ── Budget headroom ───────────────────────────────────────────────────────── +// Renders the per-provider rolling-window spend headroom from GET /api/budget. + +// formatBudgetHeadroom turns one provider's headroom into a short label. +// Returns '' for unlimited providers (nothing to show). +export function formatBudgetHeadroom(h) { + if (!h || !h.limited) return ''; + const pct = Math.round((h.fraction_remaining || 0) * 100); + const remaining = (h.remaining_usd || 0).toFixed(2); + const limit = (h.limit_usd || 0).toFixed(2); + const name = h.provider ? h.provider.charAt(0).toUpperCase() + h.provider.slice(1) : '?'; + return `${name}: ${pct}% left ($${remaining} of $${limit})`; +} + +// renderBudgetHeadroom builds a chip per limited provider, flagging +// providers under 20% remaining with a --low modifier. Returns the container. +export function renderBudgetHeadroom(headrooms, doc = (typeof document !== 'undefined' ? document : null)) { + if (doc == null) return null; + const wrap = doc.createElement('div'); + wrap.className = 'budget-bar'; + for (const h of headrooms || []) { + if (!h || !h.limited) continue; + const chip = doc.createElement('span'); + chip.className = 'budget-chip' + ((h.fraction_remaining || 0) < 0.2 ? ' budget-chip--low' : ''); + chip.textContent = formatBudgetHeadroom(h); + wrap.appendChild(chip); + } + return wrap; +} + +// loadBudget fetches headroom and injects chips into #budget-bar. Best-effort. +async function loadBudget(fetchImpl = (typeof fetch !== 'undefined' ? fetch : null)) { + if (!fetchImpl || typeof document === 'undefined') return; + const slot = document.getElementById('budget-bar'); + if (!slot) return; + try { + const resp = await fetchImpl(`${BASE_PATH}/api/budget`); + if (!resp.ok) return; + const headrooms = await resp.json(); + const rendered = renderBudgetHeadroom(headrooms); + slot.innerHTML = ''; + if (rendered) for (const c of rendered.children) slot.appendChild(c); + } catch { + // budget display is non-critical; ignore. + } +} + function truncateToWordBoundary(text, maxLen = 120) { if (!text || text.length <= maxLen) return text; const cut = text.lastIndexOf(' ', maxLen); @@ -1297,6 +1344,7 @@ function renderActiveTab(allTasks) { async function poll() { try { + loadBudget(); // fire-and-forget; budget can change independently of tasks const health = await fetchHealth(); const serverUpdate = health.last_updated; diff --git a/web/index.html b/web/index.html index 8a705cc..5aa7b44 100644 --- a/web/index.html +++ b/web/index.html @@ -37,6 +37,7 @@ +