lkn
/
ollie-core
Archived
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

rename agent/ → session/, absorb session.Session into harness 

The agent/ package is now session/ — because a session is the
top-level concept. The package contains:
- harness: implements Core, the session orchestrator
- Agent: the reasoning entity (history, runtime, tools)
- History: conversation accumulator
- Fifo, loop, compaction, hooks, commands

The old session/ package (which just held extracted fields) is
dissolved back into the harness. One package, clean ownership.
This commit is contained in:
Levi Neely 2026-07-29 19:13:26 +02:00
parent 761dd51e24
commit 6ee949bd7c
28 changed files with 784 additions and 1011 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

696
agent/session.go
View File

@ -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()
}

2
agent/agent.go → session/agent.go
View File

@ -1,4 +1,4 @@
package agent
package session
import (
"sync"

12
agent/commands.go → session/commands.go
View File

@ -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 := ""

2
agent/compaction.go → session/compaction.go
View File

@ -1,4 +1,4 @@
package agent
package session
import "os"

2
agent/compaction_test.go → session/compaction_test.go
View File

@ -1,4 +1,4 @@
package agent
package session
import (
"encoding/json"

2
agent/config.go → session/config.go
View File

@ -1,4 +1,4 @@
package agent
package session
import (
"encoding/json"

2
agent/config_test.go → session/config_test.go
View File

@ -1,4 +1,4 @@
package agent
package session
import (
"strings"

28
agent/core_test.go → session/core_test.go
View File

@ -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"])
}

2
agent/cost.go → session/cost.go
View File

@ -1,4 +1,4 @@
package agent
package session
import (
"strings"

2
agent/cost_test.go → session/cost_test.go
View File

@ -1,4 +1,4 @@
package agent
package session
import (
"math"

14
agent/fifo_test.go → session/fifo_test.go
View File

@ -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++ {

128
agent/harness.go → session/harness.go
View File

@ -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: <tool>"
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
}

2
agent/hooks.go → session/hooks.go
View File

@ -1,4 +1,4 @@
package agent
package session
import (
"bytes"

2
agent/loop.go → session/loop.go
View File

@ -1,4 +1,4 @@
package agent
package session
import (
"context"

2
agent/loop_cache_test.go → session/loop_cache_test.go
View File

@ -1,4 +1,4 @@
package agent
package session
import (
"context"

2
agent/loop_errlimit_test.go → session/loop_errlimit_test.go
View File

@ -1,4 +1,4 @@
package agent
package session
import (
"context"

2
agent/loop_maxsteps_helpers_test.go → session/loop_maxsteps_helpers_test.go
View File

@ -1,4 +1,4 @@
package agent
package session
import (
"context"

2
agent/loop_maxsteps_test.go → session/loop_maxsteps_test.go
View File

@ -1,4 +1,4 @@
package agent
package session
import (
"context"

2
agent/loop_parallel_test.go → session/loop_parallel_test.go
View File

@ -1,4 +1,4 @@
package agent
package session
import (
"context"

2
agent/loop_truncate_test.go → session/loop_truncate_test.go
View File

@ -1,4 +1,4 @@
package agent
package session
import (
"context"

2
agent/loop_turnerror_test.go → session/loop_turnerror_test.go
View File

@ -1,4 +1,4 @@
package agent
package session
import (
"context"

2
agent/prompt_resolver.go → session/prompt_resolver.go
View File

@ -1,4 +1,4 @@
package agent
package session
import (
"bytes"

2
agent/runtime.go → session/runtime.go
View File

@ -1,4 +1,4 @@
package agent
package session
import (
"encoding/json"

835
session/session.go
View File

@ -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: <tool>"
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()
}

2
agent/state.go → session/state.go
View File

@ -1,4 +1,4 @@
package agent
package session
import (
"context"

26
agent/types.go → session/types.go
View File

@ -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
}

2
agent/usage_log.go → session/usage_log.go
View File

@ -1,4 +1,4 @@
package agent
package session
import (
"encoding/json"

16
agent/waitchange_test.go → session/waitchange_test.go
View File

@ -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