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/executor/executor.go | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) (limited to 'internal/executor/executor.go') 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) -- cgit v1.2.3 From 54f6631c28a8b85f6f874e17822549faba916a38 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 24 May 2026 09:38:06 +0000 Subject: feat(executor): per-task agent MCP server + token registry (Phase 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- go.mod | 14 ++- go.sum | 47 +++++++- internal/executor/agentmcp.go | 160 ++++++++++++++++++++++++++ internal/executor/agentmcp_test.go | 218 ++++++++++++++++++++++++++++++++++++ internal/executor/channel.go | 59 +++++++--- internal/executor/channel_test.go | 42 ++++--- internal/executor/claude_test.go | 8 +- internal/executor/container_test.go | 12 +- internal/executor/executor.go | 12 +- internal/executor/gemini_test.go | 14 +-- internal/executor/local_test.go | 6 +- 11 files changed, 531 insertions(+), 61 deletions(-) create mode 100644 internal/executor/agentmcp.go create mode 100644 internal/executor/agentmcp_test.go (limited to 'internal/executor/executor.go') diff --git a/go.mod b/go.mod index 54d5b32..fb2828c 100644 --- a/go.mod +++ b/go.mod @@ -3,27 +3,33 @@ module github.com/thepeterstone/claudomator go 1.25.3 require ( + github.com/BurntSushi/toml v1.6.0 + github.com/SherClockHolmes/webpush-go v1.4.0 github.com/google/uuid v1.6.0 github.com/mattn/go-sqlite3 v1.14.33 + github.com/modelcontextprotocol/go-sdk v1.6.1 github.com/spf13/cobra v1.10.2 golang.org/x/net v0.49.0 gopkg.in/yaml.v3 v3.0.1 + modernc.org/sqlite v1.47.0 ) require ( - github.com/BurntSushi/toml v1.6.0 // indirect - github.com/SherClockHolmes/webpush-go v1.4.0 // indirect github.com/dustin/go-humanize v1.0.1 // indirect - github.com/golang-jwt/jwt/v5 v5.2.1 // indirect + github.com/golang-jwt/jwt/v5 v5.3.1 // indirect + github.com/google/jsonschema-go v0.4.3 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/ncruces/go-strftime v1.0.0 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect + github.com/segmentio/asm v1.1.3 // indirect + github.com/segmentio/encoding v0.5.4 // indirect github.com/spf13/pflag v1.0.9 // indirect + github.com/yosida95/uritemplate/v3 v3.0.2 // indirect golang.org/x/crypto v0.47.0 // indirect + golang.org/x/oauth2 v0.35.0 // indirect golang.org/x/sys v0.42.0 // indirect modernc.org/libc v1.70.0 // indirect modernc.org/mathutil v1.7.1 // indirect modernc.org/memory v1.11.0 // indirect - modernc.org/sqlite v1.47.0 // indirect ) diff --git a/go.sum b/go.sum index a202d97..e4d2f97 100644 --- a/go.sum +++ b/go.sum @@ -5,26 +5,43 @@ github.com/SherClockHolmes/webpush-go v1.4.0/go.mod h1:XSq8pKX11vNV8MJEMwjrlTkxh github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= -github.com/golang-jwt/jwt/v5 v5.2.1 h1:OuVbFODueb089Lh128TAcimifWaLhJwVflnrgM17wHk= github.com/golang-jwt/jwt/v5 v5.2.1/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVIyoH402zdk= +github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY= +github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE= github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/jsonschema-go v0.4.3 h1:/DBOLZTfDow7pe2GmaJNhltueGTtDKICi8V8p+DQPd0= +github.com/google/jsonschema-go v0.4.3/go.mod h1:r5quNTdLOYEz95Ru18zA0ydNbBuYoo9tgaYcxEYhJVE= +github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs= +github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= +github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/mattn/go-sqlite3 v1.14.33 h1:A5blZ5ulQo2AtayQ9/limgHEkFreKj1Dv226a1K73s0= github.com/mattn/go-sqlite3 v1.14.33/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y= +github.com/modelcontextprotocol/go-sdk v1.6.1 h1:0zOSupjKUxPKSocPT1Wtago+mUHU2/uZ4xSOY0FGReU= +github.com/modelcontextprotocol/go-sdk v1.6.1/go.mod h1:kzm3kzFL1/+AziGOE0nUs3gvPoNxMCvkxokMkuFapXQ= github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= +github.com/segmentio/asm v1.1.3 h1:WM03sfUOENvvKexOLp+pCqgb/WDjsi7EK8gIsICtzhc= +github.com/segmentio/asm v1.1.3/go.mod h1:Ld3L4ZXGNcSLRg4JBsZ3//1+f/TjYl0Mzen/DQy1EJg= +github.com/segmentio/encoding v0.5.4 h1:OW1VRern8Nw6ITAtwSZ7Idrl3MXCFwXHPgqESYfvNt0= +github.com/segmentio/encoding v0.5.4/go.mod h1:HS1ZKa3kSN32ZHVZ7ZLPLXWvOVIiZtyJnO1gPH1sKt0= github.com/spf13/cobra v1.10.2 h1:DMTTonx5m65Ic0GOoRY2c16WCbHxOOw6xxezuLaBpcU= github.com/spf13/cobra v1.10.2/go.mod h1:7C1pvHqHw5A4vrJfjNwvOdzYu0Gml16OCs2GRiTUUS4= github.com/spf13/pflag v1.0.9 h1:9exaQaMOCwffKiiiYk6/BndUBv+iRViNW+4lEMi0PvY= github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4= +github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4= github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= @@ -40,6 +57,8 @@ golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= golang.org/x/mod v0.12.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= golang.org/x/mod v0.15.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c= golang.org/x/mod v0.17.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c= +golang.org/x/mod v0.33.0 h1:tHFzIWbBifEmbwtGz65eaWyGiGZatSrT9prnU8DbVL8= +golang.org/x/mod v0.33.0/go.mod h1:swjeQEj+6r7fODbD2cqrnje9PnziFuw4bmLbBZFrQ5w= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= @@ -50,6 +69,8 @@ golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44= golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM= golang.org/x/net v0.49.0 h1:eeHFmOGUTtaaPSGNmjBKpbng9MulQsJURQUAfUwY++o= golang.org/x/net v0.49.0/go.mod h1:/ysNB2EvaqvesRkuLAyjI1ycPZlQHM3q01F02UY/MV8= +golang.org/x/oauth2 v0.35.0 h1:Mv2mzuHuZuY2+bkyWXIHMfhNdJAdwW3FuWeCPYN5GVQ= +golang.org/x/oauth2 v0.35.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= @@ -57,6 +78,8 @@ golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y= golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= +golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= +golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= @@ -95,16 +118,38 @@ golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU= golang.org/x/tools v0.13.0/go.mod h1:HvlwmtVNQAhOuCjW7xxvovg8wbNq7LwfXh/k7wXUl58= golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk= +golang.org/x/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k= +golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +modernc.org/cc/v4 v4.27.1 h1:9W30zRlYrefrDV2JE2O8VDtJ1yPGownxciz5rrbQZis= +modernc.org/cc/v4 v4.27.1/go.mod h1:uVtb5OGqUKpoLWhqwNQo/8LwvoiEBLvZXIQ/SmO6mL0= +modernc.org/ccgo/v4 v4.32.0 h1:hjG66bI/kqIPX1b2yT6fr/jt+QedtP2fqojG2VrFuVw= +modernc.org/ccgo/v4 v4.32.0/go.mod h1:6F08EBCx5uQc38kMGl+0Nm0oWczoo1c7cgpzEry7Uc0= +modernc.org/fileutil v1.4.0 h1:j6ZzNTftVS054gi281TyLjHPp6CPHr2KCxEXjEbD6SM= +modernc.org/fileutil v1.4.0/go.mod h1:EqdKFDxiByqxLk8ozOxObDSfcVOv/54xDs/DUHdvCUU= +modernc.org/gc/v2 v2.6.5 h1:nyqdV8q46KvTpZlsw66kWqwXRHdjIlJOhG6kxiV/9xI= +modernc.org/gc/v2 v2.6.5/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito= +modernc.org/gc/v3 v3.1.2 h1:ZtDCnhonXSZexk/AYsegNRV1lJGgaNZJuKjJSWKyEqo= +modernc.org/gc/v3 v3.1.2/go.mod h1:HFK/6AGESC7Ex+EZJhJ2Gni6cTaYpSMmU/cT9RmlfYY= +modernc.org/goabi0 v0.2.0 h1:HvEowk7LxcPd0eq6mVOAEMai46V+i7Jrj13t4AzuNks= +modernc.org/goabi0 v0.2.0/go.mod h1:CEFRnnJhKvWT1c1JTI3Avm+tgOWbkOu5oPA8eH8LnMI= modernc.org/libc v1.70.0 h1:U58NawXqXbgpZ/dcdS9kMshu08aiA6b7gusEusqzNkw= modernc.org/libc v1.70.0/go.mod h1:OVmxFGP1CI/Z4L3E0Q3Mf1PDE0BucwMkcXjjLntvHJo= modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU= modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg= modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI= modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw= +modernc.org/opt v0.1.4 h1:2kNGMRiUjrp4LcaPuLY2PzUfqM/w9N23quVwhKt5Qm8= +modernc.org/opt v0.1.4/go.mod h1:03fq9lsNfvkYSfxrfUhZCWPk1lm4cq4N+Bh//bEtgns= +modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w= +modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE= modernc.org/sqlite v1.47.0 h1:R1XyaNpoW4Et9yly+I2EeX7pBza/w+pmYee/0HJDyKk= modernc.org/sqlite v1.47.0/go.mod h1:hWjRO6Tj/5Ik8ieqxQybiEOUXy0NJFNp2tpvVpKlvig= +modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0= +modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A= +modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y= +modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM= diff --git a/internal/executor/agentmcp.go b/internal/executor/agentmcp.go new file mode 100644 index 0000000..4368031 --- /dev/null +++ b/internal/executor/agentmcp.go @@ -0,0 +1,160 @@ +package executor + +import ( + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "net/http" + "strings" + "sync" + + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +// Registry maps per-task MCP bearer tokens to a built agent MCP server. A token +// is minted when a runner spawns an agent subprocess and revoked when it exits, +// so the server resolves task context from the token alone and never trusts an +// agent-supplied task ID. +type Registry struct { + mu sync.RWMutex + servers map[string]*mcp.Server +} + +func NewRegistry() *Registry { + return &Registry{servers: make(map[string]*mcp.Server)} +} + +// Mint creates a token bound to a freshly built MCP server for ch. +func (r *Registry) Mint(ch AgentChannel) (string, error) { + buf := make([]byte, 32) + if _, err := rand.Read(buf); err != nil { + return "", err + } + token := hex.EncodeToString(buf) + r.mu.Lock() + r.servers[token] = newAgentServer(ch) + r.mu.Unlock() + return token, nil +} + +func (r *Registry) server(token string) (*mcp.Server, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + s, ok := r.servers[token] + return s, ok +} + +func (r *Registry) Revoke(token string) { + r.mu.Lock() + delete(r.servers, token) + r.mu.Unlock() +} + +type askUserInput struct { + Question string `json:"question" jsonschema:"the question to ask the user; phrase it as a real question ending with a question mark"` + Options []string `json:"options,omitempty" jsonschema:"optional list of suggested answer choices"` +} + +type reportSummaryInput struct { + Summary string `json:"summary" jsonschema:"a 2-5 sentence summary of what you did and the outcome"` +} + +type spawnSubtaskInput struct { + Name string `json:"name" jsonschema:"short descriptive name for the subtask"` + Instructions string `json:"instructions" jsonschema:"complete instructions for the subtask agent"` + Model string `json:"model,omitempty" jsonschema:"optional model override, e.g. sonnet or opus"` + MaxBudgetUSD float64 `json:"max_budget_usd,omitempty" jsonschema:"optional budget cap in USD"` +} + +type recordProgressInput struct { + Message string `json:"message" jsonschema:"a short progress note describing what you are doing"` +} + +func textResult(text string) *mcp.CallToolResult { + return &mcp.CallToolResult{Content: []mcp.Content{&mcp.TextContent{Text: text}}} +} + +// newAgentServer builds an MCP server exposing the four agent tools bound to ch. +func newAgentServer(ch AgentChannel) *mcp.Server { + s := mcp.NewServer(&mcp.Implementation{Name: "claudomator", Version: "1"}, nil) + + mcp.AddTool(s, &mcp.Tool{ + Name: "ask_user", + Description: "Ask the user a question when you genuinely need a decision to proceed. Your turn ends after calling this; the task is resumed with the user's answer. Prefer making a reasonable decision and noting it in report_summary over asking.", + }, func(ctx context.Context, _ *mcp.CallToolRequest, in askUserInput) (*mcp.CallToolResult, any, error) { + q := map[string]any{"text": in.Question} + if len(in.Options) > 0 { + q["options"] = in.Options + } + payload, _ := json.Marshal(q) + ans, err := ch.AskUser(ctx, string(payload)) + if errors.Is(err, ErrAgentBlocked) { + return textResult("Question recorded. End your turn now without calling any more tools; the task will be resumed once the user answers."), nil, nil + } + if err != nil { + return nil, nil, err + } + return textResult(ans), nil, nil + }) + + mcp.AddTool(s, &mcp.Tool{ + Name: "report_summary", + Description: "Record a concise summary of what you accomplished. Call this before finishing.", + }, func(ctx context.Context, _ *mcp.CallToolRequest, in reportSummaryInput) (*mcp.CallToolResult, any, error) { + if err := ch.ReportSummary(ctx, in.Summary); err != nil { + return nil, nil, err + } + return textResult("Summary recorded."), nil, nil + }) + + mcp.AddTool(s, &mcp.Tool{ + Name: "spawn_subtask", + Description: "Create a child task to be executed separately. Use this to break large work into focused pieces, then finish your turn.", + }, func(ctx context.Context, _ *mcp.CallToolRequest, in spawnSubtaskInput) (*mcp.CallToolResult, any, error) { + id, err := ch.SpawnSubtask(ctx, SubtaskSpec{ + Name: in.Name, + Instructions: in.Instructions, + Model: in.Model, + MaxBudgetUSD: in.MaxBudgetUSD, + }) + if err != nil { + return nil, nil, err + } + return textResult("Created subtask " + id), nil, nil + }) + + mcp.AddTool(s, &mcp.Tool{ + Name: "record_progress", + Description: "Record a short progress note that appears in the task timeline.", + }, func(ctx context.Context, _ *mcp.CallToolRequest, in recordProgressInput) (*mcp.CallToolResult, any, error) { + if err := ch.RecordProgress(ctx, in.Message); err != nil { + return nil, nil, err + } + return textResult("Noted."), nil, nil + }) + + return s +} + +func bearerToken(r *http.Request) string { + h := r.Header.Get("Authorization") + if h == "" { + return "" + } + return strings.TrimSpace(strings.TrimPrefix(h, "Bearer ")) +} + +// NewAgentMCPHandler returns the HTTP handler for the per-task agent MCP server. +// It resolves the request's bearer token to the server for that task's run; +// unknown tokens yield a 400 from the underlying handler. +func NewAgentMCPHandler(reg *Registry) http.Handler { + return mcp.NewStreamableHTTPHandler(func(r *http.Request) *mcp.Server { + s, ok := reg.server(bearerToken(r)) + if !ok { + return nil + } + return s + }, nil) +} diff --git a/internal/executor/agentmcp_test.go b/internal/executor/agentmcp_test.go new file mode 100644 index 0000000..3973af0 --- /dev/null +++ b/internal/executor/agentmcp_test.go @@ -0,0 +1,218 @@ +package executor + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +// recordingChannel is a fake AgentChannel that records tool invocations. +type recordingChannel struct { + asked string + summary string + spawned []SubtaskSpec + progress []string + spawnID string +} + +func (c *recordingChannel) AskUser(_ context.Context, q string) (string, error) { + c.asked = q + return "", ErrAgentBlocked +} +func (c *recordingChannel) ReportSummary(_ context.Context, s string) error { + c.summary = s + return nil +} +func (c *recordingChannel) SpawnSubtask(_ context.Context, spec SubtaskSpec) (string, error) { + c.spawned = append(c.spawned, spec) + return c.spawnID, nil +} +func (c *recordingChannel) RecordProgress(_ context.Context, m string) error { + c.progress = append(c.progress, m) + return nil +} + +func resultText(t *testing.T, res *mcp.CallToolResult) string { + t.Helper() + if len(res.Content) == 0 { + t.Fatal("expected content in tool result") + } + tc, ok := res.Content[0].(*mcp.TextContent) + if !ok { + t.Fatalf("expected TextContent, got %T", res.Content[0]) + } + return tc.Text +} + +func TestRegistry_MintLookupRevoke(t *testing.T) { + reg := NewRegistry() + tok, err := reg.Mint(&recordingChannel{}) + if err != nil { + t.Fatalf("Mint: %v", err) + } + if tok == "" { + t.Fatal("expected non-empty token") + } + if _, ok := reg.server(tok); !ok { + t.Error("expected server for minted token") + } + if _, ok := reg.server("nonsense"); ok { + t.Error("expected no server for unknown token") + } + reg.Revoke(tok) + if _, ok := reg.server(tok); ok { + t.Error("expected no server after revoke") + } +} + +func TestRegistry_MintUniqueTokens(t *testing.T) { + reg := NewRegistry() + a, _ := reg.Mint(&recordingChannel{}) + b, _ := reg.Mint(&recordingChannel{}) + if a == b { + t.Error("expected unique tokens per mint") + } +} + +// connectInMemory wires a client to a server over the SDK's in-memory transport. +func connectInMemory(t *testing.T, srv *mcp.Server) *mcp.ClientSession { + t.Helper() + ctx := context.Background() + clientT, serverT := mcp.NewInMemoryTransports() + if _, err := srv.Connect(ctx, serverT, nil); err != nil { + t.Fatalf("server connect: %v", err) + } + client := mcp.NewClient(&mcp.Implementation{Name: "test", Version: "1"}, nil) + cs, err := client.Connect(ctx, clientT, nil) + if err != nil { + t.Fatalf("client connect: %v", err) + } + t.Cleanup(func() { _ = cs.Close() }) + return cs +} + +func TestAgentServer_AskUser_RecordsAndInstructsStop(t *testing.T) { + ch := &recordingChannel{} + cs := connectInMemory(t, newAgentServer(ch)) + res, err := cs.CallTool(context.Background(), &mcp.CallToolParams{ + Name: "ask_user", + Arguments: map[string]any{"question": "Proceed?", "options": []string{"yes", "no"}}, + }) + if err != nil { + t.Fatalf("CallTool: %v", err) + } + if !strings.Contains(ch.asked, `"text":"Proceed?"`) || !strings.Contains(ch.asked, "yes") { + t.Errorf("channel did not receive question+options: %q", ch.asked) + } + if txt := resultText(t, res); !strings.Contains(strings.ToLower(txt), "recorded") { + t.Errorf("expected stop instruction, got %q", txt) + } +} + +func TestAgentServer_ReportSummary(t *testing.T) { + ch := &recordingChannel{} + cs := connectInMemory(t, newAgentServer(ch)) + if _, err := cs.CallTool(context.Background(), &mcp.CallToolParams{ + Name: "report_summary", + Arguments: map[string]any{"summary": "did the work"}, + }); err != nil { + t.Fatalf("CallTool: %v", err) + } + if ch.summary != "did the work" { + t.Errorf("summary not recorded, got %q", ch.summary) + } +} + +func TestAgentServer_SpawnSubtask(t *testing.T) { + ch := &recordingChannel{spawnID: "sub-1"} + cs := connectInMemory(t, newAgentServer(ch)) + res, err := cs.CallTool(context.Background(), &mcp.CallToolParams{ + Name: "spawn_subtask", + Arguments: map[string]any{"name": "child", "instructions": "do it", "model": "sonnet"}, + }) + if err != nil { + t.Fatalf("CallTool: %v", err) + } + if len(ch.spawned) != 1 || ch.spawned[0].Name != "child" || ch.spawned[0].Instructions != "do it" || ch.spawned[0].Model != "sonnet" { + t.Errorf("subtask spec not propagated: %+v", ch.spawned) + } + if txt := resultText(t, res); !strings.Contains(txt, "sub-1") { + t.Errorf("expected returned subtask ID in result, got %q", txt) + } +} + +func TestAgentServer_RecordProgress(t *testing.T) { + ch := &recordingChannel{} + cs := connectInMemory(t, newAgentServer(ch)) + if _, err := cs.CallTool(context.Background(), &mcp.CallToolParams{ + Name: "record_progress", + Arguments: map[string]any{"message": "halfway"}, + }); err != nil { + t.Fatalf("CallTool: %v", err) + } + if len(ch.progress) != 1 || ch.progress[0] != "halfway" { + t.Errorf("progress not recorded: %+v", ch.progress) + } +} + +type bearerRT struct { + token string + base http.RoundTripper +} + +func (b bearerRT) RoundTrip(r *http.Request) (*http.Response, error) { + r = r.Clone(r.Context()) + if b.token != "" { + r.Header.Set("Authorization", "Bearer "+b.token) + } + return b.base.RoundTrip(r) +} + +func TestAgentMCPHandler_HTTP_AuthAndDispatch(t *testing.T) { + ch := &recordingChannel{} + reg := NewRegistry() + tok, _ := reg.Mint(ch) + + httpSrv := httptest.NewServer(NewAgentMCPHandler(reg)) + defer httpSrv.Close() + + ctx := context.Background() + client := mcp.NewClient(&mcp.Implementation{Name: "test", Version: "1"}, nil) + cs, err := client.Connect(ctx, &mcp.StreamableClientTransport{ + Endpoint: httpSrv.URL, + HTTPClient: &http.Client{Transport: bearerRT{token: tok, base: http.DefaultTransport}}, + }, nil) + if err != nil { + t.Fatalf("client connect with valid token: %v", err) + } + defer cs.Close() + + if _, err := cs.CallTool(ctx, &mcp.CallToolParams{ + Name: "record_progress", + Arguments: map[string]any{"message": "over http"}, + }); err != nil { + t.Fatalf("CallTool over HTTP: %v", err) + } + if len(ch.progress) != 1 || ch.progress[0] != "over http" { + t.Errorf("HTTP dispatch did not reach channel: %+v", ch.progress) + } +} + +func TestAgentMCPHandler_HTTP_BadTokenRejected(t *testing.T) { + reg := NewRegistry() + httpSrv := httptest.NewServer(NewAgentMCPHandler(reg)) + defer httpSrv.Close() + + client := mcp.NewClient(&mcp.Implementation{Name: "test", Version: "1"}, nil) + _, err := client.Connect(context.Background(), &mcp.StreamableClientTransport{ + Endpoint: httpSrv.URL, + HTTPClient: &http.Client{Transport: bearerRT{token: "invalid", base: http.DefaultTransport}}, + }, nil) + if err == nil { + t.Fatal("expected connect to fail with invalid token") + } +} 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{ diff --git a/internal/executor/channel_test.go b/internal/executor/channel_test.go index 822dd35..250e8eb 100644 --- a/internal/executor/channel_test.go +++ b/internal/executor/channel_test.go @@ -6,7 +6,6 @@ import ( "testing" "github.com/thepeterstone/claudomator/internal/event" - "github.com/thepeterstone/claudomator/internal/storage" "github.com/thepeterstone/claudomator/internal/task" ) @@ -29,8 +28,8 @@ func (f *fakeChannelStore) CreateEvent(e *event.Event) error { return nil } -func TestStoreChannel_AskUser_ReturnsBlocked(t *testing.T) { - ch := newStoreChannel(&fakeChannelStore{}, "task-1", &storage.Execution{}) +func TestStoreChannel_AskUser_BuffersAndBlocks(t *testing.T) { + ch := newStoreChannel(&fakeChannelStore{}, "task-1") answer, err := ch.AskUser(context.Background(), `{"text":"q"}`) if !errors.Is(err, ErrAgentBlocked) { t.Errorf("expected ErrAgentBlocked, got %v", err) @@ -38,29 +37,36 @@ func TestStoreChannel_AskUser_ReturnsBlocked(t *testing.T) { if answer != "" { t.Errorf("expected empty answer, got %q", answer) } + q, blocked := ch.PendingQuestion() + if !blocked || q != `{"text":"q"}` { + t.Errorf("expected buffered question, got q=%q blocked=%v", q, blocked) + } } -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_PendingQuestion_DefaultNotBlocked(t *testing.T) { + ch := newStoreChannel(&fakeChannelStore{}, "task-1") + if q, blocked := ch.PendingQuestion(); blocked || q != "" { + t.Errorf("expected no pending question, got q=%q blocked=%v", q, blocked) } } -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_ReportSummary_Buffers(t *testing.T) { + ch := newStoreChannel(&fakeChannelStore{}, "task-1") + if _, ok := ch.ReportedSummary(); ok { + t.Error("expected no summary before report") + } + if err := ch.ReportSummary(context.Background(), "did the thing"); err != nil { + t.Fatalf("ReportSummary: %v", err) + } + sum, ok := ch.ReportedSummary() + if !ok || sum != "did the thing" { + t.Errorf("expected buffered summary, got %q ok=%v", sum, ok) } } func TestStoreChannel_SpawnSubtask_CreatesChildWithParent(t *testing.T) { store := &fakeChannelStore{} - ch := newStoreChannel(store, "parent-1", &storage.Execution{}) + ch := newStoreChannel(store, "parent-1") id, err := ch.SpawnSubtask(context.Background(), SubtaskSpec{ Name: "child", Instructions: "do it", @@ -93,7 +99,7 @@ func TestStoreChannel_SpawnSubtask_CreatesChildWithParent(t *testing.T) { func TestStoreChannel_SpawnSubtask_PropagatesError(t *testing.T) { store := &fakeChannelStore{createTaskErr: errors.New("boom")} - ch := newStoreChannel(store, "parent-1", &storage.Execution{}) + ch := newStoreChannel(store, "parent-1") if _, err := ch.SpawnSubtask(context.Background(), SubtaskSpec{Name: "x"}); err == nil { t.Error("expected error from CreateTask") } @@ -101,7 +107,7 @@ func TestStoreChannel_SpawnSubtask_PropagatesError(t *testing.T) { func TestStoreChannel_RecordProgress_EmitsAgentMessage(t *testing.T) { store := &fakeChannelStore{} - ch := newStoreChannel(store, "task-1", &storage.Execution{}) + ch := newStoreChannel(store, "task-1") if err := ch.RecordProgress(context.Background(), "halfway done"); err != nil { t.Fatalf("RecordProgress: %v", err) } diff --git a/internal/executor/claude_test.go b/internal/executor/claude_test.go index 0e76260..414f6cf 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, newStoreChannel(nil, tk.ID, exec)) + _ = r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID)) // 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, newStoreChannel(nil, tk.ID, exec)) + err := r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID)) 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, newStoreChannel(nil, tk.ID, exec)) + _ = r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID)) 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, newStoreChannel(nil, tk.ID, e)); err != nil { + if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)); err != nil { t.Fatalf("Run with stale sandbox: %v", err) } diff --git a/internal/executor/container_test.go b/internal/executor/container_test.go index 5ee3a3c..9cd80dc 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, newStoreChannel(nil, tk.ID, exec)) + err := runner.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID)) 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, newStoreChannel(nil, tk.ID, e)) + err := runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)) 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, newStoreChannel(nil, tk.ID, e)) + err := runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)) 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, newStoreChannel(nil, tk.ID, e)) + runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)) // 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, newStoreChannel(nil, tk.ID, e)) + runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)) 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, newStoreChannel(nil, tk.ID, e)) + runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)) os.RemoveAll(e.SandboxDir) for _, a := range cloneArgs { diff --git a/internal/executor/executor.go b/internal/executor/executor.go index 76e67b8..6d6c528 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -356,8 +356,12 @@ func (p *Pool) executeResume(ctx context.Context, t *task.Task, exec *storage.Ex } } - err = runner.Run(ctx, t, exec, newStoreChannel(p.store, t.ID, exec)) + sc := newStoreChannel(p.store, t.ID) + err = runner.Run(ctx, t, exec, sc) exec.EndTime = time.Now().UTC() + if sum, ok := sc.ReportedSummary(); ok { + exec.Summary = sum + } p.decActiveAgent(agentType, &cleaned) p.handleRunResult(ctx, t, exec, err, agentType) @@ -1077,8 +1081,12 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) { } // Run the task. - err = runner.Run(ctx, t, exec, newStoreChannel(p.store, t.ID, exec)) + sc := newStoreChannel(p.store, t.ID) + err = runner.Run(ctx, t, exec, sc) exec.EndTime = time.Now().UTC() + if sum, ok := sc.ReportedSummary(); ok { + exec.Summary = sum + } p.decActiveAgent(agentType, &cleaned) p.handleRunResult(ctx, t, exec, err, agentType) diff --git a/internal/executor/gemini_test.go b/internal/executor/gemini_test.go index c8f7422..a906f2c 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, newStoreChannel(nil, tk.ID, exec)) + err := r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID)) 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, newStoreChannel(nil, tk.ID, e)); err != nil { + if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)); 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, newStoreChannel(nil, tk.ID, e)) + err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)) 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, newStoreChannel(nil, tk.ID, e)) + err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)) 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, newStoreChannel(nil, tk.ID, e)); err != nil { + if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)); 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, newStoreChannel(nil, tk.ID, e)); err != nil { + if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)); 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, newStoreChannel(nil, tk.ID, e)); err != nil { + if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)); err != nil { t.Fatalf("Run without project_dir: %v", err) } if e.SandboxDir != "" { diff --git a/internal/executor/local_test.go b/internal/executor/local_test.go index 8aa8791..ffe87f9 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, newStoreChannel(nil, tt.ID, exec)); err != nil { + if err := r.Run(context.Background(), tt, exec, newStoreChannel(nil, tt.ID)); 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, newStoreChannel(nil, tt.ID, exec)) + err := r.Run(context.Background(), tt, exec, newStoreChannel(nil, tt.ID)) 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, newStoreChannel(nil, tt.ID, exec)) + err := r.Run(context.Background(), tt, exec, newStoreChannel(nil, tt.ID)) if err == nil || !strings.Contains(err.Error(), "empty instructions") { t.Errorf("expected empty-instructions error, got %v", err) } -- cgit v1.2.3 From 39f74dbbc7adca0e2058112408bb99694deefcaf Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 26 May 2026 20:27:36 +0000 Subject: feat(budget,storage,config): per-provider spend accounting substrate (Phase 6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the budget accountant foundation: - storage: a per-execution `agent` column (additive migration, populated by the dispatcher) and SpendByProviderSince(since), summing cost_usd per provider in a window. Accurate attribution survives a task running on different providers across retries. - internal/budget: Accountant over a SpendSource + Limits, exposing Headroom (remaining/fraction per provider), Allow (would estCost breach the cap), and All (one query, sorted). Unconfigured/local providers are unlimited. - config: a [budget] section (window + provider_5h_usd map). No default cap — gating is opt-in by configuring limits. Fully unit-tested; dispatcher gating and the API/UI surface follow. https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39 --- internal/budget/budget.go | 122 +++++++++++++++++++++++++++++++++++++++++ internal/budget/budget_test.go | 112 +++++++++++++++++++++++++++++++++++++ internal/config/config.go | 11 ++++ internal/executor/executor.go | 1 + internal/storage/db.go | 50 +++++++++++++---- internal/storage/db_test.go | 38 +++++++++++++ 6 files changed, 324 insertions(+), 10 deletions(-) create mode 100644 internal/budget/budget.go create mode 100644 internal/budget/budget_test.go (limited to 'internal/executor/executor.go') diff --git a/internal/budget/budget.go b/internal/budget/budget.go new file mode 100644 index 0000000..60ce9d2 --- /dev/null +++ b/internal/budget/budget.go @@ -0,0 +1,122 @@ +// Package budget tracks agent spend against rolling per-provider windows so the +// dispatcher can refuse (or reroute) paid work that would breach a cap. Local +// runners are free and simply carry no configured limit. +package budget + +import ( + "sort" + "time" +) + +// DefaultWindow is the rolling spend window when none is configured. +const DefaultWindow = 5 * time.Hour + +// Limits configures per-provider spend caps within a rolling window. +type Limits struct { + // Window is the rolling lookback; DefaultWindow when zero. + Window time.Duration + // PerProvider maps a provider (claude/gemini/…) to its USD cap within the + // window. A missing or non-positive entry means "unlimited" (e.g. local). + PerProvider map[string]float64 +} + +// SpendSource supplies total spend per provider since a given time. +type SpendSource interface { + SpendByProviderSince(since time.Time) (map[string]float64, error) +} + +// Headroom describes remaining budget for one provider in the current window. +type Headroom struct { + Provider string `json:"provider"` + Limited bool `json:"limited"` + Limit float64 `json:"limit_usd"` + Spent float64 `json:"spent_usd"` + Remaining float64 `json:"remaining_usd"` + // Fraction is the share of the cap still available, 0..1 (0 when unlimited). + Fraction float64 `json:"fraction_remaining"` +} + +// Accountant answers spend questions against a SpendSource and Limits. +type Accountant struct { + src SpendSource + lim Limits + now func() time.Time +} + +// New builds an Accountant. A zero Window defaults to DefaultWindow. +func New(src SpendSource, lim Limits) *Accountant { + if lim.Window <= 0 { + lim.Window = DefaultWindow + } + return &Accountant{src: src, lim: lim, now: time.Now} +} + +func (a *Accountant) since() time.Time { return a.now().Add(-a.lim.Window) } + +func (a *Accountant) headroomFrom(provider string, spends map[string]float64) Headroom { + limit := a.lim.PerProvider[provider] + if limit <= 0 { + return Headroom{Provider: provider, Limited: false} + } + spent := spends[provider] + remaining := limit - spent + if remaining < 0 { + remaining = 0 + } + return Headroom{ + Provider: provider, + Limited: true, + Limit: limit, + Spent: spent, + Remaining: remaining, + Fraction: remaining / limit, + } +} + +// Headroom returns the remaining budget for one provider in the current window. +func (a *Accountant) Headroom(provider string) (Headroom, error) { + if a.lim.PerProvider[provider] <= 0 { + return Headroom{Provider: provider, Limited: false}, nil + } + spends, err := a.src.SpendByProviderSince(a.since()) + if err != nil { + return Headroom{}, err + } + return a.headroomFrom(provider, spends), nil +} + +// Allow reports whether starting a task on provider that may cost up to estCost +// is permitted without breaching the window cap. Unlimited providers always +// pass. estCost is typically the task's max_budget_usd (0 = unknown). +func (a *Accountant) Allow(provider string, estCost float64) (bool, error) { + h, err := a.Headroom(provider) + if err != nil { + return false, err + } + if !h.Limited { + return true, nil + } + return h.Spent+estCost <= h.Limit, nil +} + +// All returns headroom for every configured provider, sorted by name, in a +// single spend query. +func (a *Accountant) All() ([]Headroom, error) { + if len(a.lim.PerProvider) == 0 { + return []Headroom{}, nil + } + spends, err := a.src.SpendByProviderSince(a.since()) + if err != nil { + return nil, err + } + providers := make([]string, 0, len(a.lim.PerProvider)) + for p := range a.lim.PerProvider { + providers = append(providers, p) + } + sort.Strings(providers) + out := make([]Headroom, 0, len(providers)) + for _, p := range providers { + out = append(out, a.headroomFrom(p, spends)) + } + return out, nil +} diff --git a/internal/budget/budget_test.go b/internal/budget/budget_test.go new file mode 100644 index 0000000..c257c91 --- /dev/null +++ b/internal/budget/budget_test.go @@ -0,0 +1,112 @@ +package budget + +import ( + "testing" + "time" +) + +type fakeSpend struct { + spends map[string]float64 + since time.Time +} + +func (f *fakeSpend) SpendByProviderSince(since time.Time) (map[string]float64, error) { + f.since = since + return f.spends, nil +} + +func newAcct(spends map[string]float64, lim Limits) (*Accountant, *fakeSpend) { + src := &fakeSpend{spends: spends} + a := New(src, lim) + a.now = func() time.Time { return time.Date(2026, 5, 26, 12, 0, 0, 0, time.UTC) } + return a, src +} + +func TestNew_DefaultsWindow(t *testing.T) { + a, _ := newAcct(nil, Limits{}) + if a.lim.Window != DefaultWindow { + t.Errorf("want default window %v, got %v", DefaultWindow, a.lim.Window) + } +} + +func TestHeadroom_UnlimitedWhenNoLimit(t *testing.T) { + a, _ := newAcct(map[string]float64{"local": 999}, Limits{PerProvider: map[string]float64{"claude": 10}}) + h, err := a.Headroom("local") + if err != nil { + t.Fatal(err) + } + if h.Limited { + t.Errorf("local should be unlimited, got %+v", h) + } +} + +func TestHeadroom_ComputesRemainingAndFraction(t *testing.T) { + a, src := newAcct(map[string]float64{"claude": 4}, Limits{Window: 5 * time.Hour, PerProvider: map[string]float64{"claude": 10}}) + h, err := a.Headroom("claude") + if err != nil { + t.Fatal(err) + } + if !h.Limited || h.Limit != 10 || h.Spent != 4 || h.Remaining != 6 { + t.Errorf("unexpected headroom: %+v", h) + } + if h.Fraction != 0.6 { + t.Errorf("fraction: want 0.6, got %v", h.Fraction) + } + // Window lookback is now-5h. + want := time.Date(2026, 5, 26, 7, 0, 0, 0, time.UTC) + if !src.since.Equal(want) { + t.Errorf("since: want %v, got %v", want, src.since) + } +} + +func TestHeadroom_ClampsNegativeRemaining(t *testing.T) { + a, _ := newAcct(map[string]float64{"claude": 15}, Limits{PerProvider: map[string]float64{"claude": 10}}) + h, _ := a.Headroom("claude") + if h.Remaining != 0 { + t.Errorf("remaining should clamp to 0, got %v", h.Remaining) + } +} + +func TestAllow(t *testing.T) { + a, _ := newAcct(map[string]float64{"claude": 8}, Limits{PerProvider: map[string]float64{"claude": 10}}) + + if ok, _ := a.Allow("claude", 1.5); !ok { + t.Error("8 + 1.5 <= 10 should be allowed") + } + if ok, _ := a.Allow("claude", 3); ok { + t.Error("8 + 3 > 10 should be denied") + } + if ok, _ := a.Allow("local", 1000); !ok { + t.Error("unlimited provider should always allow") + } +} + +func TestAll_SortedAndSingleQuery(t *testing.T) { + a, src := newAcct(map[string]float64{"claude": 2, "gemini": 1}, + Limits{PerProvider: map[string]float64{"gemini": 5, "claude": 10}}) + calls := 0 + orig := src.spends + a2 := New(&countingSpend{spends: orig, calls: &calls}, a.lim) + a2.now = a.now + + hs, err := a2.All() + if err != nil { + t.Fatal(err) + } + if len(hs) != 2 || hs[0].Provider != "claude" || hs[1].Provider != "gemini" { + t.Errorf("want [claude gemini] sorted, got %+v", hs) + } + if calls != 1 { + t.Errorf("All should issue exactly one spend query, got %d", calls) + } +} + +type countingSpend struct { + spends map[string]float64 + calls *int +} + +func (c *countingSpend) SpendByProviderSince(time.Time) (map[string]float64, error) { + *c.calls++ + return c.spends, nil +} diff --git a/internal/config/config.go b/internal/config/config.go index 94f3ec7..58de95c 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -69,6 +69,17 @@ type Config struct { VAPIDEmail string `toml:"vapid_email"` ClaudeConfigDir string `toml:"claude_config_dir"` LocalModel LocalModel `toml:"local_model"` + Budget BudgetConfig `toml:"budget"` +} + +// BudgetConfig configures rolling per-provider spend caps. With no providers +// set, budget gating is disabled (nothing is blocked) — there is intentionally +// no default cap; the user must opt in by configuring limits. +type BudgetConfig struct { + // Window is the rolling lookback duration (e.g. "5h"); defaults to 5h. + Window string `toml:"window"` + // Provider5hUSD maps a provider (claude/gemini) to its USD cap within Window. + Provider5hUSD map[string]float64 `toml:"provider_5h_usd"` } func Default() (*Config, error) { diff --git a/internal/executor/executor.go b/internal/executor/executor.go index 6d6c528..4333f51 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -1025,6 +1025,7 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) { TaskID: t.ID, StartTime: time.Now().UTC(), Status: "RUNNING", + Agent: agentType, } // Pre-populate log paths so they're available in the DB immediately — diff --git a/internal/storage/db.go b/internal/storage/db.go index adb7c02..e03a902 100644 --- a/internal/storage/db.go +++ b/internal/storage/db.go @@ -140,6 +140,7 @@ func (s *DB) migrate() error { `ALTER TABLE tasks ADD COLUMN checker_report TEXT NOT NULL DEFAULT ''`, `ALTER TABLE executions ADD COLUMN tokens_in INTEGER`, `ALTER TABLE executions ADD COLUMN tokens_out INTEGER`, + `ALTER TABLE executions ADD COLUMN agent TEXT`, `CREATE TABLE IF NOT EXISTS events ( id TEXT PRIMARY KEY, task_id TEXT NOT NULL, @@ -536,6 +537,7 @@ type Execution struct { ErrorMsg string SessionID string // claude --session-id; persisted for resume SandboxDir string // preserved sandbox path when task is BLOCKED; resume must run here + Agent string // provider that ran this execution (claude/gemini/local); for budget accounting Changestats *task.Changestats // stored as JSON; nil if not yet recorded Commits []task.GitCommit // stored as JSON; empty if no commits @@ -584,10 +586,10 @@ func (s *DB) CreateExecutionAndSetRunning(e *Execution) error { commitsJSON = string(b) } if _, err := tx.Exec(` - INSERT INTO executions (id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?)`, + INSERT INTO executions (id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, agent) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?)`, e.ID, e.TaskID, e.StartTime.UTC(), e.EndTime.UTC(), e.ExitCode, e.Status, - e.StdoutPath, e.StderrPath, e.ArtifactDir, e.CostUSD, e.ErrorMsg, e.SessionID, e.SandboxDir, commitsJSON, + e.StdoutPath, e.StderrPath, e.ArtifactDir, e.CostUSD, e.ErrorMsg, e.SessionID, e.SandboxDir, commitsJSON, e.Agent, ); err != nil { return err } @@ -601,6 +603,32 @@ func (s *DB) CreateExecutionAndSetRunning(e *Execution) error { return tx.Commit() } +// SpendByProviderSince returns total cost_usd per provider (executions.agent) +// for executions started at or after `since`. Executions with no recorded agent +// are grouped under the empty string and can be ignored by callers. +func (s *DB) SpendByProviderSince(since time.Time) (map[string]float64, error) { + rows, err := s.db.Query(` + SELECT COALESCE(agent, ''), COALESCE(SUM(cost_usd), 0) + FROM executions + WHERE start_time >= ? + GROUP BY agent`, since.UTC()) + if err != nil { + return nil, err + } + defer rows.Close() + + out := make(map[string]float64) + for rows.Next() { + var agent string + var cost float64 + if err := rows.Scan(&agent, &cost); err != nil { + return nil, err + } + out[agent] = cost + } + return out, rows.Err() +} + // CreateExecution inserts an execution record. func (s *DB) CreateExecution(e *Execution) error { var changestatsJSON *string @@ -621,23 +649,23 @@ func (s *DB) CreateExecution(e *Execution) error { commitsJSON = string(b) } _, err := s.db.Exec(` - INSERT INTO executions (id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + INSERT INTO executions (id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, e.ID, e.TaskID, e.StartTime.UTC(), e.EndTime.UTC(), e.ExitCode, e.Status, - e.StdoutPath, e.StderrPath, e.ArtifactDir, e.CostUSD, e.ErrorMsg, e.SessionID, e.SandboxDir, changestatsJSON, commitsJSON, e.TokensIn, e.TokensOut, + e.StdoutPath, e.StderrPath, e.ArtifactDir, e.CostUSD, e.ErrorMsg, e.SessionID, e.SandboxDir, changestatsJSON, commitsJSON, e.TokensIn, e.TokensOut, e.Agent, ) return err } // GetExecution retrieves an execution by ID. func (s *DB) GetExecution(id string) (*Execution, error) { - row := s.db.QueryRow(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out FROM executions WHERE id = ?`, id) + row := s.db.QueryRow(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent FROM executions WHERE id = ?`, id) return scanExecution(row) } // ListExecutions returns executions for a task. func (s *DB) ListExecutions(taskID string) ([]*Execution, error) { - rows, err := s.db.Query(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out FROM executions WHERE task_id = ? ORDER BY start_time DESC`, taskID) + rows, err := s.db.Query(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent FROM executions WHERE task_id = ? ORDER BY start_time DESC`, taskID) if err != nil { return nil, err } @@ -656,7 +684,7 @@ func (s *DB) ListExecutions(taskID string) ([]*Execution, error) { // GetLatestExecution returns the most recent execution for a task. func (s *DB) GetLatestExecution(taskID string) (*Execution, error) { - row := s.db.QueryRow(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out FROM executions WHERE task_id = ? ORDER BY start_time DESC LIMIT 1`, taskID) + row := s.db.QueryRow(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent FROM executions WHERE task_id = ? ORDER BY start_time DESC LIMIT 1`, taskID) return scanExecution(row) } @@ -1137,8 +1165,9 @@ func scanExecution(row scanner) (*Execution, error) { var commitsJSON sql.NullString var tokensIn sql.NullInt64 var tokensOut sql.NullInt64 + var agent sql.NullString err := row.Scan(&e.ID, &e.TaskID, &e.StartTime, &e.EndTime, &e.ExitCode, &e.Status, - &e.StdoutPath, &e.StderrPath, &e.ArtifactDir, &e.CostUSD, &e.ErrorMsg, &sessionID, &sandboxDir, &changestatsJSON, &commitsJSON, &tokensIn, &tokensOut) + &e.StdoutPath, &e.StderrPath, &e.ArtifactDir, &e.CostUSD, &e.ErrorMsg, &sessionID, &sandboxDir, &changestatsJSON, &commitsJSON, &tokensIn, &tokensOut, &agent) if err != nil { return nil, err } @@ -1146,6 +1175,7 @@ func scanExecution(row scanner) (*Execution, error) { e.SandboxDir = sandboxDir.String e.TokensIn = tokensIn.Int64 e.TokensOut = tokensOut.Int64 + e.Agent = agent.String if changestatsJSON.Valid && changestatsJSON.String != "" { var cs task.Changestats if err := json.Unmarshal([]byte(changestatsJSON.String), &cs); err != nil { diff --git a/internal/storage/db_test.go b/internal/storage/db_test.go index 0e67e02..4d744a4 100644 --- a/internal/storage/db_test.go +++ b/internal/storage/db_test.go @@ -309,6 +309,44 @@ func TestListTasks_WithLimit(t *testing.T) { } } +func TestSpendByProviderSince(t *testing.T) { + db := testDB(t) + now := time.Now().UTC() + + tk := &task.Task{ + ID: "spend-task", Name: "S", Agent: task.AgentConfig{Type: "claude", Instructions: "x"}, + Priority: task.PriorityNormal, Retry: task.RetryConfig{MaxAttempts: 1, Backoff: "linear"}, + Tags: []string{}, DependsOn: []string{}, State: task.StatePending, CreatedAt: now, UpdatedAt: now, + } + if err := db.CreateTask(tk); err != nil { + t.Fatal(err) + } + + // Two recent claude execs, one recent gemini, one old claude (outside window). + execs := []*Execution{ + {ID: "s1", TaskID: "spend-task", StartTime: now.Add(-1 * time.Hour), Status: "COMPLETED", CostUSD: 1.5, Agent: "claude"}, + {ID: "s2", TaskID: "spend-task", StartTime: now.Add(-2 * time.Hour), Status: "COMPLETED", CostUSD: 2.0, Agent: "claude"}, + {ID: "s3", TaskID: "spend-task", StartTime: now.Add(-30 * time.Minute), Status: "COMPLETED", CostUSD: 0.75, Agent: "gemini"}, + {ID: "s4", TaskID: "spend-task", StartTime: now.Add(-10 * time.Hour), Status: "COMPLETED", CostUSD: 99, Agent: "claude"}, + } + for _, e := range execs { + if err := db.CreateExecution(e); err != nil { + t.Fatalf("create exec %s: %v", e.ID, err) + } + } + + spends, err := db.SpendByProviderSince(now.Add(-5 * time.Hour)) + if err != nil { + t.Fatalf("SpendByProviderSince: %v", err) + } + if got := spends["claude"]; got != 3.5 { + t.Errorf("claude spend: want 3.5 (old $99 excluded), got %v", got) + } + if got := spends["gemini"]; got != 0.75 { + t.Errorf("gemini spend: want 0.75, got %v", got) + } +} + func TestCreateExecution_AndGet(t *testing.T) { db := testDB(t) now := time.Now().UTC().Truncate(time.Second) -- cgit v1.2.3 From 32715355fe2eed321df4f7083dfe580d35f8a62a Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 26 May 2026 20:31:24 +0000 Subject: feat(executor): budget-gate the dispatcher — reroute to local or block (Phase 6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pool now consults a BudgetGate before taking an agent slot. If running on the chosen paid provider could breach its rolling window (estimated by the task's max_budget_usd), it reroutes to the free local runner when one is registered; otherwise it stops before running and marks the task BUDGET_EXCEEDED. serve.go builds a budget.Accountant from [budget] config and wires it into the pool (only when limits are configured). Allows the QUEUED → BUDGET_EXCEEDED transition, since the gate blocks a queued task before it ever reaches RUNNING. Covered by pool tests for both the block and reroute paths against a fake gate. https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39 --- internal/cli/serve.go | 18 ++++++++++++ internal/executor/executor.go | 46 ++++++++++++++++++++++++++++++ internal/executor/executor_test.go | 57 ++++++++++++++++++++++++++++++++++++++ internal/task/task.go | 2 +- 4 files changed, 122 insertions(+), 1 deletion(-) (limited to 'internal/executor/executor.go') diff --git a/internal/cli/serve.go b/internal/cli/serve.go index e545763..55fdaf5 100644 --- a/internal/cli/serve.go +++ b/internal/cli/serve.go @@ -11,6 +11,7 @@ import ( "time" "github.com/thepeterstone/claudomator/internal/api" + "github.com/thepeterstone/claudomator/internal/budget" "github.com/thepeterstone/claudomator/internal/executor" "github.com/thepeterstone/claudomator/internal/notify" "github.com/thepeterstone/claudomator/internal/storage" @@ -144,6 +145,23 @@ func serve(addr string) error { pool.LLM = localClient } + // Budget accountant: gate paid providers against rolling per-provider caps. + // With no limits configured this is created but allows everything. + var accountant *budget.Accountant + if len(cfg.Budget.Provider5hUSD) > 0 { + window := budget.DefaultWindow + if cfg.Budget.Window != "" { + if d, derr := time.ParseDuration(cfg.Budget.Window); derr == nil { + window = d + } else { + logger.Warn("invalid budget.window; using default", "value", cfg.Budget.Window, "default", window) + } + } + accountant = budget.New(store, budget.Limits{Window: window, PerProvider: cfg.Budget.Provider5hUSD}) + pool.Budget = accountant + logger.Info("budget gating enabled", "window", window, "limits", cfg.Budget.Provider5hUSD) + } + if err := store.SeedProjects(); err != nil { logger.Error("failed to seed projects", "error", err) } diff --git a/internal/executor/executor.go b/internal/executor/executor.go index 4333f51..8614481 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -93,6 +93,14 @@ type Pool struct { dispatchDone chan struct{} // closed when the dispatch goroutine exits Classifier *Classifier LLM *llm.Client + // Budget gates paid providers against a rolling window; nil disables gating. + Budget BudgetGate +} + +// BudgetGate reports whether a task estimated at estCost may run on a provider +// without breaching its rolling spend window. Satisfied by *budget.Accountant. +type BudgetGate interface { + Allow(provider string, estCost float64) (bool, error) } // Result is emitted when a task execution completes. @@ -937,6 +945,44 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) { agentType = "claude" } + // Budget gating: if running on this provider could breach its rolling window + // (estimated by the task's max_budget_usd), reroute to the free local runner + // when one is registered, otherwise stop and mark the task BUDGET_EXCEEDED. + if p.Budget != nil { + if ok, berr := p.Budget.Allow(agentType, t.Agent.MaxBudgetUSD); berr != nil { + p.logger.Warn("budget check failed; proceeding", "error", berr, "taskID", t.ID) + } else if !ok { + if _, hasLocal := p.runners["local"]; hasLocal && agentType != "local" { + p.logger.Info("budget window would be breached; rerouting to local", "taskID", t.ID, "from", agentType) + t.Agent.Type = "local" + t.Agent.Model = "" + agentType = "local" + if uerr := p.store.UpdateTaskAgent(t.ID, t.Agent); uerr != nil { + p.logger.Error("failed to persist rerouted agent", "error", uerr, "taskID", t.ID) + } + } else { + now := time.Now().UTC() + exec := &storage.Execution{ + ID: uuid.New().String(), + TaskID: t.ID, + StartTime: now, + EndTime: now, + Status: "BUDGET_EXCEEDED", + Agent: agentType, + ErrorMsg: fmt.Sprintf("rolling budget window cap reached for provider %q", agentType), + } + if createErr := p.store.CreateExecution(exec); createErr != nil { + p.logger.Error("failed to create execution record", "error", createErr) + } + if err := p.store.UpdateTaskState(t.ID, task.StateBudgetExceeded); err != nil { + p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateBudgetExceeded, "error", err) + } + p.resultCh <- &Result{TaskID: t.ID, Execution: exec, Err: fmt.Errorf("budget cap reached for %s", agentType)} + return + } + } + } + // Check dependencies before taking the per-agent slot to avoid deadlock: // if a dependent task holds the slot while waiting for its dependency to run, // the dependency can never start (maxPerAgent=1). diff --git a/internal/executor/executor_test.go b/internal/executor/executor_test.go index d98efbf..b737e22 100644 --- a/internal/executor/executor_test.go +++ b/internal/executor/executor_test.go @@ -217,6 +217,63 @@ func TestPool_Submit_Subtask_GoesToCompleted(t *testing.T) { } } +type fakeGate struct{ deny map[string]bool } + +func (g fakeGate) Allow(provider string, _ float64) (bool, error) { return !g.deny[provider], nil } + +func TestPool_BudgetGate_BlocksWhenNoLocalFallback(t *testing.T) { + store := testStore(t) + runner := &mockRunner{} + pool := NewPool(2, map[string]Runner{"claude": runner}, store, slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError}))) + pool.Budget = fakeGate{deny: map[string]bool{"claude": true}} + + tk := makeTask("bg-1") + store.CreateTask(tk) + if err := pool.Submit(context.Background(), tk); err != nil { + t.Fatalf("submit: %v", err) + } + + result := <-pool.Results() + if result.Execution.Status != "BUDGET_EXCEEDED" { + t.Errorf("status: want BUDGET_EXCEEDED, got %q", result.Execution.Status) + } + if runner.callCount() != 0 { + t.Errorf("claude runner should not have run, got %d calls", runner.callCount()) + } + got, _ := store.GetTask("bg-1") + if got.State != task.StateBudgetExceeded { + t.Errorf("task state: want BUDGET_EXCEEDED, got %v", got.State) + } +} + +func TestPool_BudgetGate_ReroutesToLocal(t *testing.T) { + store := testStore(t) + claudeRunner := &mockRunner{} + localRunner := &mockRunner{} + pool := NewPool(2, map[string]Runner{"claude": claudeRunner, "local": localRunner}, store, slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError}))) + pool.Budget = fakeGate{deny: map[string]bool{"claude": true}} // local is allowed + + tk := makeTask("bg-2") + store.CreateTask(tk) + if err := pool.Submit(context.Background(), tk); err != nil { + t.Fatalf("submit: %v", err) + } + + result := <-pool.Results() + if result.Err != nil { + t.Errorf("expected success after reroute, got %v", result.Err) + } + if localRunner.callCount() != 1 { + t.Errorf("local runner should have run once, got %d", localRunner.callCount()) + } + if claudeRunner.callCount() != 0 { + t.Errorf("claude runner should not have run, got %d", claudeRunner.callCount()) + } + if result.Execution.Agent != "local" { + t.Errorf("execution agent: want local, got %q", result.Execution.Agent) + } +} + func TestPool_Submit_Failure(t *testing.T) { store := testStore(t) runner := &mockRunner{err: fmt.Errorf("boom"), exitCode: 1} diff --git a/internal/task/task.go b/internal/task/task.go index 935a238..eeac49e 100644 --- a/internal/task/task.go +++ b/internal/task/task.go @@ -138,7 +138,7 @@ type BatchFile struct { // BLOCKED may advance to READY when all subtasks complete, or back to QUEUED on user answer. var validTransitions = map[State]map[State]bool{ StatePending: {StateQueued: true, StateCancelled: true}, - StateQueued: {StateRunning: true, StateCancelled: true, StateReady: true}, // READY: parent task completed via subtask delegation + StateQueued: {StateRunning: true, StateCancelled: true, StateReady: true, StateBudgetExceeded: true}, // READY: subtask delegation; BUDGET_EXCEEDED: dispatcher budget gate blocks before running StateRunning: {StateReady: true, StateCompleted: true, StateFailed: true, StateTimedOut: true, StateCancelled: true, StateBudgetExceeded: true, StateBlocked: true}, StateReady: {StateCompleted: true, StatePending: true}, StateFailed: {StateQueued: true}, // retry -- cgit v1.2.3