summaryrefslogtreecommitdiff
path: root/internal/executor/channel.go
diff options
context:
space:
mode:
authorClaude <noreply@anthropic.com>2026-05-24 09:38:06 +0000
committerClaude <noreply@anthropic.com>2026-05-24 09:38:06 +0000
commit54f6631c28a8b85f6f874e17822549faba916a38 (patch)
treebfd2a45e77cdbe60ca95362caf90721da6c8ea22 /internal/executor/channel.go
parentc7d95f3992d24f86ff71e5f3e18260a8ef8a09f0 (diff)
feat(executor): per-task agent MCP server + token registry (Phase 2)
Adds the agent-facing MCP transport foundation: a Registry that mints a per-task bearer token bound to a freshly built MCP server exposing the four agent tools (ask_user, report_summary, spawn_subtask, record_progress), and an HTTP handler (StreamableHTTP) that resolves the token to that server. The server never trusts an agent-supplied task ID — context comes from the token. The default storeChannel now buffers summary and question signals under a mutex (an MCP tool call lands on an HTTP-handler goroutine mid-run), exposing ReportedSummary/PendingQuestion. The pool flushes the buffered summary onto the execution after the run, replacing the runner's direct exec.Summary write and keeping the read race-free. ask_user follows the record-and-resume model: it buffers the question, returns ErrAgentBlocked, and the tool tells the agent to end its turn; the run blocks and resumes later via claude --resume (no live slot held). Tests cover registry lifecycle, in-memory tool dispatch, and HTTP end-to-end with bearer auth (valid token dispatches; invalid token rejected). Not yet wired into the runners or mounted on the API server — next increment. https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39
Diffstat (limited to 'internal/executor/channel.go')
-rw-r--r--internal/executor/channel.go59
1 files changed, 43 insertions, 16 deletions
diff --git a/internal/executor/channel.go b/internal/executor/channel.go
index 76df94f..541694b 100644
--- a/internal/executor/channel.go
+++ b/internal/executor/channel.go
@@ -4,10 +4,10 @@ import (
"context"
"encoding/json"
"errors"
+ "sync"
"time"
"github.com/thepeterstone/claudomator/internal/event"
- "github.com/thepeterstone/claudomator/internal/storage"
"github.com/thepeterstone/claudomator/internal/task"
"github.com/google/uuid"
)
@@ -50,37 +50,64 @@ type channelStore interface {
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.
+// 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
+// the runner reads the pending question to build a BlockedError after the
+// subprocess exits. SpawnSubtask/RecordProgress write immediately.
type storeChannel struct {
store channelStore
taskID string
- exec *storage.Execution
+
+ mu sync.Mutex
+ summary string
+ summarySet bool
+ question string
+ blocked bool
}
var _ AgentChannel = (*storeChannel)(nil)
-func newStoreChannel(store channelStore, taskID string, exec *storage.Execution) *storeChannel {
- return &storeChannel{store: store, taskID: taskID, exec: exec}
+func newStoreChannel(store channelStore, taskID string) *storeChannel {
+ return &storeChannel{store: store, taskID: taskID}
}
-// 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) {
+// AskUser buffers the question and flags the run as blocked. The default
+// transport cannot deliver an answer in-session, so it returns ErrAgentBlocked;
+// the runner converts the buffered question into a BlockedError after exit, and
+// question persistence is owned by the pool's BlockedError handling.
+func (c *storeChannel) AskUser(_ context.Context, questionJSON string) (string, error) {
+ c.mu.Lock()
+ c.question = questionJSON
+ c.blocked = true
+ c.mu.Unlock()
return "", ErrAgentBlocked
}
-// ReportSummary buffers the summary onto the execution record. The pool persists
-// it (preferring this value over its extract/synthesize fallbacks).
+// ReportSummary buffers the summary; the pool flushes it onto the execution
+// after the run (preferring it over its extract/synthesize fallbacks).
func (c *storeChannel) ReportSummary(_ context.Context, summary string) error {
- if c.exec != nil {
- c.exec.Summary = summary
- }
+ c.mu.Lock()
+ c.summary = summary
+ c.summarySet = true
+ c.mu.Unlock()
return nil
}
+// ReportedSummary returns the buffered summary if report_summary was called.
+func (c *storeChannel) ReportedSummary() (string, bool) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ return c.summary, c.summarySet
+}
+
+// PendingQuestion returns the buffered ask_user question if the run is blocked.
+func (c *storeChannel) PendingQuestion() (string, bool) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ return c.question, c.blocked
+}
+
func (c *storeChannel) SpawnSubtask(_ context.Context, spec SubtaskSpec) (string, error) {
now := time.Now().UTC()
child := &task.Task{