feat(executor,llm): LocalRunner agent-channel via OpenAI tool-use (Phase 5)

LocalRunner previously ignored the AgentChannel and produced a single
fire-and-forget completion. It now declares the four agent back-channel tools
(ask_user/report_summary/spawn_subtask/record_progress) as OpenAI
function-calling definitions and runs a tool-use loop: each turn feeds tool
results back as message history (re-feed) until the model stops calling tools,
bounded by maxLocalToolTurns. ask_user converts a buffered question into a
*BlockedError so the task blocks like the container runners.

Adds tool-use support to the llm client (Tool/ToolCall/ToolFunction types,
Tools on ChatRequest, ToolCalls on ChatResponse + wire request/response). The
loop uses non-streaming Chat (tool_calls don't stream cleanly); assistant text
is still written to stdout.log in the Claude stream-json envelope so summary/
changestats parsing is unchanged.

Fully tested against a mock OpenAI endpoint + storeChannel: spawn/summary/
progress dispatch, ask_user blocking, token accumulation, and the llm tools
round-trip. NOTE: local resume re-feeds conversation state (Decision #8) — not
yet wired, so a blocked local task resumes fresh for now.

https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39

3 files changed, 352 insertions, 90 deletions

diff --git a/internal/executor/local.go b/internal/executor/local.go
index e3c9c2c..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, _ AgentChannel) 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
 }
 
