summaryrefslogtreecommitdiff
path: root/internal/executor/local.go
diff options
context:
space:
mode:
authorPeter Stone <thepeterstone@gmail.com>2026-06-03 23:01:39 +0000
committerPeter Stone <thepeterstone@gmail.com>2026-06-03 23:01:39 +0000
commit68cf1e24dd2ca2612b72babc4b35280cd3512f92 (patch)
tree14b73f21570a2f42ae729ca7e9676de6628d6819 /internal/executor/local.go
parentb28cfc6ff288d083f6c8e9c055b69bfcadbceccc (diff)
parent7388337d3be5cc7f65ef547e30b2e39884dd165b (diff)
merge: integrate oss branch — budget gating, MCP back-channel, events, loopback-only bind
Key features from OSS branch: - Budget gating: rolling per-provider spend caps with BUDGET_EXCEEDED task state - Agent MCP back-channel: runners mint tokens; /mcp endpoint resolves them - Chatbot MCP server at /chatbot/mcp (requires api_token in config) - Event timeline: unified event stream replacing ad-hoc question/summary flows - Loopback-only default bind (127.0.0.1:8484); external_bind_allowed=true to expose - repository_url required on task creation (enforces traceability) - Auto-checker: spawns verification task after each execution Conflict resolutions: - serve.go: keep cfg.Runners.XEnabled() conditional registration from main + agentRegistry from oss - config.go: keep RunnersConfig from main + BudgetConfig/ExternalBindAllowed from oss - server.go: handleStaticFiles (base-path rewrite) kept; deduplicated duplicate from both sides - executor/claude.go: accepted oss deletion (ClaudeRunner replaced by ContainerRunner) - container_test.go: added noopChannel + AgentChannel parameter to Run call Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Diffstat (limited to 'internal/executor/local.go')
-rw-r--r--internal/executor/local.go114
1 files changed, 80 insertions, 34 deletions
diff --git a/internal/executor/local.go b/internal/executor/local.go
index 5d874c6..3f20fe7 100644
--- a/internal/executor/local.go
+++ b/internal/executor/local.go
@@ -37,10 +37,17 @@ func (r *LocalRunner) ExecLogDir(execID string) string {
return filepath.Join(r.LogDir, execID)
}
-// 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 {
+// maxLocalToolTurns bounds the tool-use loop so a misbehaving model cannot spin
+// forever calling tools without finishing.
+const maxLocalToolTurns = 12
+
+// Run drives a chat completion against the local endpoint with the agent
+// back-channel tools (ask_user/report_summary/spawn_subtask/record_progress)
+// declared. It loops, feeding tool results back as message history, until the
+// model stops calling tools. Assistant text is written to stdout.log in the
+// same stream-json envelope Claude uses so downstream parsers keep working.
+// If the model calls ask_user, the run stops and returns a *BlockedError.
+func (r *LocalRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution, ch AgentChannel) error {
if r.Client == nil {
return fmt.Errorf("local runner: no LLM client configured")
}
@@ -70,56 +77,95 @@ func (r *LocalRunner) Run(ctx context.Context, t *task.Task, e *storage.Executio
if sys := strings.TrimSpace(t.Agent.SystemPromptAppend); sys != "" {
messages = append(messages, llm.Message{Role: "system", Content: sys})
}
- messages = append(messages, llm.Message{Role: "user", Content: t.Agent.Instructions})
+ // The agent tools are real here, so guide the model toward them with the
+ // same planning preamble the container runners use (unless skip_planning).
+ messages = append(messages, llm.Message{Role: "user", Content: buildAgentInstructions(t, true)})
temperature := t.Agent.Temperature
if temperature == nil && r.DefaultTemperature > 0 {
v := r.DefaultTemperature
temperature = &v
}
-
- req := llm.ChatRequest{
- Model: t.Agent.Model,
- Messages: messages,
- Temperature: temperature,
- MaxTokens: t.Agent.MaxTokens,
- }
+ tools := agentToolDefs()
start := time.Now()
- resp, err := r.Client.ChatStream(ctx, req, func(delta string) {
- if delta == "" {
- return
+ var totalIn, totalOut int
+ var lastModel, lastFinish string
+ blocked := false
+
+ for turn := 0; turn < maxLocalToolTurns; turn++ {
+ resp, chatErr := r.Client.Chat(ctx, llm.ChatRequest{
+ Model: t.Agent.Model,
+ Messages: messages,
+ Temperature: temperature,
+ MaxTokens: t.Agent.MaxTokens,
+ Tools: tools,
+ })
+ if chatErr != nil {
+ writeResultLine(stdout, "error", chatErr.Error(), totalIn, totalOut)
+ return fmt.Errorf("local runner: chat: %w", chatErr)
+ }
+ totalIn += resp.PromptTokens
+ totalOut += resp.OutputTokens
+ lastModel, lastFinish = resp.Model, resp.FinishReason
+
+ if resp.Content != "" {
+ writeAssistantTextLine(stdout, resp.Content)
}
- writeAssistantTextLine(stdout, delta)
- })
- if err != nil {
- writeResultLine(stdout, "error", err.Error(), 0, 0)
- return fmt.Errorf("local runner: chat: %w", err)
- }
- elapsed := time.Since(start)
- // Write one consolidated assistant envelope containing the full response.
- // extractSummary and ParseChangestatFromOutput operate per-line, so a
- // single envelope with the full text is what they expect to find.
- if resp.Content != "" {
- writeAssistantTextLine(stdout, resp.Content)
+ if len(resp.ToolCalls) == 0 {
+ break
+ }
+
+ // Re-feed: record the assistant's tool-call turn, then each tool result.
+ messages = append(messages, llm.Message{Role: "assistant", Content: resp.Content, ToolCalls: resp.ToolCalls})
+ for _, tc := range resp.ToolCalls {
+ result, didBlock, toolErr := dispatchAgentTool(ctx, ch, tc.Function.Name, tc.Function.Arguments)
+ if toolErr != nil {
+ writeResultLine(stdout, "error", toolErr.Error(), totalIn, totalOut)
+ return fmt.Errorf("local runner: tool %s: %w", tc.Function.Name, toolErr)
+ }
+ if didBlock {
+ blocked = true
+ break
+ }
+ messages = append(messages, llm.Message{
+ Role: "tool",
+ ToolCallID: tc.ID,
+ Name: tc.Function.Name,
+ Content: result,
+ })
+ }
+ if blocked {
+ break
+ }
}
- writeResultLine(stdout, "success", "", resp.PromptTokens, resp.OutputTokens)
+ elapsed := time.Since(start)
e.CostUSD = 0
- e.TokensIn = int64(resp.PromptTokens)
- e.TokensOut = int64(resp.OutputTokens)
+ e.TokensIn = int64(totalIn)
+ e.TokensOut = int64(totalOut)
if r.Logger != nil {
r.Logger.Info("local runner completed",
"taskID", t.ID,
- "model", resp.Model,
- "tokens_in", resp.PromptTokens,
- "tokens_out", resp.OutputTokens,
- "finish_reason", resp.FinishReason,
+ "model", lastModel,
+ "tokens_in", totalIn,
+ "tokens_out", totalOut,
+ "finish_reason", lastFinish,
+ "blocked", blocked,
"elapsed_ms", elapsed.Milliseconds(),
)
}
+
+ // If the model asked the user, stop and block. Local resume re-feeds the
+ // conversation (per-runner sovereign state) — not yet wired, so the session
+ // ID is empty and resume starts fresh for now.
+ if q, isBlocked := channelPendingQuestion(ch); isBlocked {
+ return &BlockedError{QuestionJSON: q}
+ }
+
+ writeResultLine(stdout, "success", "", totalIn, totalOut)
return nil
}