diff --git a/agent/session.go b/agent/session.go deleted file mode 100644 index ca1c23f..0000000 --- a/agent/session.go +++ /dev/null @@ -1,696 +0,0 @@ -package agent - -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/agent/agent.go b/session/agent.go similarity index 98% rename from agent/agent.go rename to session/agent.go index 2cda683..3c49e71 100644 --- a/agent/agent.go +++ b/session/agent.go @@ -1,4 +1,4 @@ -package agent +package session import ( "sync" diff --git a/agent/commands.go b/session/commands.go similarity index 97% rename from agent/commands.go rename to session/commands.go index e7322bb..be8a6de 100644 --- a/agent/commands.go +++ b/session/commands.go @@ -1,4 +1,4 @@ -package agent +package session import ( "context" @@ -175,9 +175,9 @@ func (a *harness) handleCommand(ctx context.Context, input string) bool { return } d := a.r.newDispatcher() - env := []string{"OLLIE_SESSION_ID=" + a.sess.ID(), "OLLIE_UNAME=" + a.sess.Uname()} + env := []string{"OLLIE_SESSION_ID=" + a.id, "OLLIE_UNAME=" + a.uname} env = append(env, a.r.promptEnvExtra...) - rt := BuildRuntime(cfg, d, a.sess.CWD(), env, a.r.baseLayers...) + rt := BuildRuntime(cfg, d, a.cwd, env, a.r.baseLayers...) if rt.CfgBackend != "" { newBe, err := a.r.newBackend(rt.CfgBackend) if err != nil { @@ -226,8 +226,8 @@ func (a *harness) handleCommand(ctx context.Context, input string) bool { a.emit(infoEvent("nothing to compact")) return } - if a.sessionsDir != "" && a.sess.ID() != "" { - histPath := a.activeSessionPath(a.sess.ID(), ".compaction.jsonl") + if a.sessionsDir != "" && a.id != "" { + histPath := a.activeSessionPath(a.id, ".compaction.jsonl") if err := os.MkdirAll(filepath.Dir(histPath), 0700); err != nil { a.emit(infoEvent("compaction history save: " + err.Error())) } else if data, err := json.Marshal(snapshot); err == nil { @@ -351,7 +351,7 @@ func (a *harness) handleCommand(ctx context.Context, input string) bool { continue } marker := " " - if file.id == a.sess.ID() { + if file.id == a.id { marker = "* " } goal := "" diff --git a/agent/compaction.go b/session/compaction.go similarity index 98% rename from agent/compaction.go rename to session/compaction.go index 06a8654..836e251 100644 --- a/agent/compaction.go +++ b/session/compaction.go @@ -1,4 +1,4 @@ -package agent +package session import "os" diff --git a/agent/compaction_test.go b/session/compaction_test.go similarity index 99% rename from agent/compaction_test.go rename to session/compaction_test.go index 51c5216..70f40ef 100644 --- a/agent/compaction_test.go +++ b/session/compaction_test.go @@ -1,4 +1,4 @@ -package agent +package session import ( "encoding/json" diff --git a/agent/config.go b/session/config.go similarity index 99% rename from agent/config.go rename to session/config.go index 4315c1b..b71823f 100644 --- a/agent/config.go +++ b/session/config.go @@ -1,4 +1,4 @@ -package agent +package session import ( "encoding/json" diff --git a/agent/config_test.go b/session/config_test.go similarity index 99% rename from agent/config_test.go rename to session/config_test.go index 4ca7d3a..81cea54 100644 --- a/agent/config_test.go +++ b/session/config_test.go @@ -1,4 +1,4 @@ -package agent +package session import ( "strings" diff --git a/agent/core_test.go b/session/core_test.go similarity index 99% rename from agent/core_test.go rename to session/core_test.go index b4a9d28..c2c55ca 100644 --- a/agent/core_test.go +++ b/session/core_test.go @@ -1,4 +1,4 @@ -package agent +package session import ( "context" @@ -603,13 +603,13 @@ func TestCommand_Clear(t *testing.T) { if c.r.history == nil { t.Fatal("session nil after first turn") } - oldID := c.sess.ID() + oldID := c.id collectEvents(context.Background(), c, "/clear") if c.r.history != nil { t.Error("session not nil after /clear") } - if c.sess.ID() != oldID { - t.Errorf("sessionID changed after /clear: %q → %q; want unchanged", oldID, c.sess.ID()) + if c.id != oldID { + t.Errorf("sessionID changed after /clear: %q → %q; want unchanged", oldID, c.id) } } @@ -1758,13 +1758,13 @@ func TestCore_Usage_WithSession(t *testing.T) { func TestSetSessionID_Rename(t *testing.T) { c := newCore(t, nil, nil) collectEvents(context.Background(), c, "hello") // saves session file - oldID := c.sess.ID() + oldID := c.id newID := NewSessionID() if err := c.SetSessionID(newID); err != nil { t.Fatalf("SetSessionID: %v", err) } - if c.sess.ID() != newID { - t.Errorf("sessionID = %q; want %q", c.sess.ID(), newID) + if c.id != newID { + t.Errorf("sessionID = %q; want %q", c.id, newID) } if _, err := os.Stat(filepath.Join(c.sessionsDir, "active", oldID+".json")); !os.IsNotExist(err) { t.Errorf("old active session file still exists after rename; err=%v", err) @@ -1803,12 +1803,12 @@ func TestSetSessionID_UpdatesPreamble(t *testing.T) { func TestSetSessionID_SameID(t *testing.T) { c := newCore(t, nil, nil) - id := c.sess.ID() + id := c.id if err := c.SetSessionID(id); err != nil { t.Fatalf("SetSessionID with same ID: %v", err) } - if c.sess.ID() != id { - t.Errorf("sessionID changed: got %q; want %q", c.sess.ID(), id) + if c.id != id { + t.Errorf("sessionID changed: got %q; want %q", c.id, id) } } @@ -2312,7 +2312,9 @@ func TestSetEnv_StoredInCore(t *testing.T) { c.SetEnv("FOO", "bar") c.SetEnv("BAZ", "qux") - env := c.sess.Env() + c.envMu.RLock() + env := c.env + c.envMu.RUnlock() if env["FOO"] != "bar" { t.Errorf("env[FOO] = %q; want bar", env["FOO"]) } @@ -2326,7 +2328,9 @@ func TestSetEnv_NilDispatcher_NoPanic(t *testing.T) { c.r.runtime.Dispatcher = nil // Must not panic. c.SetEnv("K", "V") - env := c.sess.Env() + c.envMu.RLock() + env := c.env + c.envMu.RUnlock() if env["K"] != "V" { t.Errorf("env[K] = %q; want V", env["K"]) } diff --git a/agent/cost.go b/session/cost.go similarity index 99% rename from agent/cost.go rename to session/cost.go index d0686ac..301e9d0 100644 --- a/agent/cost.go +++ b/session/cost.go @@ -1,4 +1,4 @@ -package agent +package session import ( "strings" diff --git a/agent/cost_test.go b/session/cost_test.go similarity index 99% rename from agent/cost_test.go rename to session/cost_test.go index e87f5ba..3e7f6e6 100644 --- a/agent/cost_test.go +++ b/session/cost_test.go @@ -1,4 +1,4 @@ -package agent +package session import ( "math" diff --git a/agent/fifo_test.go b/session/fifo_test.go similarity index 78% rename from agent/fifo_test.go rename to session/fifo_test.go index 827996e..e9d5fdb 100644 --- a/agent/fifo_test.go +++ b/session/fifo_test.go @@ -1,12 +1,12 @@ -package agent +package session import ( "sync" "testing" ) -func TestPromptFIFO_BasicOrder(t *testing.T) { - var f PromptFIFO +func TestFifo_BasicOrder(t *testing.T) { + var f Fifo f.Push("a") f.Push("b") f.Push("c") @@ -26,15 +26,15 @@ func TestPromptFIFO_BasicOrder(t *testing.T) { } } -func TestPromptFIFO_EmptyPop(t *testing.T) { - var f PromptFIFO +func TestFifo_EmptyPop(t *testing.T) { + var f Fifo if _, ok := f.Pop(); ok { t.Error("Pop() on zero-value FIFO returned ok=true") } } -func TestPromptFIFO_ConcurrentPushPop(t *testing.T) { - var f PromptFIFO +func TestFifo_ConcurrentPushPop(t *testing.T) { + var f Fifo const n = 100 var wg sync.WaitGroup for i := 0; i < n; i++ { diff --git a/agent/harness.go b/session/harness.go similarity index 95% rename from agent/harness.go rename to session/harness.go index 8a64658..a0badea 100644 --- a/agent/harness.go +++ b/session/harness.go @@ -1,4 +1,4 @@ -package agent +package session import ( "context" @@ -22,7 +22,6 @@ import ( "ollie/backend" olog "ollie/log" "ollie/paths" - "ollie/session" "ollie/tools" ) @@ -310,7 +309,21 @@ type AgentCoreConfig struct { // harness is the Core implementation. It owns all harness and session state // but has no knowledge of how output is rendered. type harness struct { - sess *session.Session + // Session-level state + id string + uname string + cwd string + state string // "idle", "thinking", "calling: " + reply string + bus *pubsub.Bus + fifo Fifo + envMu sync.RWMutex + env map[string]string + plan []byte + prevPrompt string + peers map[string]bool + + // Agent r *Agent log *olog.Logger sessionsDir string @@ -343,7 +356,9 @@ func (a *harness) ToolCallCount() int64 { // SetEnv stores a session-scoped variable and propagates it to the execute server. func (a *harness) SetEnv(key, value string) { - a.sess.SetEnv(key, value) + a.envMu.Lock() + a.env[key] = value + a.envMu.Unlock() if a.r.runtime == nil || a.r.runtime.Dispatcher == nil { return } @@ -356,14 +371,14 @@ func (a *harness) SetEnv(key, value string) { // pushSessionEnv injects OLLIE_SESSION_ID into the execute server subprocess env. func (a *harness) pushSessionEnv() { - if a.r.runtime == nil || a.r.runtime.Dispatcher == nil || a.sess.ID() == "" { + if a.r.runtime == nil || a.r.runtime.Dispatcher == nil || a.id == "" { return } if srv, ok := a.r.runtime.Dispatcher.GetServer("execute"); ok { if es, ok := srv.(tools.EnvSetter); ok { - es.SetEnv("OLLIE_SESSION_ID", a.sess.ID()) - if a.sess.Uname() != "" { - es.SetEnv("OLLIE_UNAME", a.sess.Uname()) + es.SetEnv("OLLIE_SESSION_ID", a.id) + if a.uname != "" { + es.SetEnv("OLLIE_UNAME", a.uname) } } } @@ -371,7 +386,7 @@ func (a *harness) pushSessionEnv() { // pushLockDir sets the flock directory on the execute server to the session tmpdir. func (a *harness) pushLockDir() { - if a.r.runtime == nil || a.r.runtime.Dispatcher == nil || a.sess.ID() == "" { + if a.r.runtime == nil || a.r.runtime.Dispatcher == nil || a.id == "" { return } @@ -454,11 +469,12 @@ func NewAgentCore(cfg AgentCoreConfig) Core { } a := &harness{ - sess: session.New(session.Config{ - ID: cfg.SessionID, - Uname: cfg.Uname, - CWD: paths.ExpandHome(cfg.CWD), - }), + id: cfg.SessionID, + uname: cfg.Uname, + cwd: paths.ExpandHome(cfg.CWD), + state: "idle", + bus: pubsub.NewBus(), + env: make(map[string]string), r: &Agent{ history: cfg.History, runtime: rt, @@ -482,7 +498,7 @@ func NewAgentCore(cfg AgentCoreConfig) Core { hookCtx, cancel := context.WithTimeout(context.Background(), time.Duration(hookTimeout)*time.Second) defer cancel() return a.r.runtime.Hooks.Run(hookCtx, HookTurnError, map[string]string{ - "session_id": a.sess.ID(), + "session_id": a.id, "cwd": a.CWD(), "model": a.r.runtime.Backend.Model(), "error_type": errType, @@ -496,7 +512,7 @@ func NewAgentCore(cfg AgentCoreConfig) Core { // Close releases resources for this session, including its tmpdir. func (a *harness) Close() { - a.log.Debug("Close() session=%q", a.sess.ID()) + a.log.Debug("Close() session=%q", a.id) a.flushSave() if a.r.runtime != nil && a.r.runtime.Dispatcher != nil { if srv, ok := a.r.runtime.Dispatcher.GetServer("execute"); ok { @@ -506,8 +522,8 @@ func (a *harness) Close() { } } } - if a.sess.ID() != "" { - os.RemoveAll(filepath.Join(ollieTmpDir(), a.sess.ID())) //nolint:errcheck + if a.id != "" { + os.RemoveAll(filepath.Join(ollieTmpDir(), a.id)) //nolint:errcheck } } @@ -704,7 +720,7 @@ func (a *harness) ModelName() string { } func (a *harness) State() string { - return a.sess.State() + return a.state } func (a *harness) notifyChange() { @@ -714,7 +730,7 @@ func (a *harness) notifyChange() { } func (a *harness) setState(state string) { - a.sess.SetState(state) + a.state = state a.log.Debug("state -> %q", state) a.notifyChange() } @@ -760,16 +776,16 @@ func (a *harness) WaitChange(ctx context.Context, field, current string) (string func (a *harness) Reply() string { - r := a.sess.Reply() + r := a.reply a.log.Debug("Reply() len=%d", len(r)) return r } // CWD returns the current working directory for tool execution. func (a *harness) CWD() string { - if a.sess.CWD() != "" { - a.log.Debug("CWD() = %q", a.sess.CWD()) - return a.sess.CWD() + if a.cwd != "" { + a.log.Debug("CWD() = %q", a.cwd) + return a.cwd } wd, _ := os.Getwd() a.log.Debug("CWD() = %q (from getwd)", wd) @@ -786,8 +802,8 @@ func (a *harness) SetCWD(dir string) error { return fmt.Errorf("cwd: %w", err) } } - oldCwd := a.sess.CWD() - a.sess.SetCWD(dir) + oldCwd := a.cwd + a.cwd = dir // Update cwd references in the system prompt. if oldCwd != "" && dir != "" && oldCwd != dir { a.r.runtime.Preamble = strings.ReplaceAll(a.r.runtime.Preamble, oldCwd, dir) @@ -807,8 +823,8 @@ func (a *harness) SetCWD(dir string) error { // SetSessionID renames the session. It updates the in-memory ID, renames // persisted files on disk, and propagates to the execute server env. func (a *harness) SetSessionID(newID string) error { - a.log.Debug("SetSessionID(%q) old=%q", newID, a.sess.ID()) - oldID := a.sess.ID() + a.log.Debug("SetSessionID(%q) old=%q", newID, a.id) + oldID := a.id if oldID == newID { return nil } @@ -823,7 +839,7 @@ func (a *harness) SetSessionID(newID string) error { } } } - a.sess.SetID(newID) + a.id = newID // Update session ID references in the system prompt. a.r.runtime.Preamble = strings.ReplaceAll(a.r.runtime.Preamble, oldID, newID) // Rename tmpdir so isread markers remain valid after rename. @@ -865,7 +881,7 @@ func (a *harness) autoWarnLimit(ctx context.Context) int { // agent-specific prompt with any agentSpawn hook output. func (a *harness) spawnContext(ctx context.Context) string { result := a.r.runtime.Hooks.Run(ctx, HookAgentSpawn, map[string]string{ - "session_id": a.sess.ID(), + "session_id": a.id, "agent": a.r.agentName, "cwd": a.CWD(), "model": a.r.runtime.Backend.Model(), @@ -887,7 +903,7 @@ func (a *harness) spawnContext(ctx context.Context) string { // re-injection, post-hook. Returns (n compacted, error). Returns (0, nil) if // the pre-hook blocked or there was nothing to compact. Caller manages setState. func (a *harness) runCompact(ctx context.Context, trigger string) (int, error) { - payload := map[string]string{"session_id": a.sess.ID(), "trigger": trigger, "cwd": a.CWD()} + payload := map[string]string{"session_id": a.id, "trigger": trigger, "cwd": a.CWD()} pre := a.r.runtime.Hooks.Run(ctx, HookPreCompact, payload, a.log) if pre.Warning != "" { a.emit(infoEvent(pre.Warning)) @@ -914,7 +930,7 @@ func (a *harness) runCompact(ctx context.Context, trigger string) (int, error) { return 0, err } if n > 0 { - a.auditLog.Debug("compact: removed %d messages trigger=%s session=%s", n, trigger, a.sess.ID()) + a.auditLog.Debug("compact: removed %d messages trigger=%s session=%s", n, trigger, a.id) a.r.warnedContext = false if sc := a.spawnContext(ctx); sc != "" { a.r.history.appendUserMessage(sc) @@ -959,15 +975,15 @@ func (a *harness) flushSave() { if !dirty { return } - if a.r.history == nil || a.sess.ID() == "" || a.sessionsDir == "" { + if a.r.history == nil || a.id == "" || a.sessionsDir == "" { return } - path := a.activeSessionPath(a.sess.ID(), ".json") + path := a.activeSessionPath(a.id, ".json") if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil { a.log.Error("session save: %v", err) return } - if err := a.r.history.saveToFull(path, a.sess.ID(), a.r.agentName, + if err := a.r.history.saveToFull(path, a.id, a.r.agentName, a.r.runtime.Backend.Name(), a.r.runtime.Backend.Model(), a.CWD(), a.remote); err != nil { a.log.Error("session save: %v", err) } @@ -981,7 +997,7 @@ func (a *harness) SaveSession(path string) error { if a.r.history == nil { return fmt.Errorf("no active session") } - return a.r.history.saveToFull(path, a.sess.ID(), a.r.agentName, + return a.r.history.saveToFull(path, a.id, a.r.agentName, a.r.runtime.Backend.Name(), a.r.runtime.Backend.Model(), a.CWD(), a.remote) } @@ -1007,7 +1023,7 @@ func (a *harness) Inject(prompt string) { // If an inject is already pending, fall back to the normal FIFO so nothing // is lost. Use CompareAndSwap to avoid a race between the nil check and store. if !a.pendingInject.CompareAndSwap(nil, &prompt) { - a.sess.Queue(prompt) + a.fifo.Push(prompt) return } a.emit(Event{Role: "info", Content: "\n"}) @@ -1021,26 +1037,26 @@ func (a *harness) injectRewrite(prompt string) { } func (a *harness) Queue(prompt string) { - a.sess.Queue(prompt) - a.sess.Bus().Publish("queued", prompt) + a.fifo.Push(prompt) + a.bus.Publish("queued", prompt) } func (a *harness) drainQueue() { - if prompt, ok := a.sess.PopQueue(); ok { + if prompt, ok := a.fifo.Pop(); ok { a.Submit(context.Background(), prompt) } } func (a *harness) Bus() *pubsub.Bus { - return a.sess.Bus() + return a.bus } func (a *harness) emit(ev Event) { - a.sess.Bus().Publish("event", ev) + a.bus.Publish("event", ev) } func (a *harness) PopQueue() (string, bool) { - return a.sess.PopQueue() + return a.fifo.Pop() } func (a *harness) IsRunning() bool { @@ -1189,7 +1205,7 @@ func (a *harness) Submit(ctx context.Context, input string) { if a.handleCommand(ctx, input) { return } - a.sess.Queue(input) + a.fifo.Push(input) return } @@ -1202,7 +1218,7 @@ func (a *harness) Submit(ctx context.Context, input string) { return } if a.IsRunning() { - a.sess.Queue(input) + a.fifo.Push(input) return } @@ -1217,7 +1233,7 @@ func (a *harness) executeTurn(ctx context.Context, input string) string { a.emit(Event{Role: "user", Content: input}) hookResult := a.r.runtime.Hooks.Run(ctx, HookPreTurn, map[string]string{ - "session_id": a.sess.ID(), + "session_id": a.id, "cwd": a.CWD(), "prompt": input, }, a.log) @@ -1263,7 +1279,7 @@ func (a *harness) executeTurn(ctx context.Context, input string) string { a.r.currentAction.Store(handle) a.setState("thinking") - a.auditLog.Debug("turn: start input=%s session=%s", auditTruncate(input), a.sess.ID()) + a.auditLog.Debug("turn: start input=%s session=%s", auditTruncate(input), a.id) // Build per-turn agentConfig from the current runtime. a.r.cfg = agentConfig{ @@ -1325,7 +1341,7 @@ func (a *harness) executeTurn(ctx context.Context, input string) string { } a.r.cfg.PreTool = func(ctx context.Context, name string, args json.RawMessage) HookResult { return a.r.runtime.Hooks.Run(ctx, HookPreTool, map[string]string{ - "session_id": a.sess.ID(), + "session_id": a.id, "cwd": a.CWD(), "tool": name, "args": string(args), @@ -1333,7 +1349,7 @@ func (a *harness) executeTurn(ctx context.Context, input string) string { } a.r.cfg.PostTool = func(ctx context.Context, name string, args json.RawMessage, result string) HookResult { return a.r.runtime.Hooks.Run(ctx, HookPostTool, map[string]string{ - "session_id": a.sess.ID(), + "session_id": a.id, "cwd": a.CWD(), "tool": name, "args": string(args), @@ -1442,7 +1458,7 @@ func (a *harness) executeTurn(ctx context.Context, input string) string { } a.mu.Lock() - a.sess.SetReply(replyBuf.String()) + a.reply = replyBuf.String() a.mu.Unlock() replyBuf.Reset() a.setState("idle") @@ -1455,20 +1471,20 @@ func (a *harness) executeTurn(ctx context.Context, input string) string { a.r.history.removeCancelledToolResults() } if errors.Is(err, context.Canceled) || errors.Is(err, ErrInterrupted) { - a.auditLog.Debug("turn: interrupted session=%s", a.sess.ID()) + a.auditLog.Debug("turn: interrupted session=%s", a.id) a.saveSession() return "" } a.emit(Event{Role: "error", Content: err.Error()}) // Drain one FIFO item — the turnError hook may have queued a recovery prompt. - if next, ok := a.sess.PopQueue(); ok { + if next, ok := a.fifo.Pop(); ok { return next } return "" } stopResult := a.r.runtime.Hooks.Run(ctx, HookPostTurn, map[string]string{ - "session_id": a.sess.ID(), + "session_id": a.id, "cwd": a.CWD(), }, a.log) if stopResult.Warning != "" { @@ -1483,12 +1499,12 @@ func (a *harness) executeTurn(ctx context.Context, input string) string { if a.r.history != nil { a.r.history.recordTurnCost(a.r.cfg.Backend.Model()) - appendUsageLog(a.sess.ID(), a.r.cfg.Backend.Name(), a.r.cfg.Backend.Model(), a.r.history) + appendUsageLog(a.id, a.r.cfg.Backend.Name(), a.r.cfg.Backend.Model(), a.r.history) if a.r.history.LastTurnCostUSD > 0 { a.emit(Event{Role: "info", Content: fmt.Sprintf("costLast=$%.4f\n", a.r.history.LastTurnCostUSD)}) } a.auditLog.Debug("turn: end reply=%s cost=$%.4f session_total=$%.4f session=%s", - auditTruncate(a.sess.Reply()), a.r.history.LastTurnCostUSD, a.r.history.SessionCostUSD, a.sess.ID()) + auditTruncate(a.reply), a.r.history.LastTurnCostUSD, a.r.history.SessionCostUSD, a.id) a.notifyChange() } a.saveSession() @@ -1505,7 +1521,7 @@ func (a *harness) executeTurn(ctx context.Context, input string) string { } // Drain one item from the FIFO; the outer loop handles the rest. - if next, ok := a.sess.PopQueue(); ok { + if next, ok := a.fifo.Pop(); ok { return next } diff --git a/agent/hooks.go b/session/hooks.go similarity index 99% rename from agent/hooks.go rename to session/hooks.go index 26e374a..45771a9 100644 --- a/agent/hooks.go +++ b/session/hooks.go @@ -1,4 +1,4 @@ -package agent +package session import ( "bytes" diff --git a/agent/loop.go b/session/loop.go similarity index 99% rename from agent/loop.go rename to session/loop.go index d642c6f..06610be 100644 --- a/agent/loop.go +++ b/session/loop.go @@ -1,4 +1,4 @@ -package agent +package session import ( "context" diff --git a/agent/loop_cache_test.go b/session/loop_cache_test.go similarity index 99% rename from agent/loop_cache_test.go rename to session/loop_cache_test.go index 8719f56..915c7cf 100644 --- a/agent/loop_cache_test.go +++ b/session/loop_cache_test.go @@ -1,4 +1,4 @@ -package agent +package session import ( "context" diff --git a/agent/loop_errlimit_test.go b/session/loop_errlimit_test.go similarity index 99% rename from agent/loop_errlimit_test.go rename to session/loop_errlimit_test.go index 9975388..bfcbe4e 100644 --- a/agent/loop_errlimit_test.go +++ b/session/loop_errlimit_test.go @@ -1,4 +1,4 @@ -package agent +package session import ( "context" diff --git a/agent/loop_maxsteps_helpers_test.go b/session/loop_maxsteps_helpers_test.go similarity index 99% rename from agent/loop_maxsteps_helpers_test.go rename to session/loop_maxsteps_helpers_test.go index 90afee5..aa5aea1 100644 --- a/agent/loop_maxsteps_helpers_test.go +++ b/session/loop_maxsteps_helpers_test.go @@ -1,4 +1,4 @@ -package agent +package session import ( "context" diff --git a/agent/loop_maxsteps_test.go b/session/loop_maxsteps_test.go similarity index 99% rename from agent/loop_maxsteps_test.go rename to session/loop_maxsteps_test.go index feb518e..b471048 100644 --- a/agent/loop_maxsteps_test.go +++ b/session/loop_maxsteps_test.go @@ -1,4 +1,4 @@ -package agent +package session import ( "context" diff --git a/agent/loop_parallel_test.go b/session/loop_parallel_test.go similarity index 99% rename from agent/loop_parallel_test.go rename to session/loop_parallel_test.go index 2b9bab9..a62a4c6 100644 --- a/agent/loop_parallel_test.go +++ b/session/loop_parallel_test.go @@ -1,4 +1,4 @@ -package agent +package session import ( "context" diff --git a/agent/loop_truncate_test.go b/session/loop_truncate_test.go similarity index 99% rename from agent/loop_truncate_test.go rename to session/loop_truncate_test.go index 60f8a3e..74403fe 100644 --- a/agent/loop_truncate_test.go +++ b/session/loop_truncate_test.go @@ -1,4 +1,4 @@ -package agent +package session import ( "context" diff --git a/agent/loop_turnerror_test.go b/session/loop_turnerror_test.go similarity index 99% rename from agent/loop_turnerror_test.go rename to session/loop_turnerror_test.go index c411f74..d61fbe2 100644 --- a/agent/loop_turnerror_test.go +++ b/session/loop_turnerror_test.go @@ -1,4 +1,4 @@ -package agent +package session import ( "context" diff --git a/agent/prompt_resolver.go b/session/prompt_resolver.go similarity index 99% rename from agent/prompt_resolver.go rename to session/prompt_resolver.go index dab0643..e851adf 100644 --- a/agent/prompt_resolver.go +++ b/session/prompt_resolver.go @@ -1,4 +1,4 @@ -package agent +package session import ( "bytes" diff --git a/agent/runtime.go b/session/runtime.go similarity index 98% rename from agent/runtime.go rename to session/runtime.go index 5074a05..6c83c4f 100644 --- a/agent/runtime.go +++ b/session/runtime.go @@ -1,4 +1,4 @@ -package agent +package session import ( "encoding/json" diff --git a/session/session.go b/session/session.go index 1c43ed5..8bd99f7 100644 --- a/session/session.go +++ b/session/session.go @@ -1,225 +1,696 @@ -// Package session defines the runtime environment for an agent. -// A Session holds identity, working directory, environment, state machine, -// event bus, and prompt queue — everything that persists across agent swaps. package session import ( - "sync" + "context" + "encoding/json" + "fmt" + "os" + "slices" + "strings" + "time" - "github.com/simonfxr/pubsub" + "ollie/backend" ) -// Session is the runtime environment in which an agent operates. -// It outlives any particular agent configuration and maintains -// identity, state, and communication channels. -type Session struct { - mu sync.RWMutex - id string - uname string // immutable user principal - cwd string - state string // "idle", "thinking", "calling: " - reply string // assistant text from last completed turn +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: - bus *pubsub.Bus - fifo Fifo - - envMu sync.RWMutex - env map[string]string - - plan []byte - prevPrompt string - peers map[string]bool // peer session IDs (bidirectional) - - changeMu sync.Mutex - changeCond *sync.Cond +{ + "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" } -// Config holds parameters for creating a new Session. -type Config struct { - ID string - Uname string - CWD string +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"` } -// New creates a Session with the given configuration. -func New(cfg Config) *Session { - s := &Session{ - id: cfg.ID, - uname: cfg.Uname, - cwd: cfg.CWD, - state: "idle", - bus: pubsub.NewBus(), - env: make(map[string]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"` +} + +// 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 + } } - s.changeCond = sync.NewCond(&s.changeMu) return s } -// ID returns the session identifier. -func (s *Session) ID() string { return s.id } - -// Uname returns the immutable user principal. -func (s *Session) Uname() string { return s.uname } - -// CWD returns the current working directory. -func (s *Session) CWD() string { - s.mu.RLock() - defer s.mu.RUnlock() - return s.cwd +// 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 } -// SetCWD updates the working directory. -func (s *Session) SetCWD(dir string) { - s.mu.Lock() - s.cwd = dir - s.mu.Unlock() - s.notifyChange() +// 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} } -// State returns the current session state. -func (s *Session) State() string { - s.mu.RLock() - defer s.mu.RUnlock() - return s.state +// 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 } -// SetState transitions to a new state and notifies waiters. -func (s *Session) SetState(state string) { - s.mu.Lock() - s.state = state - s.mu.Unlock() - s.notifyChange() -} - -// Reply returns the assistant text from the last completed turn. -func (s *Session) Reply() string { - s.mu.RLock() - defer s.mu.RUnlock() - return s.reply -} - -// SetReply stores the assistant reply text. -func (s *Session) SetReply(text string) { - s.mu.Lock() - s.reply = text - s.mu.Unlock() -} - -// Bus returns the session event bus. -func (s *Session) Bus() *pubsub.Bus { return s.bus } - -// SetEnv sets a session-scoped environment variable. -func (s *Session) SetEnv(key, value string) { - s.envMu.Lock() - s.env[key] = value - s.envMu.Unlock() -} - -// Env returns a copy of all session environment variables. -func (s *Session) Env() map[string]string { - s.envMu.RLock() - defer s.envMu.RUnlock() - out := make(map[string]string, len(s.env)) - for k, v := range s.env { - out[k] = v +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 } -// SetID renames the session. -func (s *Session) SetID(id string) { - s.mu.Lock() - s.id = id - s.mu.Unlock() +func (s *History) taskState() *TaskState { + return s.TaskState } -// Queue pushes a prompt onto the FIFO. -func (s *Session) Queue(prompt string) { s.fifo.Push(prompt) } +func (s *History) updateTaskState(ts TaskState) { + s.TaskState = &ts +} -// PopQueue removes and returns the next queued prompt. -func (s *Session) PopQueue() (string, bool) { return s.fifo.Pop() } +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 + } +} -// WaitChange blocks until the state changes from current, then returns the new state. -// Returns ("", false) if ctx-based cancellation would be needed (not implemented here). -func (s *Session) WaitChange(current string) string { - s.changeMu.Lock() - defer s.changeMu.Unlock() - for { - s.mu.RLock() - now := s.state - s.mu.RUnlock() - if now != current { - return now +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++ } - s.changeCond.Wait() } } -func (s *Session) notifyChange() { - s.changeMu.Lock() - s.changeCond.Broadcast() - s.changeMu.Unlock() +func (s *History) resetTurnAccumulators() { + s.turnInputTokens = 0 + s.turnCachedTokens = 0 + s.turnCreationTokens = 0 + s.turnOutputTokens = 0 + s.turnCostUSD = 0 } -// Plan returns the session plan. -func (s *Session) Plan() []byte { - s.mu.RLock() - defer s.mu.RUnlock() - return s.plan -} - -// SetPlan updates the session plan. -func (s *Session) SetPlan(p []byte) { - s.mu.Lock() - s.plan = p - s.mu.Unlock() -} - -// PrevPrompt returns the last submitted prompt. -func (s *Session) PrevPrompt() string { - s.mu.RLock() - defer s.mu.RUnlock() - return s.prevPrompt -} - -// SetPrevPrompt stores the last submitted prompt. -func (s *Session) SetPrevPrompt(p string) { - s.mu.Lock() - s.prevPrompt = p - s.mu.Unlock() -} - -// HasPeer reports whether peerID is linked to this session. -func (s *Session) HasPeer(id string) bool { - s.mu.RLock() - defer s.mu.RUnlock() - return s.peers[id] -} - -// AddPeer adds a bidirectional peer link. -func (s *Session) AddPeer(id string) { - s.mu.Lock() - if s.peers == nil { - s.peers = make(map[string]bool) +// 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.peers[id] = true - s.mu.Unlock() + s.LastTurnCostUSD = cost + s.SessionCostUSD += cost } -// RemovePeer removes a peer link. -func (s *Session) RemovePeer(id string) { - s.mu.Lock() - delete(s.peers, id) - s.mu.Unlock() +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, + }) + } } -// Peers returns a copy of all peer IDs. -func (s *Session) Peers() []string { - s.mu.RLock() - defer s.mu.RUnlock() - out := make([]string, 0, len(s.peers)) - for id := range s.peers { - out = append(out, id) +// 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/agent/state.go b/session/state.go similarity index 99% rename from agent/state.go rename to session/state.go index 3e9ce93..92623f1 100644 --- a/agent/state.go +++ b/session/state.go @@ -1,4 +1,4 @@ -package agent +package session import ( "context" diff --git a/agent/types.go b/session/types.go similarity index 93% rename from agent/types.go rename to session/types.go index c70dc79..a18178c 100644 --- a/agent/types.go +++ b/session/types.go @@ -1,9 +1,8 @@ -package agent +package session import ( "context" "errors" - "sync" "github.com/simonfxr/pubsub" "ollie/backend" @@ -186,26 +185,3 @@ type DetachedInfo struct { Exited bool ExitCode int } - -// PromptFIFO is a simple thread-safe FIFO for queued prompts. -type PromptFIFO struct { - mu sync.Mutex - items []string -} - -func (f *PromptFIFO) Push(s string) { - f.mu.Lock() - f.items = append(f.items, s) - f.mu.Unlock() -} - -func (f *PromptFIFO) Pop() (string, bool) { - f.mu.Lock() - defer f.mu.Unlock() - if len(f.items) == 0 { - return "", false - } - s := f.items[0] - f.items = f.items[1:] - return s, true -} diff --git a/agent/usage_log.go b/session/usage_log.go similarity index 99% rename from agent/usage_log.go rename to session/usage_log.go index b9e9cbc..9f262b4 100644 --- a/agent/usage_log.go +++ b/session/usage_log.go @@ -1,4 +1,4 @@ -package agent +package session import ( "encoding/json" diff --git a/agent/waitchange_test.go b/session/waitchange_test.go similarity index 93% rename from agent/waitchange_test.go rename to session/waitchange_test.go index 98c2175..fe24f66 100644 --- a/agent/waitchange_test.go +++ b/session/waitchange_test.go @@ -1,23 +1,25 @@ -package agent +package session import ( "context" "io" "sync" + + "github.com/simonfxr/pubsub" "testing" "time" olog "ollie/log" - "ollie/session" ) -// newTestCore returns a minimal agent with session wired up. +// newTestCore returns a minimal harness for testing. func newTestCore(initialState string) *harness { - sess := session.New(session.Config{ID: "test"}) - sess.SetState(initialState) a := &harness{ - sess: sess, - log: olog.NewWriter("test", olog.LevelError+1, io.Discard, io.Discard), + id: "test", + state: initialState, + bus: pubsub.NewBus(), + env: make(map[string]string), + log: olog.NewWriter("test", olog.LevelError+1, io.Discard, io.Discard), } a.changeCond = sync.NewCond(&a.changeMu) return a