diff --git a/internal/executor/local_test.go b/internal/executor/local_test.go
index ffe87f9..2ae8380 100644
--- a/internal/executor/local_test.go
+++ b/internal/executor/local_test.go
@@ -3,7 +3,7 @@ package executor
 import (
 	"context"
 	"encoding/json"
-	"fmt"
+	"errors"
 	"io"
 	"log/slog"
 	"net/http"
@@ -11,6 +11,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/google/uuid"
@@ -19,60 +20,77 @@ import (
 	"github.com/thepeterstone/claudomator/internal/task"
 )
 
-// fakeOpenAIServer returns an httptest.Server that replies with a streaming
-// chat completion containing the supplied content (split into chunks) plus a
-// usage record.
-func fakeOpenAIServer(t *testing.T, chunks []string, promptTok, outTok int) *httptest.Server {
+// fakeTurn is one canned chat-completion response the fake server returns.
+type fakeTurn struct {
+	content   string
+	toolCalls []llm.ToolCall
+	promptTok int
+	outTok    int
+}
+
+// fakeChatServer replies to /chat/completions with the supplied turns in order,
+// one per request (non-streaming JSON), so a tool-use loop can be driven
+// deterministically. Requests beyond the list repeat the final turn.
+func fakeChatServer(t *testing.T, turns []fakeTurn) *httptest.Server {
 	t.Helper()
-	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		w.Header().Set("Content-Type", "text/event-stream")
-		flusher, _ := w.(http.Flusher)
-		for _, c := range chunks {
-			payload := map[string]any{
-				"model":   "fake",
-				"choices": []map[string]any{{"delta": map[string]string{"content": c}}},
-			}
-			b, _ := json.Marshal(payload)
-			fmt.Fprintf(w, "data: %s\n\n", b)
-			if flusher != nil {
-				flusher.Flush()
-			}
+	var mu sync.Mutex
+	idx := 0
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		mu.Lock()
+		turn := turns[idx]
+		if idx < len(turns)-1 {
+			idx++
+		}
+		mu.Unlock()
+
+		msg := map[string]any{"role": "assistant", "content": turn.content}
+		if len(turn.toolCalls) > 0 {
+			msg["tool_calls"] = turn.toolCalls
 		}
-		final := map[string]any{
+		resp := map[string]any{
 			"model":   "fake",
-			"choices": []map[string]any{{"delta": map[string]string{}, "finish_reason": "stop"}},
-			"usage":   map[string]int{"prompt_tokens": promptTok, "completion_tokens": outTok},
+			"choices": []map[string]any{{"message": msg, "finish_reason": "stop"}},
+			"usage":   map[string]int{"prompt_tokens": turn.promptTok, "completion_tokens": turn.outTok},
 		}
-		fb, _ := json.Marshal(final)
-		fmt.Fprintf(w, "data: %s\n\ndata: [DONE]\n\n", fb)
+		_ = json.NewEncoder(w).Encode(resp)
 	}))
 }
 
-func TestLocalRunner_Run_WritesStreamJSON(t *testing.T) {
-	srv := fakeOpenAIServer(t,
-		[]string{"## Summary\n", "All ", "good."},
-		11, 22,
-	)
-	defer srv.Close()
+func toolCall(id, name, args string) llm.ToolCall {
+	return llm.ToolCall{ID: id, Type: "function", Function: llm.ToolCallFunction{Name: name, Arguments: args}}
+}
 
-	logRoot := t.TempDir()
-	r := &LocalRunner{
+func newLocalRunner(t *testing.T, srv *httptest.Server) *LocalRunner {
+	t.Helper()
+	return &LocalRunner{
 		Client: &llm.Client{Endpoint: srv.URL + "/v1", Model: "fake"},
 		Logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
-		LogDir: logRoot,
+		LogDir: t.TempDir(),
 	}
-	tt := &task.Task{
+}
+
+func localTask() *task.Task {
+	return &task.Task{
 		ID:   "task-1",
 		Name: "test",
 		Agent: task.AgentConfig{
 			Type:         "local",
 			Model:        "fake",
 			Instructions: "Do a thing.",
+			SkipPlanning: true, // keep the preamble out of these assertions
 		},
 	}
+}
+
+func TestLocalRunner_Run_WritesStreamJSON(t *testing.T) {
+	srv := fakeChatServer(t, []fakeTurn{{content: "## Summary\nAll good.", promptTok: 11, outTok: 22}})
+	defer srv.Close()
+
+	r := newLocalRunner(t, srv)
+	tt := localTask()
 	exec := &storage.Execution{ID: uuid.New().String(), TaskID: tt.ID}
 
-	if err := r.Run(context.Background(), tt, exec, newStoreChannel(nil, tt.ID)); err != nil {
+	if err := r.Run(context.Background(), tt, exec, newStoreChannel(&fakeChannelStore{}, tt.ID)); err != nil {
 		t.Fatalf("Run: %v", err)
 	}
 
@@ -83,40 +101,103 @@ func TestLocalRunner_Run_WritesStreamJSON(t *testing.T) {
 		t.Errorf("tokens: want 11/22 got %d/%d", exec.TokensIn, exec.TokensOut)
 	}
 
-	// Verify stdout.log contains stream-json envelopes that extractSummary can parse.
 	stdoutPath := filepath.Join(r.ExecLogDir(exec.ID), "stdout.log")
 	data, err := os.ReadFile(stdoutPath)
 	if err != nil {
 		t.Fatalf("read stdout: %v", err)
 	}
 	lines := strings.Split(strings.TrimSpace(string(data)), "\n")
-	if len(lines) < 4 {
-		t.Fatalf("expected at least 4 lines (3 deltas + 1 result), got %d:\n%s", len(lines), data)
-	}
-	for i, line := range lines[:3] {
-		var env struct {
-			Type    string `json:"type"`
-			Message struct {
-				Content []struct {
-					Type string `json:"type"`
-					Text string `json:"text"`
-				}
-			}
-		}
-		if err := json.Unmarshal([]byte(line), &env); err != nil {
-			t.Fatalf("line %d not JSON: %v: %s", i, err, line)
-		}
-		if env.Type != "assistant" {
-			t.Errorf("line %d: want type=assistant, got %q", i, env.Type)
-		}
+	// One assistant envelope + one result line.
+	if len(lines) != 2 {
+		t.Fatalf("expected 2 lines (assistant + result), got %d:\n%s", len(lines), data)
+	}
+	var env struct {
+		Type string `json:"type"`
+	}
+	if err := json.Unmarshal([]byte(lines[0]), &env); err != nil || env.Type != "assistant" {
+		t.Fatalf("line 0 should be an assistant envelope: %v: %s", err, lines[0])
 	}
 
-	summary := extractSummary(stdoutPath)
-	if !strings.Contains(summary, "All good.") {
+	if summary := extractSummary(stdoutPath); !strings.Contains(summary, "All good.") {
 		t.Errorf("extractSummary should find 'All good.', got %q", summary)
 	}
 }
 
+func TestLocalRunner_Run_ToolLoop_ReportSummaryAndSpawn(t *testing.T) {
+	// Turn 1: model spawns a subtask. Turn 2: reports summary. Turn 3: finishes.
+	srv := fakeChatServer(t, []fakeTurn{
+		{toolCalls: []llm.ToolCall{toolCall("c1", "spawn_subtask", `{"name":"sub A","instructions":"do sub"}`)}, promptTok: 5, outTok: 3},
+		{toolCalls: []llm.ToolCall{toolCall("c2", "report_summary", `{"summary":"all done"}`)}, promptTok: 4, outTok: 2},
+		{content: "finished", promptTok: 2, outTok: 1},
+	})
+	defer srv.Close()
+
+	r := newLocalRunner(t, srv)
+	tt := localTask()
+	store := &fakeChannelStore{}
+	ch := newStoreChannel(store, tt.ID)
+	exec := &storage.Execution{ID: uuid.New().String(), TaskID: tt.ID}
+
+	if err := r.Run(context.Background(), tt, exec, ch); err != nil {
+		t.Fatalf("Run: %v", err)
+	}
+
+	if len(store.createdTasks) != 1 || store.createdTasks[0].Name != "sub A" {
+		t.Errorf("expected one spawned subtask 'sub A', got %+v", store.createdTasks)
+	}
+	if store.createdTasks[0].ParentTaskID != tt.ID {
+		t.Errorf("subtask parent: want %q, got %q", tt.ID, store.createdTasks[0].ParentTaskID)
+	}
+	if sum, ok := ch.ReportedSummary(); !ok || sum != "all done" {
+		t.Errorf("expected reported summary 'all done', got %q (set=%v)", sum, ok)
+	}
+	// Tokens accumulate across all three turns.
+	if exec.TokensIn != 11 || exec.TokensOut != 6 {
+		t.Errorf("tokens should accumulate: want 11/6, got %d/%d", exec.TokensIn, exec.TokensOut)
+	}
+}
+
+func TestLocalRunner_Run_RecordProgress(t *testing.T) {
+	srv := fakeChatServer(t, []fakeTurn{
+		{toolCalls: []llm.ToolCall{toolCall("c1", "record_progress", `{"message":"halfway there"}`)}},
+		{content: "done"},
+	})
+	defer srv.Close()
+
+	store := &fakeChannelStore{}
+	tt := localTask()
+	exec := &storage.Execution{ID: uuid.New().String(), TaskID: tt.ID}
+	if err := newLocalRunner(t, srv).Run(context.Background(), tt, exec, newStoreChannel(store, tt.ID)); err != nil {
+		t.Fatalf("Run: %v", err)
+	}
+	if len(store.createdEvents) != 1 {
+		t.Fatalf("expected 1 progress event, got %d", len(store.createdEvents))
+	}
+	if !strings.Contains(string(store.createdEvents[0].Payload), "halfway there") {
+		t.Errorf("progress event payload: %s", store.createdEvents[0].Payload)
+	}
+}
+
+func TestLocalRunner_Run_AskUser_Blocks(t *testing.T) {
+	srv := fakeChatServer(t, []fakeTurn{
+		{toolCalls: []llm.ToolCall{toolCall("c1", "ask_user", `{"question":"which branch?"}`)}},
+		{content: "should not be reached"},
+	})
+	defer srv.Close()
+
+	tt := localTask()
+	exec := &storage.Execution{ID: uuid.New().String(), TaskID: tt.ID}
+	err := newLocalRunner(t, srv).Run(context.Background(), tt, exec, newStoreChannel(&fakeChannelStore{}, tt.ID))
+
+	var be *BlockedError
+	if !errors.As(err, &be) {
+		t.Fatalf("expected *BlockedError, got %v", err)
+	}
+	if !strings.Contains(be.QuestionJSON, "which branch?") {
+		t.Errorf("BlockedError question: %q", be.QuestionJSON)
+	}
+}
+
 func TestLocalRunner_Run_NoClient_Errors(t *testing.T) {
 	r := &LocalRunner{LogDir: t.TempDir()}
 	tt := &task.Task{ID: "x", Agent: task.AgentConfig{Instructions: "hi"}}
diff --git a/internal/executor/localtools.go b/internal/executor/localtools.go
new file mode 100644
index 0000000..48b862f
--- /dev/null
+++ b/internal/executor/localtools.go
@@ -0,0 +1,135 @@
+package executor
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/thepeterstone/claudomator/internal/llm"
+)
+
+// agentToolDefs returns the four agent back-channel tools as OpenAI
+// function-calling definitions, for runners that talk to an OpenAI-compatible
+// endpoint (LocalRunner). They mirror the MCP tools the container runners expose.
+func agentToolDefs() []llm.Tool {
+	strProp := func(desc string) map[string]any {
+		return map[string]any{"type": "string", "description": desc}
+	}
+	return []llm.Tool{
+		{Type: "function", Function: llm.ToolFunction{
+			Name:        "ask_user",
+			Description: "Ask the user a question when you genuinely need a decision to proceed. The task pauses until the user answers; do not call other tools after this.",
+			Parameters: map[string]any{
+				"type": "object",
+				"properties": map[string]any{
+					"question": strProp("the question to ask, phrased as a real question"),
+					"options":  map[string]any{"type": "array", "items": map[string]any{"type": "string"}, "description": "optional suggested answer choices"},
+				},
+				"required": []string{"question"},
+			},
+		}},
+		{Type: "function", Function: llm.ToolFunction{
+			Name:        "report_summary",
+			Description: "Record a concise 2-5 sentence summary of what you accomplished. Call this before finishing.",
+			Parameters: map[string]any{
+				"type":       "object",
+				"properties": map[string]any{"summary": strProp("the summary text")},
+				"required":   []string{"summary"},
+			},
+		}},
+		{Type: "function", Function: llm.ToolFunction{
+			Name:        "spawn_subtask",
+			Description: "Create a child task to be executed separately. Use this to break large work into focused pieces.",
+			Parameters: map[string]any{
+				"type": "object",
+				"properties": map[string]any{
+					"name":           strProp("short descriptive name for the subtask"),
+					"instructions":   strProp("complete instructions for the subtask agent"),
+					"model":          strProp("optional model override"),
+					"max_budget_usd": map[string]any{"type": "number", "description": "optional budget cap in USD"},
+				},
+				"required": []string{"name", "instructions"},
+			},
+		}},
+		{Type: "function", Function: llm.ToolFunction{
+			Name:        "record_progress",
+			Description: "Record a short progress note that appears in the task timeline.",
+			Parameters: map[string]any{
+				"type":       "object",
+				"properties": map[string]any{"message": strProp("a short progress note")},
+				"required":   []string{"message"},
+			},
+		}},
+	}
+}
+
+// dispatchAgentTool invokes one model-requested tool against the AgentChannel.
+// It returns the text to feed back to the model as the tool result, and a
+// blocked flag set when ask_user could not be answered in-session (the run must
+// stop and the task block).
+func dispatchAgentTool(ctx context.Context, ch AgentChannel, name, argsJSON string) (result string, blocked bool, err error) {
+	switch name {
+	case "ask_user":
+		var a struct {
+			Question string   `json:"question"`
+			Options  []string `json:"options"`
+		}
+		_ = json.Unmarshal([]byte(argsJSON), &a)
+		q := map[string]any{"text": a.Question}
+		if len(a.Options) > 0 {
+			q["options"] = a.Options
+		}
+		payload, _ := json.Marshal(q)
+		ans, askErr := ch.AskUser(ctx, string(payload))
+		if errors.Is(askErr, ErrAgentBlocked) {
+			return "", true, nil
+		}
+		if askErr != nil {
+			return "", false, askErr
+		}
+		return ans, false, nil
+
+	case "report_summary":
+		var a struct {
+			Summary string `json:"summary"`
+		}
+		_ = json.Unmarshal([]byte(argsJSON), &a)
+		if rsErr := ch.ReportSummary(ctx, a.Summary); rsErr != nil {
+			return "", false, rsErr
+		}
+		return "Summary recorded.", false, nil
+
+	case "spawn_subtask":
+		var a struct {
+			Name         string  `json:"name"`
+			Instructions string  `json:"instructions"`
+			Model        string  `json:"model"`
+			MaxBudgetUSD float64 `json:"max_budget_usd"`
+		}
+		_ = json.Unmarshal([]byte(argsJSON), &a)
+		id, ssErr := ch.SpawnSubtask(ctx, SubtaskSpec{
+			Name:         a.Name,
+			Instructions: a.Instructions,
+			Model:        a.Model,
+			MaxBudgetUSD: a.MaxBudgetUSD,
+		})
+		if ssErr != nil {
+			return "", false, ssErr
+		}
+		return "Created subtask " + id, false, nil
+
+	case "record_progress":
+		var a struct {
+			Message string `json:"message"`
+		}
+		_ = json.Unmarshal([]byte(argsJSON), &a)
+		if rpErr := ch.RecordProgress(ctx, a.Message); rpErr != nil {
+			return "", false, rpErr
+		}
+		return "Noted.", false, nil
+
+	default:
+		return "", false, fmt.Errorf("unknown tool %q", name)
+	}
+}