Capstone: агентная платформа на Go от каркаса до прода (гл. 1–15)

package platform

import (
	"context"
	"log/slog"
	"sync"

	"github.com/anthropics/anthropic-sdk-go"
)

// Task — единица работы. ID нужен для идемпотентности и durable-чекпойнтов.
type Task struct {
	ID      string
	AgentID string // ключ в реестре агентов (control plane)
	Input   string
}

// Platform — корень сервиса. Держит клиента модели, реестр агентов,
// очередь задач и пул воркеров. Поля-теги json здесь опущены намеренно:
// в учебном коде конфиг грузится отдельным слоем (см. гл.14).
type Platform struct {
	model    anthropic.Client
	agents   *AgentRegistry  // системные промпты + наборы инструментов
	tools    *ToolRouter     // локальные обработчики + MCP-клиенты (гл.8)
	store    DurableStore     // чекпойнты состояния (гл.1)
	tracer   Tracer           // трейсинг каждого шага (гл.9)
	tasks    chan Task        // очередь задач (data plane)
	wg       sync.WaitGroup
}

func New(opts ...Option) *Platform {
	p := &Platform{
		model: anthropic.NewClient(), // ANTHROPIC_API_KEY из окружения
		tasks: make(chan Task, 256),
	}
	for _, o := range opts {
		o(p)
	}
	return p
}

// Start поднимает n воркеров. Каждый воркер — goroutine, читающая из общей
// очереди. Конкурентность на каналах — идиоматичный Go (гл.13).
func (p *Platform) Start(ctx context.Context, n int) {
	for i := 0; i < n; i++ {
		p.wg.Add(1)
		go p.worker(ctx, i)
	}
}

func (p *Platform) worker(ctx context.Context, id int) {
	defer p.wg.Done()
	for {
		select {
		case <-ctx.Done():
			return // graceful shutdown: ctx отменён — воркер выходит
		case task, ok := <-p.tasks:
			if !ok {
				return
			}
			if err := p.runTask(ctx, task); err != nil {
				slog.Error("задача провалена", "task", task.ID, "worker", id, "err", err)
			}
		}
	}
}

// Submit ставит задачу в очередь (неблокирующе при наличии места в буфере).
func (p *Platform) Submit(task Task) { p.tasks <- task }

// Shutdown закрывает приём задач и ждёт, пока воркеры доработают текущие.
// Незавершённые durable-задачи переподнимутся при следующем старте.
func (p *Platform) Shutdown() {
	close(p.tasks)
	p.wg.Wait()
}

// runTask исполняет одну задачу как durable agent loop. Цикл завершается
// СТРОГО по stop_reason (гл.1), а maxSteps/maxTokens — лишь страховка.
func (p *Platform) runTask(ctx context.Context, task Task) error {
	agent := p.agents.Get(task.AgentID)

	// Durable: пробуем восстановить состояние с последнего чекпойнта.
	// Если воркер падал — продолжим с шага N, а не с нуля.
	state, err := p.store.Load(ctx, task.ID)
	if err != nil {
		return err
	}
	if state == nil {
		state = &RunState{Messages: []anthropic.MessageParam{
			anthropic.NewUserMessage(anthropic.NewTextBlock(task.Input)),
		}}
	}

	const maxSteps = 30
	for state.Step < maxSteps {
		// Бюджет токенов на задачу (гл.12): превысили — мягко останавливаемся.
		if state.TokensUsed > agent.TokenBudget {
			return p.store.Finish(ctx, task.ID, "token_budget_exceeded")
		}

		// Управление окном перед вызовом (гл.2): компакция старых
		// tool-результатов, удержание стабильного префикса ради кэша.
		state.Messages = p.compactWindow(state.Messages)

		// callModel инкапсулирует цепочку middleware: трейс → guardrail →
		// модель → guardrail → эвал-семпл. Ядро loop про них не знает.
		resp, err := p.callModel(ctx, agent, state.Messages)
		if err != nil {
			return err // ретраи/фолбэки уже внутри callModel (гл.10)
		}
		state.TokensUsed += int(resp.Usage.InputTokens + resp.Usage.OutputTokens)
		state.Messages = append(state.Messages, resp.ToParam())

		// Единственный корректный критерий завершения хода.
		if resp.StopReason != anthropic.StopReasonToolUse {
			return p.store.Finish(ctx, task.ID, "done")
		}

		// Исполняем запрошенные инструменты через роутер (локально или MCP).
		results := p.execTools(ctx, agent, resp.Content)
		state.Messages = append(state.Messages, anthropic.NewUserMessage(results...))

		state.Step++
		// Чекпойнт ПОСЛЕ каждого шага: падение между шагами не теряет прогресс.
		if err := p.store.Save(ctx, task.ID, state); err != nil {
			return err
		}
	}
	return p.store.Finish(ctx, task.ID, "max_steps")
}

package platform

import (
	"context"
	"encoding/json"

	"github.com/anthropics/anthropic-sdk-go"
	mcp "github.com/modelcontextprotocol/go-sdk/mcp"
)

// ToolRouter маршрутизирует tool_use: сначала ищет локальный обработчик,
// иначе — внешний MCP-сервер. Схемы MCP-инструментов кэшируются на старте.
type ToolRouter struct {
	local map[string]LocalHandler          // имя -> типизированный обработчик
	mcp   map[string]*mcp.ClientSession     // имя инструмента -> MCP-сессия
}

