summaryrefslogtreecommitdiff
path: root/internal/executor/local.go
diff options
context:
space:
mode:
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
}