From 916b31ee89e43e1b98c5df8414865212c63cb36c Mon Sep 17 00:00:00 2001 From: Levi Neely Date: Wed, 29 Jul 2026 19:17:00 +0200 Subject: [PATCH] session: rename exports for clean public API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Core interface → Session interface - NewAgentCore → New - AgentCoreConfig → Config - Config (agent JSON) → AgentConfig - types.go → session.go (defines Session interface) - session.go → history.go (History type, persistence) --- session/config.go | 8 +- session/core_test.go | 22 +- session/harness.go | 14 +- session/history.go | 696 +++++++++++++++++++++++++++++++++++ session/session.go | 851 +++++++++---------------------------------- session/types.go | 187 ---------- 6 files changed, 889 insertions(+), 889 deletions(-) create mode 100644 session/history.go delete mode 100644 session/types.go diff --git a/session/config.go b/session/config.go index b71823f..fc8f1db 100644 --- a/session/config.go +++ b/session/config.go @@ -58,7 +58,7 @@ func (p Prompt) MarshalJSON() ([]byte, error) { return json.Marshal(p.Value) } -type Config struct { +type AgentConfig struct { Hooks map[string]HookCmds `json:"hooks,omitempty"` Prompt Prompt `json:"prompt,omitempty"` Backend string `json:"backend,omitempty"` @@ -90,8 +90,8 @@ type Config struct { } // Load parses a Config from r. -func Load(r io.Reader) (*Config, error) { - var cfg Config +func Load(r io.Reader) (*AgentConfig, error) { + var cfg AgentConfig if err := json.NewDecoder(r).Decode(&cfg); err != nil { return nil, err } @@ -100,6 +100,6 @@ func Load(r io.Reader) (*Config, error) { // ToolsEnabled reports whether tool use is enabled. Defaults to true when // the field is omitted from the config. -func (c *Config) ToolsEnabled() bool { +func (c *AgentConfig) ToolsEnabled() bool { return c.Tools == nil || *c.Tools } diff --git a/session/core_test.go b/session/core_test.go index c2c55ca..59608b0 100644 --- a/session/core_test.go +++ b/session/core_test.go @@ -96,7 +96,7 @@ func newCore(t *testing.T, be backend.Backend, hooks Hooks) *harness { Hooks: hooks, Preamble: "test system prompt", } - c := NewAgentCore(AgentCoreConfig{ + c := New(Config{ Backend: be, AgentName: "test", AgentsDir: t.TempDir(), @@ -111,7 +111,7 @@ func newCore(t *testing.T, be backend.Backend, hooks Hooks) *harness { } // collectEvents runs Submit synchronously and returns all emitted events. -func collectEvents(ctx context.Context, c Core, input string) []Event { +func collectEvents(ctx context.Context, c Session, input string) []Event { var mu sync.Mutex var evs []Event sub := c.Bus().Subscribe("event", func(ev Event) { @@ -138,7 +138,7 @@ func byRole(evs []Event, role string) []string { } // waitState blocks until c.State() == want, failing after 2 s. -func waitState(t *testing.T, c Core, want string) { +func waitState(t *testing.T, c Session, want string) { t.Helper() ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() @@ -1779,7 +1779,7 @@ func TestSetSessionID_UpdatesPreamble(t *testing.T) { env := &Runtime{ Preamble: "session is " + oldID + " end", } - c := NewAgentCore(AgentCoreConfig{ + c := New(Config{ Backend: defaultBE(), AgentName: "test", AgentsDir: t.TempDir(), @@ -2280,7 +2280,7 @@ func newCoreWithExecServer(t *testing.T, srv *mockEnvServer) *harness { Preamble: "test system prompt", Dispatcher: d, } - c := NewAgentCore(AgentCoreConfig{ + c := New(Config{ Backend: defaultBE(), AgentName: "test", AgentsDir: t.TempDir(), @@ -2396,7 +2396,7 @@ func TestBuildRuntime_NilConfig(t *testing.T) { func TestBuildRuntime_PromptBecomesPreamble(t *testing.T) { setupCfgDir(t) d := tools.NewDispatcher() - cfg := &Config{Prompt: Prompt{Value: []string{"the prompt"}}} + cfg := &AgentConfig{Prompt: Prompt{Value: []string{"the prompt"}}} env := BuildRuntime(cfg, d, t.TempDir(), nil) if env.Preamble != "the prompt" { @@ -2408,7 +2408,7 @@ func TestBuildRuntime_HooksAndParams(t *testing.T) { setupCfgDir(t) d := tools.NewDispatcher() temp := 0.7 - cfg := &Config{ + cfg := &AgentConfig{ Hooks: map[string]HookCmds{"preTurn": {"echo hi"}}, MaxTokens: 512, Temperature: &temp, @@ -2492,7 +2492,7 @@ func TestBuildRuntime_ToolsDisabled(t *testing.T) { tools: []tools.ToolInfo{{Server: "s1", Name: "mytool", Description: "desc", InputSchema: json.RawMessage(`{}`)}}, } f := false - cfg := &Config{Tools: &f} + cfg := &AgentConfig{Tools: &f} env := BuildRuntime(cfg, d, t.TempDir(), nil) if len(env.Tools) != 0 { t.Errorf("expected no tools when disabled; got %+v", env.Tools) @@ -2502,7 +2502,7 @@ func TestBuildRuntime_ToolsDisabled(t *testing.T) { func TestBuildRuntime_PromptOnly(t *testing.T) { setupCfgDir(t) d := tools.NewDispatcher() - cfg := &Config{Prompt: Prompt{Value: []string{"only agent"}}} + cfg := &AgentConfig{Prompt: Prompt{Value: []string{"only agent"}}} env := BuildRuntime(cfg, d, t.TempDir(), nil) if env.Preamble != "only agent" { t.Errorf("preamble = %q; want %q", env.Preamble, "only agent") @@ -2512,7 +2512,7 @@ func TestBuildRuntime_PromptOnly(t *testing.T) { func TestBuildRuntime_ExecPrompt(t *testing.T) { setupCfgDir(t) d := tools.NewDispatcher() - cfg := &Config{Prompt: Prompt{ + cfg := &AgentConfig{Prompt: Prompt{ Value: []string{"echo hello", "echo 'You are a security auditor.'", "echo world"}, IsExec: true, }} @@ -2532,7 +2532,7 @@ func TestBuildRuntime_ExecPromptFileResolution(t *testing.T) { t.Setenv("OLLIE_PROMPTS_PATH", promptsDir) d := tools.NewDispatcher() - cfg := &Config{Prompt: Prompt{ + cfg := &AgentConfig{Prompt: Prompt{ Value: []string{"test-prompt", "echo extra"}, IsExec: true, }} diff --git a/session/harness.go b/session/harness.go index a0badea..79c2fcb 100644 --- a/session/harness.go +++ b/session/harness.go @@ -35,7 +35,7 @@ type toolClassifier func(name string) bool // env provides additional environment variables injected into prompt resolution // subprocesses (e.g. OLLIE_SESSION_ID=xxx). // The caller is responsible for registering all servers on d before calling this. -func BuildRuntime(cfg *Config, d tools.Dispatcher, cwd string, env []string, baseLayers ...string) *Runtime { +func BuildRuntime(cfg *AgentConfig, d tools.Dispatcher, cwd string, env []string, baseLayers ...string) *Runtime { var messages []string var allToolInfos []tools.ToolInfo @@ -269,8 +269,8 @@ type actionHandle struct { cancel context.CancelCauseFunc } -// AgentCoreConfig is the configuration for creating an agent. -type AgentCoreConfig struct { +// Config is the configuration for creating an agent. +type Config struct { Backend backend.Backend ModelName string // if non-empty, overrides backend's default model AgentName string @@ -392,7 +392,7 @@ func (a *harness) pushLockDir() { } -var _ Core = (*harness)(nil) // compile-time interface check +var _ Session = (*harness)(nil) // compile-time interface check var sweepTmpOnce sync.Once @@ -435,8 +435,8 @@ func sweepStaleTmpDirs() { }) } -// NewAgentCore creates an agent from the given configuration. -func NewAgentCore(cfg AgentCoreConfig) Core { +// New creates an agent from the given configuration. +func New(cfg Config) Session { sweepStaleTmpDirs() if cfg.ModelName != "" { cfg.Backend.SetModel(cfg.ModelName) @@ -450,7 +450,7 @@ func NewAgentCore(cfg AgentCoreConfig) Core { } // Store the backend on the runtime so it's the single source of truth. rt.Backend = cfg.Backend - // A non-zero MaxSteps in AgentCoreConfig takes precedence over the + // A non-zero MaxSteps in Config takes precedence over the // value loaded from the agent JSON. if cfg.MaxSteps > 0 { rt.MaxSteps = cfg.MaxSteps diff --git a/session/history.go b/session/history.go new file mode 100644 index 0000000..8bd99f7 --- /dev/null +++ b/session/history.go @@ -0,0 +1,696 @@ +package session + +import ( + "context" + "encoding/json" + "fmt" + "os" + "slices" + "strings" + "time" + + "ollie/backend" +) + +const ( + compactionPrompt = `You are performing a CONTEXT CHECKPOINT COMPACTION. Produce a JSON object summarizing the current task state. Output ONLY valid JSON matching this schema: + +{ + "objective": "the user's overall goal", + "plan_step": "what step you are currently on", + "constraints": ["any constraints or user preferences"], + "last_action": "brief: last tool/action + outcome", + "open_questions": ["unresolved questions"], + "next_decision": "what to do next" +} + +Be concise. Capture what another LLM needs to seamlessly continue.` + + // hotTailSize is the number of most recent messages kept verbatim (hot zone). + hotTailSize = 8 + + // warmIndexSize is the number of messages preceding the hot zone + // summarized as a brief decision index (warm zone). + warmIndexSize = 10 +) + +type Reaction struct { + ID string `json:"id"` + ResponseID string `json:"responseId"` + Emoji string `json:"emoji"` + Category string `json:"category"` + CreatedAt time.Time `json:"createdAt"` +} + +// PersistedSession is the on-disk format for a saved session. +type PersistedSession struct { + ID string `json:"id"` + Agent string `json:"agent,omitempty"` + Backend string `json:"backend,omitempty"` + Model string `json:"model,omitempty"` + CWD string `json:"cwd,omitempty"` + Remote string `json:"remote,omitempty"` + Messages []backend.Message `json:"messages"` + TaskState *TaskState `json:"taskState,omitempty"` + // Usage and cost tracking, persisted across restarts. + TotalInputTokens int `json:"totalInputTokens,omitempty"` + TotalCachedInputTokens int `json:"totalCachedInputTokens,omitempty"` + TotalCacheCreationTokens int `json:"totalCacheCreationTokens,omitempty"` + TotalOutputTokens int `json:"totalOutputTokens,omitempty"` + TotalRequests int `json:"totalRequests,omitempty"` + Estimated bool `json:"estimated,omitempty"` + LastTurnCostUSD float64 `json:"lastTurnCostUSD,omitempty"` + SessionCostUSD float64 `json:"sessionCostUSD,omitempty"` + // Reaction tracking. + PositiveReactions int `json:"positiveReactions,omitempty"` + NegativeReactions int `json:"negativeReactions,omitempty"` + Reactions []Reaction `json:"reactions,omitempty"` +} + +// TaskState is a compact structured overlay that summarizes the agent's +// current position in a task. Injected at the top of every turn so the +// model doesn't rely on full history recall. +type TaskState struct { + Objective string `json:"objective"` + PlanStep string `json:"plan_step"` + Constraints []string `json:"constraints,omitempty"` + LastAction string `json:"last_action"` + OpenQuestions []string `json:"open_questions,omitempty"` + NextDecision string `json:"next_decision"` +} + +// render produces a compact text representation for injection into the message history. +func (ts *TaskState) render() string { + if ts.Objective == "" { + return "" + } + var sb strings.Builder + sb.WriteString("[task state]\n") + sb.WriteString("objective: " + ts.Objective + "\n") + if ts.PlanStep != "" { + sb.WriteString("plan_step: " + ts.PlanStep + "\n") + } + if len(ts.Constraints) > 0 { + sb.WriteString("constraints: " + strings.Join(ts.Constraints, "; ") + "\n") + } + if ts.LastAction != "" { + sb.WriteString("last_action: " + ts.LastAction + "\n") + } + if len(ts.OpenQuestions) > 0 { + sb.WriteString("open_questions: " + strings.Join(ts.OpenQuestions, "; ") + "\n") + } + if ts.NextDecision != "" { + sb.WriteString("next_decision: " + ts.NextDecision + "\n") + } + sb.WriteString("[/task state]") + return sb.String() +} + +// saveTo writes the full message history to path as JSON. +func (s *History) saveTo(path, id, agentName, cwd string) error { + return s.saveToFull(path, id, agentName, "", "", cwd, "") +} + +// saveToFull writes session state including backend/model info. +func sanitizeMessages(msgs []backend.Message) []backend.Message { + out := make([]backend.Message, 0, len(msgs)) + for _, m := range msgs { + for j, tc := range m.ToolCalls { + if len(tc.Arguments) > 0 { + var dummy json.RawMessage + if err := json.Unmarshal(tc.Arguments, &dummy); err != nil { + // Corrupted arguments — replace with empty object so save succeeds. + m.ToolCalls[j].Arguments = json.RawMessage("{}") + } + } + } + out = append(out, m) + } + return out +} + +func (s *History) saveToFull(path, id, agentName, backendName, modelName, cwd, remote string) error { + ps := PersistedSession{ + ID: id, + Agent: agentName, + Backend: backendName, + Model: modelName, + CWD: cwd, + Remote: remote, + Messages: sanitizeMessages(s.messages), + TaskState: s.TaskState, + TotalInputTokens: s.TotalInputTokens, + TotalCachedInputTokens: s.TotalCachedInputTokens, + TotalCacheCreationTokens: s.TotalCacheCreationTokens, + TotalOutputTokens: s.TotalOutputTokens, + TotalRequests: s.TotalRequests, + Estimated: s.Estimated, + LastTurnCostUSD: s.LastTurnCostUSD, + SessionCostUSD: s.SessionCostUSD, + PositiveReactions: s.PositiveReactions, + NegativeReactions: s.NegativeReactions, + Reactions: s.Reactions, + } + data, err := json.Marshal(ps) + if err != nil { + return fmt.Errorf("session save: %w", err) + } + // Atomic write: write to temp file, then rename to preserve last good snapshot. + tmpPath := path + ".tmp" + if err := os.WriteFile(tmpPath, data, 0600); err != nil { + return fmt.Errorf("session save tmp: %w", err) + } + return os.Rename(tmpPath, path) +} + +// LoadPersistedSession reads a PersistedSession from a JSON file. +func LoadPersistedSession(path string) (*PersistedSession, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, err + } + var ps PersistedSession + if err := json.Unmarshal(data, &ps); err != nil { + return nil, fmt.Errorf("parse session %s: %w", path, err) + } + return &ps, nil +} + +// RestoreHistory reconstructs a Session from a persisted message list. +func RestoreHistory(ps *PersistedSession) *History { + s := &History{ + messages: ps.Messages, + TaskState: ps.TaskState, + TotalInputTokens: ps.TotalInputTokens, + TotalCachedInputTokens: ps.TotalCachedInputTokens, + TotalCacheCreationTokens: ps.TotalCacheCreationTokens, + TotalOutputTokens: ps.TotalOutputTokens, + TotalRequests: ps.TotalRequests, + Estimated: ps.Estimated, + LastTurnCostUSD: ps.LastTurnCostUSD, + SessionCostUSD: ps.SessionCostUSD, + PositiveReactions: ps.PositiveReactions, + NegativeReactions: ps.NegativeReactions, + Reactions: ps.Reactions, + } + for i := range s.messages { + if s.messages[i].Role == "assistant" && s.messages[i].ID == "" { + s.messages[i].ID = NewResponseID() + } + } + for _, m := range ps.Messages { + if m.Role == "user" { + s.goal = m.Content + break + } + } + return s +} + +// History is an ephemeral in-memory state backend. +type History struct { + goal string + messages []backend.Message + TaskState *TaskState + // Cumulative usage tracking. + TotalInputTokens int + TotalCachedInputTokens int + TotalCacheCreationTokens int + TotalOutputTokens int + TotalRequests int + Estimated bool // true if any usage was estimated rather than reported by the backend + LastTurnCostUSD float64 // cost of the most recently completed turn in USD + SessionCostUSD float64 // cumulative cost of all turns in this session in USD + // per-turn accumulators; reset at the start of each turn + turnInputTokens int + turnCachedTokens int + turnCreationTokens int + turnOutputTokens int + turnCostUSD float64 // >0 when backend reported cost directly (e.g. OpenRouter) + // Reaction tracking. + PositiveReactions int + NegativeReactions int + Reactions []Reaction +} + +// newHistory creates a new empty Session. The caller is responsible for +// appending the initial user message via appendUserMessage. +func newHistory(goal string) *History { + return &History{goal: goal} +} + +// Checkpoint forks the session: returns a new Session that inherits the given +// TaskState but starts with a clean message history. This enables narrow-context +// sub-agents that know what to do without inheriting all parent message noise. +func (s *History) Checkpoint(ts TaskState) *History { + child := &History{ + goal: ts.Objective, + TaskState: &ts, + } + // Seed with a user message so the child has a valid initial turn. + child.messages = []backend.Message{{ + Role: "user", + Content: ts.render(), + }} + return child +} + +func (s *History) history() []backend.Message { + if len(s.Reactions) == 0 { + return s.messages + } + // Append feedback after the conversation so assistant tool-call messages stay + // adjacent to their tool results, as required by provider APIs. + out := make([]backend.Message, 0, len(s.messages)+len(s.Reactions)) + out = append(out, s.messages...) + for _, r := range s.Reactions { + _, desc, _ := classifyReaction(r.Emoji) + out = append(out, backend.Message{Role: "user", Content: "[reaction to assistant response " + r.ResponseID + ": " + r.Category + " (" + r.Emoji + ")] " + desc}) + } + return out +} + +func (s *History) taskState() *TaskState { + return s.TaskState +} + +func (s *History) updateTaskState(ts TaskState) { + s.TaskState = &ts +} + +func (s *History) addUsage(u backend.Usage, estimated bool) { + s.TotalInputTokens += u.InputTokens + s.TotalCachedInputTokens += u.CachedInputTokens + s.TotalCacheCreationTokens += u.CacheCreationTokens + s.TotalOutputTokens += u.OutputTokens + s.TotalRequests++ + s.turnInputTokens += u.InputTokens + s.turnCachedTokens += u.CachedInputTokens + s.turnCreationTokens += u.CacheCreationTokens + s.turnOutputTokens += u.OutputTokens + s.turnCostUSD += u.CostUSD + if estimated { + s.Estimated = true + } +} + +func (s *History) recomputeReactionCounts() { + s.PositiveReactions = 0 + s.NegativeReactions = 0 + for _, r := range s.Reactions { + switch r.Category { + case "positive", "excellent": + s.PositiveReactions++ + case "negative", "terrible": + s.NegativeReactions++ + } + } +} + +func (s *History) resetTurnAccumulators() { + s.turnInputTokens = 0 + s.turnCachedTokens = 0 + s.turnCreationTokens = 0 + s.turnOutputTokens = 0 + s.turnCostUSD = 0 +} + +// recordTurnCost computes and stores the last turn's cost, adding it to the +// session total. model is used for the pricing-table fallback when the backend +// did not report cost directly. +func (s *History) recordTurnCost(model string) { + var cost float64 + if s.turnCostUSD > 0 { + cost = s.turnCostUSD + } else { + cost = computeCostUSD(model, backend.Usage{ + InputTokens: s.turnInputTokens, + CachedInputTokens: s.turnCachedTokens, + CacheCreationTokens: s.turnCreationTokens, + OutputTokens: s.turnOutputTokens, + }) + } + s.LastTurnCostUSD = cost + s.SessionCostUSD += cost +} + +func (s *History) update(assistant backend.Message, results []toolResult) { + s.messages = append(s.messages, assistant) + for _, r := range results { + s.messages = append(s.messages, backend.Message{ + Role: "tool", + Content: r.Content, + ContentBlocks: r.ContentBlocks, + ToolCallID: r.ToolCallID, + }) + } +} + +// llmSummarizeToolResult uses the LLM to produce a concise summary of a tool result. +func llmSummarizeToolResult(ctx context.Context, b backend.Backend, name, content string) string { + prompt := fmt.Sprintf("Summarize this tool result from %q in 1-3 sentences, preserving key facts, file paths, values, and findings:\n\n%s", name, content) + ch, err := b.ChatStream(ctx, []backend.Message{{Role: "user", Content: prompt}}, nil, backend.GenerationParams{}) + if err != nil { + return content + } + var sb strings.Builder + for ev := range ch { + if ev.Content != "" { + sb.WriteString(ev.Content) + } + if ev.Done { + break + } + } + if s := strings.TrimSpace(sb.String()); s != "" { + return fmt.Sprintf("[summary of %s result]: %s", name, s) + } + return content +} + +// removeCancelledToolResults filters out tool results that were cancelled due +// to interrupt, keeping completed work. Also removes the corresponding tool +// calls from assistant messages to maintain a valid message sequence. +func (s *History) removeCancelledToolResults() { + // First pass: collect cancelled tool call IDs + cancelled := make(map[string]bool) + for _, m := range s.messages { + if m.Role == "tool" && isCancelledToolResult(m.Content) { + cancelled[m.ToolCallID] = true + } + } + if len(cancelled) == 0 { + return + } + + // Second pass: filter messages and prune tool calls from assistant messages + filtered := s.messages[:0] + for _, m := range s.messages { + if m.Role == "tool" && cancelled[m.ToolCallID] { + continue + } + if m.Role == "assistant" && len(m.ToolCalls) > 0 { + // Remove cancelled tool calls from this assistant message + kept := m.ToolCalls[:0] + for _, tc := range m.ToolCalls { + if !cancelled[tc.ID] { + kept = append(kept, tc) + } + } + if len(kept) == 0 && m.Content == "" { + // All tool calls cancelled and no text content - skip message + continue + } + m.ToolCalls = kept + } + filtered = append(filtered, m) + } + s.messages = filtered +} + +func isCancelledToolResult(content string) bool { + return strings.Contains(content, `"status":"cancelled"`) || + strings.Contains(content, "tool execution interrupted by user") +} + +// PreCompactionSnapshot returns a copy of the current messages for persistence +// before compaction. Call this before compact(). +func (s *History) PreCompactionSnapshot() []backend.Message { + return slices.Clone(s.messages) +} + +// Compact summarizes the conversation via an LLM call, replacing the history +// with system messages + preserved user messages + a structured summary. +// Returns (n compacted, summary text, error); n==0 means nothing to compact. +func (s *History) compact(ctx context.Context, b backend.Backend) (int, string, error) { + if len(s.messages) <= hotTailSize+warmIndexSize { + return 0, "", nil + } + + // Flatten tool calls into plain text so the compaction request + // doesn't need tool schemas. + flattened := flattenToolMessages(s.messages) + flattened = append(flattened, backend.Message{ + Role: "user", + Content: compactionPrompt, + }) + + ch, err := b.ChatStream(ctx, flattened, nil, backend.GenerationParams{}) + if err != nil { + return 0, "", fmt.Errorf("compact: %w", err) + } + + var summary strings.Builder + for ev := range ch { + if ev.Content != "" { + summary.WriteString(ev.Content) + } + if ev.Done { + break + } + } + + summaryText := strings.TrimSpace(summary.String()) + if summaryText == "" { + return 0, "", fmt.Errorf("compact: empty summary") + } + + // Parse structured task state from the model's JSON response. + var ts TaskState + raw := extractJSON(summaryText) + if err := json.Unmarshal([]byte(raw), &ts); err != nil { + // Fallback: treat as unstructured summary if JSON parse fails. + ts = TaskState{Objective: summaryText} + } + s.TaskState = &ts + + beforeCount := len(s.messages) + s.messages = buildCompactedHistory(ts, s.messages) + return beforeCount - len(s.messages), summaryText, nil +} + +// buildCompactedHistory constructs a three-zone post-compaction message list: +// - Cold: structured task state summary (everything older than warm+hot) +// - Warm: brief index of recent decisions/actions (warmIndexSize messages) +// - Hot: last hotTailSize messages verbatim +func buildCompactedHistory(ts TaskState, allMessages []backend.Message) []backend.Message { + total := len(allMessages) + + // Determine zone boundaries. + hotStart := total - hotTailSize + if hotStart < 0 { + hotStart = 0 + } + + // If hotStart lands on a tool message, walk backward to include the + // preceding assistant message with tool_calls. Without this, the hot + // zone starts with an orphaned tool result which violates the OpenAI + // API constraint that tool messages must follow an assistant message + // containing the corresponding tool_calls. + for hotStart > 0 && allMessages[hotStart].Role == "tool" { + hotStart-- + } + + warmStart := hotStart - warmIndexSize + if warmStart < 0 { + warmStart = 0 + } + + // Cold zone: structured task state. + stateJSON, _ := json.Marshal(ts) + cold := backend.Message{ + Role: "user", + Content: "[compacted context — cold zone]\n```json\n" + string(stateJSON) + "\n```", + } + + var out []backend.Message + out = append(out, cold) + + // Warm zone: one-line summaries of decisions in the warm window. + warmSlice := allMessages[warmStart:hotStart] + if len(warmSlice) > 0 { + var sb strings.Builder + sb.WriteString("[warm zone — recent decisions]\n") + for _, m := range warmSlice { + line := summarizeMessage(m) + if line != "" { + sb.WriteString("- ") + sb.WriteString(line) + sb.WriteByte('\n') + } + } + out = append(out, backend.Message{Role: "user", Content: sb.String()}) + } + + // Hot zone: verbatim recent messages. + out = append(out, allMessages[hotStart:]...) + return out +} + +// summarizeMessage produces a one-line summary of a message for the warm index. +func summarizeMessage(m backend.Message) string { + switch m.Role { + case "assistant": + if len(m.ToolCalls) > 0 { + names := make([]string, len(m.ToolCalls)) + for i, tc := range m.ToolCalls { + names[i] = tc.Name + } + return "called: " + strings.Join(names, ", ") + } + text := m.Content + if len(text) > 120 { + text = text[:120] + "…" + } + return "said: " + text + case "user": + text := m.Content + if len(text) > 120 { + text = text[:120] + "…" + } + return "user: " + text + case "tool": + text := m.Content + if len(text) > 80 { + text = text[:80] + "…" + } + return "result(" + m.ToolCallID + "): " + text + } + return "" +} + +// extractJSON finds the first JSON object in s (handling markdown fences). +func extractJSON(s string) string { + // Strip markdown code fence if present. + if i := strings.Index(s, "```json"); i >= 0 { + s = s[i+7:] + if j := strings.Index(s, "```"); j >= 0 { + s = s[:j] + } + } else if i := strings.Index(s, "```"); i >= 0 { + s = s[i+3:] + if j := strings.Index(s, "```"); j >= 0 { + s = s[:j] + } + } + // Find first { ... } + start := strings.Index(s, "{") + if start < 0 { + return s + } + end := strings.LastIndex(s, "}") + if end < start { + return s + } + return s[start : end+1] +} + +// flattenToolMessages converts tool call/result sequences into plain text +// so the compaction request doesn't include tool-specific structures that +// the API may reject when no tools are defined. +func flattenToolMessages(messages []backend.Message) []backend.Message { + out := make([]backend.Message, 0, len(messages)) + for _, m := range messages { + switch { + case m.Role == "assistant" && len(m.ToolCalls) > 0: + var sb strings.Builder + if m.Content != "" { + sb.WriteString(m.Content) + sb.WriteString("\n\n") + } + for _, tc := range m.ToolCalls { + fmt.Fprintf(&sb, "[Tool call: %s(%s)]\n", tc.Name, string(tc.Arguments)) + } + out = append(out, backend.Message{Role: "assistant", Content: sb.String()}) + case m.Role == "tool": + text := m.Content + if len(text) > 4000 { + text = text[:4000] + "..." + } + out = append(out, backend.Message{ + Role: "user", + Content: fmt.Sprintf("[Tool result for %s]:\n%s", m.ToolCallID, text), + }) + default: + out = append(out, m) + } + } + return out +} + +func (s *History) appendUserMessage(content string) { + s.messages = append(s.messages, backend.Message{Role: "user", Content: content}) +} + +// cloneMessages returns a deep copy of the message slice. +func cloneMessages(msgs []backend.Message) []backend.Message { + out := make([]backend.Message, len(msgs)) + for i, m := range msgs { + out[i] = backend.Message{ + Role: m.Role, + Content: m.Content, + ToolCallID: m.ToolCallID, + } + if len(m.ToolCalls) > 0 { + out[i].ToolCalls = make([]backend.ToolCall, len(m.ToolCalls)) + for j, tc := range m.ToolCalls { + out[i].ToolCalls[j] = backend.ToolCall{ + ID: tc.ID, + Name: tc.Name, + Arguments: append(json.RawMessage(nil), tc.Arguments...), + } + } + } + } + return out +} + +// estimateTokens returns a rough token count (~4 chars per token). +func (s *History) estimateTokens() int { + chars := 0 + for _, m := range s.messages { + chars += len(m.Content) + for _, tc := range m.ToolCalls { + chars += len(tc.Name) + len(tc.Arguments) + } + } + return chars / 4 +} + +// stripColdResults summarizes large tool-result messages outside the hot tail +// using an LLM call. Messages in the last hotTailSize slots are left verbatim. +func (s *History) stripColdResults(ctx context.Context, b backend.Backend) { + hot := len(s.messages) - hotTailSize + for i := range s.messages { + if i >= hot { + break + } + m := &s.messages[i] + if m.Role == "tool" && len(m.Content) > 200 { + m.Content = llmSummarizeToolResult(ctx, b, m.ToolCallID, m.Content) + } + } +} + +// stripCold implements the state interface. +func (s *History) stripCold(ctx context.Context, b backend.Backend) { + s.stripColdResults(ctx, b) +} + +// contextDebug returns a multi-line breakdown of the history. +func (s *History) contextDebug() string { + var sb strings.Builder + sb.WriteString(fmt.Sprintf("=== %d messages ===\n", len(s.messages))) + for i, m := range s.messages { + preview := m.Content + if len(preview) > 80 { + preview = preview[:80] + "..." + } + chars := len(m.Content) + for _, tc := range m.ToolCalls { + chars += len(tc.Name) + len(tc.Arguments) + } + sb.WriteString(fmt.Sprintf(" [%d] role=%-10s chars=%-6d %q\n", i, m.Role, chars, preview)) + } + return sb.String() +} diff --git a/session/session.go b/session/session.go index 8bd99f7..81411a0 100644 --- a/session/session.go +++ b/session/session.go @@ -2,695 +2,186 @@ package session import ( "context" - "encoding/json" - "fmt" - "os" - "slices" - "strings" - "time" + "errors" + "github.com/simonfxr/pubsub" "ollie/backend" ) +// WatchField names supported by Core.WaitChange. const ( - compactionPrompt = `You are performing a CONTEXT CHECKPOINT COMPACTION. Produce a JSON object summarizing the current task state. Output ONLY valid JSON matching this schema: - -{ - "objective": "the user's overall goal", - "plan_step": "what step you are currently on", - "constraints": ["any constraints or user preferences"], - "last_action": "brief: last tool/action + outcome", - "open_questions": ["unresolved questions"], - "next_decision": "what to do next" -} - -Be concise. Capture what another LLM needs to seamlessly continue.` - - // hotTailSize is the number of most recent messages kept verbatim (hot zone). - hotTailSize = 8 - - // warmIndexSize is the number of messages preceding the hot zone - // summarized as a brief decision index (warm zone). - warmIndexSize = 10 + WatchState = "state" + WatchUsage = "usage" + WatchCtxSz = "ctxsz" + WatchCWD = "cwd" + WatchAgent = "agent" ) -type Reaction struct { - ID string `json:"id"` - ResponseID string `json:"responseId"` - Emoji string `json:"emoji"` - Category string `json:"category"` - CreatedAt time.Time `json:"createdAt"` +// ErrInterrupted is returned when the user cancels an agent turn (Ctrl-C). +var ErrInterrupted = errors.New("interrupted") + +// Event is a typed output event emitted during an agent turn or in response +// to a command. +type Event struct { + Role string + Name string + Content string + ResponseID string } -// PersistedSession is the on-disk format for a saved session. -type PersistedSession struct { - ID string `json:"id"` - Agent string `json:"agent,omitempty"` - Backend string `json:"backend,omitempty"` - Model string `json:"model,omitempty"` - CWD string `json:"cwd,omitempty"` - Remote string `json:"remote,omitempty"` - Messages []backend.Message `json:"messages"` - TaskState *TaskState `json:"taskState,omitempty"` - // Usage and cost tracking, persisted across restarts. - TotalInputTokens int `json:"totalInputTokens,omitempty"` - TotalCachedInputTokens int `json:"totalCachedInputTokens,omitempty"` - TotalCacheCreationTokens int `json:"totalCacheCreationTokens,omitempty"` - TotalOutputTokens int `json:"totalOutputTokens,omitempty"` - TotalRequests int `json:"totalRequests,omitempty"` - Estimated bool `json:"estimated,omitempty"` - LastTurnCostUSD float64 `json:"lastTurnCostUSD,omitempty"` - SessionCostUSD float64 `json:"sessionCostUSD,omitempty"` - // Reaction tracking. - PositiveReactions int `json:"positiveReactions,omitempty"` - NegativeReactions int `json:"negativeReactions,omitempty"` - Reactions []Reaction `json:"reactions,omitempty"` +// EventHandler receives events from the agent. +type EventHandler func(Event) + +// Session is the interface between a frontend (TUI, HTTP handler, etc.) and the +// agent engine. All output from the agent is delivered via the event bus. +type Session interface { + // Submit processes one line of user input. Slash commands and shell + // shortcuts are dispatched synchronously; any other input starts an agent + // turn that publishes events to the bus until the turn is complete. + // After the turn, any queued prompts are drained sequentially. + Submit(ctx context.Context, input string) + + // Interrupt cancels the current in-progress agent turn. + // Returns true if an action was running and was cancelled. + Interrupt(cause error) bool + + // Inject sends a message that will be appended to the next tool result + // as a user interruption. If no turn is running, it is silently dropped. + Inject(prompt string) + + // Queue pushes a prompt onto the FIFO for execution after the current + // turn completes. + Queue(prompt string) + + // Bus returns the session event bus. + Bus() *pubsub.Bus + + // PopQueue removes and returns the next queued prompt. + // Returns ("", false) if the queue is empty. + PopQueue() (string, bool) + + // IsRunning returns true if an agent turn is currently in progress. + IsRunning() bool + + // State returns the current agent state: "idle", "thinking", or "calling: ". + State() string + + // Reply returns the assistant text from the most recently completed turn. + // Cleared when a new prompt is submitted. + Reply() string + + // AgentName returns the name of the active agent. + AgentName() string + + // BackendName returns the name of the active backend (e.g. "anthropic", "ollama"). + BackendName() string + + // ModelName returns the name of the active model. + ModelName() string + + // CtxSz returns the estimated context size as a one-line summary. + CtxSz() string + + // Usage returns billed token counts as a one-line summary. + Usage() string + + // Cost returns per-turn and session cost as a two-line key=value summary. + Cost() string + + // ListModels returns available model names, one per line. + ListModels() string + + // CWD returns the current working directory used for tool execution. + CWD() string + + // SetCWD changes the working directory for tool execution and + // updates the system prompt. Returns an error if the path does not exist. + SetCWD(dir string) error + + // SetSessionID renames the session: updates the in-memory ID, renames + // persisted files on disk, and propagates to the execute server env. + SetSessionID(newID string) error + + // Context returns the current message history as it would be sent to the + // backend: system prompt prepended, stale reads pruned. Does not include + // tool definitions (see Tools) or generation params (see GenerationParams). + Context() []backend.Message + + // SystemPrompt returns the fully rendered system prompt for this session. + SystemPrompt() string + + // GenerationParams returns the current sampling parameters. + GenerationParams() backend.GenerationParams + + // SetGenerationParams replaces the current sampling parameters. + // Returns an error if the agent is currently running. + SetGenerationParams(params backend.GenerationParams) error + + // CompactionModel returns the model override used for context compaction. + CompactionModel() string + + // SetCompactionModel changes the model used for context compaction. + SetCompactionModel(model string) + + // SetEnv injects a session-scoped environment variable into the shell + // subprocesses. Does not affect the daemon process environment. + SetEnv(key, value string) + + // WaitChange blocks until the named field changes from current, then returns + // the new value. Returns ("", false) if ctx is cancelled before a change. + // Supported fields: WatchState, WatchUsage, WatchCtxSz, WatchCWD. + WaitChange(ctx context.Context, field, current string) (string, bool) + + // ToolCallCount returns the total number of tool calls executed in this + // session since the agent was created. The counter is monotonically + // increasing and never resets. Blocked calls (pre-tool hook exit 2) are + // not counted. Timed-out or erroring calls are counted because execution + // was attempted. Use modulo arithmetic in hooks to fire every N calls: + // [ $(($(cat tcct) % 10)) -eq 0 ] && ... + ToolCallCount() int64 + + // SaveSession writes the current session state to the given path. + // The file includes all messages, agent/backend/model metadata, and task state. + SaveSession(path string) error + + // Close releases resources associated with the session, including its + // temporary directory under /tmp/ollie/. + Close() + + // Detach detaches the currently running process from the agent. + // The process continues running; returns false if nothing is executing. + Detach() bool + + // ListDetached returns info about all detached processes. + ListDetached() []DetachedInfo + + // SignalDetached sends a signal to a detached process by PID. + SignalDetached(pid, signal int) error + + // GetDetachedOutput returns the ring buffer output for a detached process. + GetDetachedOutput(pid int) (string, error) + + // DismissDetached removes an exited process from the list. + DismissDetached(pid int) bool + + // InjectSystemEvent appends a system-originated message to the session + // context and emits it on the event bus. The agent sees it on its next + // turn; it appears in the chat log immediately. + InjectSystemEvent(content string) + + // React records an emoji reaction to the most recent assistant response. + React(emoji string) + // ReactTo records an emoji reaction to a specific assistant response. + ReactTo(responseID, emoji string) error + // Reactions returns the current response ID to emoji mapping. + Reactions() map[string]string } -// TaskState is a compact structured overlay that summarizes the agent's -// current position in a task. Injected at the top of every turn so the -// model doesn't rely on full history recall. -type TaskState struct { - Objective string `json:"objective"` - PlanStep string `json:"plan_step"` - Constraints []string `json:"constraints,omitempty"` - LastAction string `json:"last_action"` - OpenQuestions []string `json:"open_questions,omitempty"` - NextDecision string `json:"next_decision"` -} - -// render produces a compact text representation for injection into the message history. -func (ts *TaskState) render() string { - if ts.Objective == "" { - return "" - } - var sb strings.Builder - sb.WriteString("[task state]\n") - sb.WriteString("objective: " + ts.Objective + "\n") - if ts.PlanStep != "" { - sb.WriteString("plan_step: " + ts.PlanStep + "\n") - } - if len(ts.Constraints) > 0 { - sb.WriteString("constraints: " + strings.Join(ts.Constraints, "; ") + "\n") - } - if ts.LastAction != "" { - sb.WriteString("last_action: " + ts.LastAction + "\n") - } - if len(ts.OpenQuestions) > 0 { - sb.WriteString("open_questions: " + strings.Join(ts.OpenQuestions, "; ") + "\n") - } - if ts.NextDecision != "" { - sb.WriteString("next_decision: " + ts.NextDecision + "\n") - } - sb.WriteString("[/task state]") - return sb.String() -} - -// saveTo writes the full message history to path as JSON. -func (s *History) saveTo(path, id, agentName, cwd string) error { - return s.saveToFull(path, id, agentName, "", "", cwd, "") -} - -// saveToFull writes session state including backend/model info. -func sanitizeMessages(msgs []backend.Message) []backend.Message { - out := make([]backend.Message, 0, len(msgs)) - for _, m := range msgs { - for j, tc := range m.ToolCalls { - if len(tc.Arguments) > 0 { - var dummy json.RawMessage - if err := json.Unmarshal(tc.Arguments, &dummy); err != nil { - // Corrupted arguments — replace with empty object so save succeeds. - m.ToolCalls[j].Arguments = json.RawMessage("{}") - } - } - } - out = append(out, m) - } - return out -} - -func (s *History) saveToFull(path, id, agentName, backendName, modelName, cwd, remote string) error { - ps := PersistedSession{ - ID: id, - Agent: agentName, - Backend: backendName, - Model: modelName, - CWD: cwd, - Remote: remote, - Messages: sanitizeMessages(s.messages), - TaskState: s.TaskState, - TotalInputTokens: s.TotalInputTokens, - TotalCachedInputTokens: s.TotalCachedInputTokens, - TotalCacheCreationTokens: s.TotalCacheCreationTokens, - TotalOutputTokens: s.TotalOutputTokens, - TotalRequests: s.TotalRequests, - Estimated: s.Estimated, - LastTurnCostUSD: s.LastTurnCostUSD, - SessionCostUSD: s.SessionCostUSD, - PositiveReactions: s.PositiveReactions, - NegativeReactions: s.NegativeReactions, - Reactions: s.Reactions, - } - data, err := json.Marshal(ps) - if err != nil { - return fmt.Errorf("session save: %w", err) - } - // Atomic write: write to temp file, then rename to preserve last good snapshot. - tmpPath := path + ".tmp" - if err := os.WriteFile(tmpPath, data, 0600); err != nil { - return fmt.Errorf("session save tmp: %w", err) - } - return os.Rename(tmpPath, path) -} - -// LoadPersistedSession reads a PersistedSession from a JSON file. -func LoadPersistedSession(path string) (*PersistedSession, error) { - data, err := os.ReadFile(path) - if err != nil { - return nil, err - } - var ps PersistedSession - if err := json.Unmarshal(data, &ps); err != nil { - return nil, fmt.Errorf("parse session %s: %w", path, err) - } - return &ps, nil -} - -// RestoreHistory reconstructs a Session from a persisted message list. -func RestoreHistory(ps *PersistedSession) *History { - s := &History{ - messages: ps.Messages, - TaskState: ps.TaskState, - TotalInputTokens: ps.TotalInputTokens, - TotalCachedInputTokens: ps.TotalCachedInputTokens, - TotalCacheCreationTokens: ps.TotalCacheCreationTokens, - TotalOutputTokens: ps.TotalOutputTokens, - TotalRequests: ps.TotalRequests, - Estimated: ps.Estimated, - LastTurnCostUSD: ps.LastTurnCostUSD, - SessionCostUSD: ps.SessionCostUSD, - PositiveReactions: ps.PositiveReactions, - NegativeReactions: ps.NegativeReactions, - Reactions: ps.Reactions, - } - for i := range s.messages { - if s.messages[i].Role == "assistant" && s.messages[i].ID == "" { - s.messages[i].ID = NewResponseID() - } - } - for _, m := range ps.Messages { - if m.Role == "user" { - s.goal = m.Content - break - } - } - return s -} - -// History is an ephemeral in-memory state backend. -type History struct { - goal string - messages []backend.Message - TaskState *TaskState - // Cumulative usage tracking. - TotalInputTokens int - TotalCachedInputTokens int - TotalCacheCreationTokens int - TotalOutputTokens int - TotalRequests int - Estimated bool // true if any usage was estimated rather than reported by the backend - LastTurnCostUSD float64 // cost of the most recently completed turn in USD - SessionCostUSD float64 // cumulative cost of all turns in this session in USD - // per-turn accumulators; reset at the start of each turn - turnInputTokens int - turnCachedTokens int - turnCreationTokens int - turnOutputTokens int - turnCostUSD float64 // >0 when backend reported cost directly (e.g. OpenRouter) - // Reaction tracking. - PositiveReactions int - NegativeReactions int - Reactions []Reaction -} - -// newHistory creates a new empty Session. The caller is responsible for -// appending the initial user message via appendUserMessage. -func newHistory(goal string) *History { - return &History{goal: goal} -} - -// Checkpoint forks the session: returns a new Session that inherits the given -// TaskState but starts with a clean message history. This enables narrow-context -// sub-agents that know what to do without inheriting all parent message noise. -func (s *History) Checkpoint(ts TaskState) *History { - child := &History{ - goal: ts.Objective, - TaskState: &ts, - } - // Seed with a user message so the child has a valid initial turn. - child.messages = []backend.Message{{ - Role: "user", - Content: ts.render(), - }} - return child -} - -func (s *History) history() []backend.Message { - if len(s.Reactions) == 0 { - return s.messages - } - // Append feedback after the conversation so assistant tool-call messages stay - // adjacent to their tool results, as required by provider APIs. - out := make([]backend.Message, 0, len(s.messages)+len(s.Reactions)) - out = append(out, s.messages...) - for _, r := range s.Reactions { - _, desc, _ := classifyReaction(r.Emoji) - out = append(out, backend.Message{Role: "user", Content: "[reaction to assistant response " + r.ResponseID + ": " + r.Category + " (" + r.Emoji + ")] " + desc}) - } - return out -} - -func (s *History) taskState() *TaskState { - return s.TaskState -} - -func (s *History) updateTaskState(ts TaskState) { - s.TaskState = &ts -} - -func (s *History) addUsage(u backend.Usage, estimated bool) { - s.TotalInputTokens += u.InputTokens - s.TotalCachedInputTokens += u.CachedInputTokens - s.TotalCacheCreationTokens += u.CacheCreationTokens - s.TotalOutputTokens += u.OutputTokens - s.TotalRequests++ - s.turnInputTokens += u.InputTokens - s.turnCachedTokens += u.CachedInputTokens - s.turnCreationTokens += u.CacheCreationTokens - s.turnOutputTokens += u.OutputTokens - s.turnCostUSD += u.CostUSD - if estimated { - s.Estimated = true - } -} - -func (s *History) recomputeReactionCounts() { - s.PositiveReactions = 0 - s.NegativeReactions = 0 - for _, r := range s.Reactions { - switch r.Category { - case "positive", "excellent": - s.PositiveReactions++ - case "negative", "terrible": - s.NegativeReactions++ - } - } -} - -func (s *History) resetTurnAccumulators() { - s.turnInputTokens = 0 - s.turnCachedTokens = 0 - s.turnCreationTokens = 0 - s.turnOutputTokens = 0 - s.turnCostUSD = 0 -} - -// recordTurnCost computes and stores the last turn's cost, adding it to the -// session total. model is used for the pricing-table fallback when the backend -// did not report cost directly. -func (s *History) recordTurnCost(model string) { - var cost float64 - if s.turnCostUSD > 0 { - cost = s.turnCostUSD - } else { - cost = computeCostUSD(model, backend.Usage{ - InputTokens: s.turnInputTokens, - CachedInputTokens: s.turnCachedTokens, - CacheCreationTokens: s.turnCreationTokens, - OutputTokens: s.turnOutputTokens, - }) - } - s.LastTurnCostUSD = cost - s.SessionCostUSD += cost -} - -func (s *History) update(assistant backend.Message, results []toolResult) { - s.messages = append(s.messages, assistant) - for _, r := range results { - s.messages = append(s.messages, backend.Message{ - Role: "tool", - Content: r.Content, - ContentBlocks: r.ContentBlocks, - ToolCallID: r.ToolCallID, - }) - } -} - -// llmSummarizeToolResult uses the LLM to produce a concise summary of a tool result. -func llmSummarizeToolResult(ctx context.Context, b backend.Backend, name, content string) string { - prompt := fmt.Sprintf("Summarize this tool result from %q in 1-3 sentences, preserving key facts, file paths, values, and findings:\n\n%s", name, content) - ch, err := b.ChatStream(ctx, []backend.Message{{Role: "user", Content: prompt}}, nil, backend.GenerationParams{}) - if err != nil { - return content - } - var sb strings.Builder - for ev := range ch { - if ev.Content != "" { - sb.WriteString(ev.Content) - } - if ev.Done { - break - } - } - if s := strings.TrimSpace(sb.String()); s != "" { - return fmt.Sprintf("[summary of %s result]: %s", name, s) - } - return content -} - -// removeCancelledToolResults filters out tool results that were cancelled due -// to interrupt, keeping completed work. Also removes the corresponding tool -// calls from assistant messages to maintain a valid message sequence. -func (s *History) removeCancelledToolResults() { - // First pass: collect cancelled tool call IDs - cancelled := make(map[string]bool) - for _, m := range s.messages { - if m.Role == "tool" && isCancelledToolResult(m.Content) { - cancelled[m.ToolCallID] = true - } - } - if len(cancelled) == 0 { - return - } - - // Second pass: filter messages and prune tool calls from assistant messages - filtered := s.messages[:0] - for _, m := range s.messages { - if m.Role == "tool" && cancelled[m.ToolCallID] { - continue - } - if m.Role == "assistant" && len(m.ToolCalls) > 0 { - // Remove cancelled tool calls from this assistant message - kept := m.ToolCalls[:0] - for _, tc := range m.ToolCalls { - if !cancelled[tc.ID] { - kept = append(kept, tc) - } - } - if len(kept) == 0 && m.Content == "" { - // All tool calls cancelled and no text content - skip message - continue - } - m.ToolCalls = kept - } - filtered = append(filtered, m) - } - s.messages = filtered -} - -func isCancelledToolResult(content string) bool { - return strings.Contains(content, `"status":"cancelled"`) || - strings.Contains(content, "tool execution interrupted by user") -} - -// PreCompactionSnapshot returns a copy of the current messages for persistence -// before compaction. Call this before compact(). -func (s *History) PreCompactionSnapshot() []backend.Message { - return slices.Clone(s.messages) -} - -// Compact summarizes the conversation via an LLM call, replacing the history -// with system messages + preserved user messages + a structured summary. -// Returns (n compacted, summary text, error); n==0 means nothing to compact. -func (s *History) compact(ctx context.Context, b backend.Backend) (int, string, error) { - if len(s.messages) <= hotTailSize+warmIndexSize { - return 0, "", nil - } - - // Flatten tool calls into plain text so the compaction request - // doesn't need tool schemas. - flattened := flattenToolMessages(s.messages) - flattened = append(flattened, backend.Message{ - Role: "user", - Content: compactionPrompt, - }) - - ch, err := b.ChatStream(ctx, flattened, nil, backend.GenerationParams{}) - if err != nil { - return 0, "", fmt.Errorf("compact: %w", err) - } - - var summary strings.Builder - for ev := range ch { - if ev.Content != "" { - summary.WriteString(ev.Content) - } - if ev.Done { - break - } - } - - summaryText := strings.TrimSpace(summary.String()) - if summaryText == "" { - return 0, "", fmt.Errorf("compact: empty summary") - } - - // Parse structured task state from the model's JSON response. - var ts TaskState - raw := extractJSON(summaryText) - if err := json.Unmarshal([]byte(raw), &ts); err != nil { - // Fallback: treat as unstructured summary if JSON parse fails. - ts = TaskState{Objective: summaryText} - } - s.TaskState = &ts - - beforeCount := len(s.messages) - s.messages = buildCompactedHistory(ts, s.messages) - return beforeCount - len(s.messages), summaryText, nil -} - -// buildCompactedHistory constructs a three-zone post-compaction message list: -// - Cold: structured task state summary (everything older than warm+hot) -// - Warm: brief index of recent decisions/actions (warmIndexSize messages) -// - Hot: last hotTailSize messages verbatim -func buildCompactedHistory(ts TaskState, allMessages []backend.Message) []backend.Message { - total := len(allMessages) - - // Determine zone boundaries. - hotStart := total - hotTailSize - if hotStart < 0 { - hotStart = 0 - } - - // If hotStart lands on a tool message, walk backward to include the - // preceding assistant message with tool_calls. Without this, the hot - // zone starts with an orphaned tool result which violates the OpenAI - // API constraint that tool messages must follow an assistant message - // containing the corresponding tool_calls. - for hotStart > 0 && allMessages[hotStart].Role == "tool" { - hotStart-- - } - - warmStart := hotStart - warmIndexSize - if warmStart < 0 { - warmStart = 0 - } - - // Cold zone: structured task state. - stateJSON, _ := json.Marshal(ts) - cold := backend.Message{ - Role: "user", - Content: "[compacted context — cold zone]\n```json\n" + string(stateJSON) + "\n```", - } - - var out []backend.Message - out = append(out, cold) - - // Warm zone: one-line summaries of decisions in the warm window. - warmSlice := allMessages[warmStart:hotStart] - if len(warmSlice) > 0 { - var sb strings.Builder - sb.WriteString("[warm zone — recent decisions]\n") - for _, m := range warmSlice { - line := summarizeMessage(m) - if line != "" { - sb.WriteString("- ") - sb.WriteString(line) - sb.WriteByte('\n') - } - } - out = append(out, backend.Message{Role: "user", Content: sb.String()}) - } - - // Hot zone: verbatim recent messages. - out = append(out, allMessages[hotStart:]...) - return out -} - -// summarizeMessage produces a one-line summary of a message for the warm index. -func summarizeMessage(m backend.Message) string { - switch m.Role { - case "assistant": - if len(m.ToolCalls) > 0 { - names := make([]string, len(m.ToolCalls)) - for i, tc := range m.ToolCalls { - names[i] = tc.Name - } - return "called: " + strings.Join(names, ", ") - } - text := m.Content - if len(text) > 120 { - text = text[:120] + "…" - } - return "said: " + text - case "user": - text := m.Content - if len(text) > 120 { - text = text[:120] + "…" - } - return "user: " + text - case "tool": - text := m.Content - if len(text) > 80 { - text = text[:80] + "…" - } - return "result(" + m.ToolCallID + "): " + text - } - return "" -} - -// extractJSON finds the first JSON object in s (handling markdown fences). -func extractJSON(s string) string { - // Strip markdown code fence if present. - if i := strings.Index(s, "```json"); i >= 0 { - s = s[i+7:] - if j := strings.Index(s, "```"); j >= 0 { - s = s[:j] - } - } else if i := strings.Index(s, "```"); i >= 0 { - s = s[i+3:] - if j := strings.Index(s, "```"); j >= 0 { - s = s[:j] - } - } - // Find first { ... } - start := strings.Index(s, "{") - if start < 0 { - return s - } - end := strings.LastIndex(s, "}") - if end < start { - return s - } - return s[start : end+1] -} - -// flattenToolMessages converts tool call/result sequences into plain text -// so the compaction request doesn't include tool-specific structures that -// the API may reject when no tools are defined. -func flattenToolMessages(messages []backend.Message) []backend.Message { - out := make([]backend.Message, 0, len(messages)) - for _, m := range messages { - switch { - case m.Role == "assistant" && len(m.ToolCalls) > 0: - var sb strings.Builder - if m.Content != "" { - sb.WriteString(m.Content) - sb.WriteString("\n\n") - } - for _, tc := range m.ToolCalls { - fmt.Fprintf(&sb, "[Tool call: %s(%s)]\n", tc.Name, string(tc.Arguments)) - } - out = append(out, backend.Message{Role: "assistant", Content: sb.String()}) - case m.Role == "tool": - text := m.Content - if len(text) > 4000 { - text = text[:4000] + "..." - } - out = append(out, backend.Message{ - Role: "user", - Content: fmt.Sprintf("[Tool result for %s]:\n%s", m.ToolCallID, text), - }) - default: - out = append(out, m) - } - } - return out -} - -func (s *History) appendUserMessage(content string) { - s.messages = append(s.messages, backend.Message{Role: "user", Content: content}) -} - -// cloneMessages returns a deep copy of the message slice. -func cloneMessages(msgs []backend.Message) []backend.Message { - out := make([]backend.Message, len(msgs)) - for i, m := range msgs { - out[i] = backend.Message{ - Role: m.Role, - Content: m.Content, - ToolCallID: m.ToolCallID, - } - if len(m.ToolCalls) > 0 { - out[i].ToolCalls = make([]backend.ToolCall, len(m.ToolCalls)) - for j, tc := range m.ToolCalls { - out[i].ToolCalls[j] = backend.ToolCall{ - ID: tc.ID, - Name: tc.Name, - Arguments: append(json.RawMessage(nil), tc.Arguments...), - } - } - } - } - return out -} - -// estimateTokens returns a rough token count (~4 chars per token). -func (s *History) estimateTokens() int { - chars := 0 - for _, m := range s.messages { - chars += len(m.Content) - for _, tc := range m.ToolCalls { - chars += len(tc.Name) + len(tc.Arguments) - } - } - return chars / 4 -} - -// stripColdResults summarizes large tool-result messages outside the hot tail -// using an LLM call. Messages in the last hotTailSize slots are left verbatim. -func (s *History) stripColdResults(ctx context.Context, b backend.Backend) { - hot := len(s.messages) - hotTailSize - for i := range s.messages { - if i >= hot { - break - } - m := &s.messages[i] - if m.Role == "tool" && len(m.Content) > 200 { - m.Content = llmSummarizeToolResult(ctx, b, m.ToolCallID, m.Content) - } - } -} - -// stripCold implements the state interface. -func (s *History) stripCold(ctx context.Context, b backend.Backend) { - s.stripColdResults(ctx, b) -} - -// contextDebug returns a multi-line breakdown of the history. -func (s *History) contextDebug() string { - var sb strings.Builder - sb.WriteString(fmt.Sprintf("=== %d messages ===\n", len(s.messages))) - for i, m := range s.messages { - preview := m.Content - if len(preview) > 80 { - preview = preview[:80] + "..." - } - chars := len(m.Content) - for _, tc := range m.ToolCalls { - chars += len(tc.Name) + len(tc.Arguments) - } - sb.WriteString(fmt.Sprintf(" [%d] role=%-10s chars=%-6d %q\n", i, m.Role, chars, preview)) - } - return sb.String() +// DetachedInfo describes a detached process for external consumers. +type DetachedInfo struct { + PID int + Command string + Started int64 // unix timestamp + Exited bool + ExitCode int } diff --git a/session/types.go b/session/types.go deleted file mode 100644 index a18178c..0000000 --- a/session/types.go +++ /dev/null @@ -1,187 +0,0 @@ -package session - -import ( - "context" - "errors" - - "github.com/simonfxr/pubsub" - "ollie/backend" -) - -// WatchField names supported by Core.WaitChange. -const ( - WatchState = "state" - WatchUsage = "usage" - WatchCtxSz = "ctxsz" - WatchCWD = "cwd" - WatchAgent = "agent" -) - -// ErrInterrupted is returned when the user cancels an agent turn (Ctrl-C). -var ErrInterrupted = errors.New("interrupted") - -// Event is a typed output event emitted during an agent turn or in response -// to a command. -type Event struct { - Role string - Name string - Content string - ResponseID string -} - -// EventHandler receives events from the agent. -type EventHandler func(Event) - -// Core is the interface between a frontend (TUI, HTTP handler, etc.) and the -// agent engine. All output from the agent is delivered via the event bus. -type Core interface { - // Submit processes one line of user input. Slash commands and shell - // shortcuts are dispatched synchronously; any other input starts an agent - // turn that publishes events to the bus until the turn is complete. - // After the turn, any queued prompts are drained sequentially. - Submit(ctx context.Context, input string) - - // Interrupt cancels the current in-progress agent turn. - // Returns true if an action was running and was cancelled. - Interrupt(cause error) bool - - // Inject sends a message that will be appended to the next tool result - // as a user interruption. If no turn is running, it is silently dropped. - Inject(prompt string) - - // Queue pushes a prompt onto the FIFO for execution after the current - // turn completes. - Queue(prompt string) - - // Bus returns the session event bus. - Bus() *pubsub.Bus - - // PopQueue removes and returns the next queued prompt. - // Returns ("", false) if the queue is empty. - PopQueue() (string, bool) - - // IsRunning returns true if an agent turn is currently in progress. - IsRunning() bool - - // State returns the current agent state: "idle", "thinking", or "calling: ". - State() string - - // Reply returns the assistant text from the most recently completed turn. - // Cleared when a new prompt is submitted. - Reply() string - - // AgentName returns the name of the active agent. - AgentName() string - - // BackendName returns the name of the active backend (e.g. "anthropic", "ollama"). - BackendName() string - - // ModelName returns the name of the active model. - ModelName() string - - // CtxSz returns the estimated context size as a one-line summary. - CtxSz() string - - // Usage returns billed token counts as a one-line summary. - Usage() string - - // Cost returns per-turn and session cost as a two-line key=value summary. - Cost() string - - // ListModels returns available model names, one per line. - ListModels() string - - // CWD returns the current working directory used for tool execution. - CWD() string - - // SetCWD changes the working directory for tool execution and - // updates the system prompt. Returns an error if the path does not exist. - SetCWD(dir string) error - - // SetSessionID renames the session: updates the in-memory ID, renames - // persisted files on disk, and propagates to the execute server env. - SetSessionID(newID string) error - - // Context returns the current message history as it would be sent to the - // backend: system prompt prepended, stale reads pruned. Does not include - // tool definitions (see Tools) or generation params (see GenerationParams). - Context() []backend.Message - - // SystemPrompt returns the fully rendered system prompt for this session. - SystemPrompt() string - - // GenerationParams returns the current sampling parameters. - GenerationParams() backend.GenerationParams - - // SetGenerationParams replaces the current sampling parameters. - // Returns an error if the agent is currently running. - SetGenerationParams(params backend.GenerationParams) error - - // CompactionModel returns the model override used for context compaction. - CompactionModel() string - - // SetCompactionModel changes the model used for context compaction. - SetCompactionModel(model string) - - // SetEnv injects a session-scoped environment variable into the shell - // subprocesses. Does not affect the daemon process environment. - SetEnv(key, value string) - - // WaitChange blocks until the named field changes from current, then returns - // the new value. Returns ("", false) if ctx is cancelled before a change. - // Supported fields: WatchState, WatchUsage, WatchCtxSz, WatchCWD. - WaitChange(ctx context.Context, field, current string) (string, bool) - - // ToolCallCount returns the total number of tool calls executed in this - // session since the agent was created. The counter is monotonically - // increasing and never resets. Blocked calls (pre-tool hook exit 2) are - // not counted. Timed-out or erroring calls are counted because execution - // was attempted. Use modulo arithmetic in hooks to fire every N calls: - // [ $(($(cat tcct) % 10)) -eq 0 ] && ... - ToolCallCount() int64 - - // SaveSession writes the current session state to the given path. - // The file includes all messages, agent/backend/model metadata, and task state. - SaveSession(path string) error - - // Close releases resources associated with the session, including its - // temporary directory under /tmp/ollie/. - Close() - - // Detach detaches the currently running process from the agent. - // The process continues running; returns false if nothing is executing. - Detach() bool - - // ListDetached returns info about all detached processes. - ListDetached() []DetachedInfo - - // SignalDetached sends a signal to a detached process by PID. - SignalDetached(pid, signal int) error - - // GetDetachedOutput returns the ring buffer output for a detached process. - GetDetachedOutput(pid int) (string, error) - - // DismissDetached removes an exited process from the list. - DismissDetached(pid int) bool - - // InjectSystemEvent appends a system-originated message to the session - // context and emits it on the event bus. The agent sees it on its next - // turn; it appears in the chat log immediately. - InjectSystemEvent(content string) - - // React records an emoji reaction to the most recent assistant response. - React(emoji string) - // ReactTo records an emoji reaction to a specific assistant response. - ReactTo(responseID, emoji string) error - // Reactions returns the current response ID to emoji mapping. - Reactions() map[string]string -} - -// DetachedInfo describes a detached process for external consumers. -type DetachedInfo struct { - PID int - Command string - Started int64 // unix timestamp - Exited bool - ExitCode int -}