type LocalHandler func(ctx context.Context, rawInput json.RawMessage) (string, error)

// ConnectMCP подключается к MCP-серверу (Streamable HTTP) и регистрирует все
// его инструменты в роутере. Один сервер обслуживает все агенты платформы.
func (r *ToolRouter) ConnectMCP(ctx context.Context, name, url string) error {
	client := mcp.NewClient(&mcp.Implementation{Name: "agent-platform"}, nil)
	session, err := client.Connect(ctx, mcp.NewStreamableClientTransport(url, nil), nil)
	if err != nil {
		return err
	}
	tools, err := session.ListTools(ctx, nil)
	if err != nil {
		return err
	}
	for _, t := range tools.Tools {
		r.mcp[t.Name] = session // помечаем: этот инструмент обслуживает MCP
	}
	return nil
}

// Exec исполняет один tool_use. block — это anthropic.ToolUseBlock из ответа.
func (r *ToolRouter) Exec(ctx context.Context, name, id string, input json.RawMessage) anthropic.ContentBlockParamUnion {
	// 1) Локальный типизированный обработчик — горячий путь.
	if h, ok := r.local[name]; ok {
		out, err := h(ctx, input)
		if err != nil {
			return anthropic.NewToolResultBlock(id, err.Error(), true)
		}
		return anthropic.NewToolResultBlock(id, out, false)
	}
	// 2) Иначе — внешний MCP-сервер (гл.8).
	if session, ok := r.mcp[name]; ok {
		var args map[string]any
		_ = json.Unmarshal(input, &args)
		res, err := session.CallTool(ctx, &mcp.CallToolParams{Name: name, Arguments: args})
		if err != nil {
			return anthropic.NewToolResultBlock(id, err.Error(), true)
		}
		return anthropic.NewToolResultBlock(id, mcpText(res), res.IsError)
	}
	// 3) Незнакомый инструмент — это ошибка конфигурации, не паника.
	return anthropic.NewToolResultBlock(id, "unknown tool: "+name, true)
}

// mcpText сворачивает контент MCP-ответа в строку для tool_result.
func mcpText(res *mcp.CallToolResult) string {
	var b []byte
	for _, c := range res.Content {
		if t, ok := c.(*mcp.TextContent); ok {
			b = append(b, t.Text...)
		}
	}
	return string(b)
}

// callModel — единственная точка, дёргающая модель. Она оборачивает вызов
// цепочкой сквозных забот. Ядро agent loop про эту цепочку ничего не знает —
// это и есть «шов» платформы (chain of responsibility).
func (p *Platform) callModel(
	ctx context.Context,
	agent *Agent,
	messages []anthropic.MessageParam,
) (*anthropic.Message, error) {
	// 1. Трейс-спан на шаг (гл.9). Дочерние субагенты наследуют parent span.
	ctx, span := p.tracer.StartSpan(ctx, "model_request")
	defer span.End()

	// 2. Guardrail на ВХОДЕ (гл.10): детектор prompt injection, лимиты.
	//    Важно: отрабатывает ДО списания токенов.
	if err := agent.Guardrails.CheckInput(messages); err != nil {
		span.SetError(err)
		return nil, err
	}

	// 3. Вызов модели с ретраями/фолбэками (гл.10 + гл.12).
	resp, err := p.callWithFallback(ctx, agent, messages)
	if err != nil {
		span.SetError(err)
		return nil, err
	}
	span.SetUsage(resp.Usage) // usage в спан — основа учёта стоимости

	// 4. Guardrail на ВЫХОДЕ (гл.10): схема-валидация, фильтр PII, политики.
	if err := agent.Guardrails.CheckOutput(resp); err != nil {
		span.SetError(err)
		return nil, err
	}

	// 5. Эвал-семпл (гл.9): часть трафика уходит в офлайн-датасет для
	//    LLM-as-judge и ловли регрессий перед выкаткой новой версии промпта.
	p.evalSink.Sample(agent.ID, messages, resp)

	return resp, nil
}

// callWithFallback: каскад моделей + circuit breaker. При overloaded_error —
// ретрай с backoff; при устойчивом отказе — деградация к дешёвой модели.
func (p *Platform) callWithFallback(
	ctx context.Context,
	agent *Agent,
	messages []anthropic.MessageParam,
) (*anthropic.Message, error) {
	params := anthropic.MessageNewParams{
		Model:     agent.Model, // напр. anthropic.ModelClaudeOpus4_8
		MaxTokens: 4096,
		Messages:  messages,
		Tools:     agent.Tools,
		System: []anthropic.TextBlockParam{{
			Text:         agent.SystemPrompt,
			CacheControl: anthropic.NewCacheControlEphemeralParam(), // prompt caching (гл.12)
		}},
	}
	resp, err := p.model.Messages.New(ctx, params)
	if err != nil {
		var apiErr *anthropic.Error
		// Перегрузка/5xx — ретраебельно; SDK уже ретраит, здесь — деградация.
		if errorsAsOverloaded(err, &apiErr) && agent.Fallback != "" {
			params.Model = agent.Fallback // более дешёвая/доступная модель
			return p.model.Messages.New(ctx, params)
		}
		return nil, err
	}
	return resp, nil
}