session: move cwd, fifo, submitMu, toolCallCount, pendingInject to Agent

Agent now owns all execution-related state:
- cwd: the directory the agent operates in
- fifo: prompt queue (prompts target a specific agent)
- submitMu: serializes turns for this agent
- toolCallCount: per-agent tool usage metric
- pendingInject: injection targeting the current agent

Session retains only environment state: identity, bus, env, plan, peers,
persistence. This completes the agent/session state separation.
This commit is contained in:
Levi Neely 2026-07-29 20:15:13 +02:00
parent 8acadaaa30
commit 7592f810a9
4 changed files with 50 additions and 35 deletions

View File

@ -30,6 +30,11 @@ type Agent struct {
// Execution state — owned by the agent, protected by stateMu.
state string // "idle", "thinking", "calling: <tool>"
reply string // last assistant response
cwd string // working directory for tool execution
fifo Fifo // prompt queue
toolCallCount atomic.Int64
pendingInject atomic.Pointer[string]
submitMu sync.Mutex // serializes Submit calls (commands + turns)
stateMu sync.RWMutex
changeMu sync.Mutex
changeCond *sync.Cond
@ -141,3 +146,18 @@ func (ag *Agent) WaitChange(ctx context.Context, field, current string) (string,
func (ag *Agent) InitCond() {
ag.changeCond = sync.NewCond(&ag.changeMu)
}
// CWD returns the agent's working directory.
func (ag *Agent) Cwd() string {
ag.stateMu.RLock()
c := ag.cwd
ag.stateMu.RUnlock()
return c
}
// SetCWD sets the agent's working directory (no validation — caller must validate).
func (ag *Agent) SetCwd(dir string) {
ag.stateMu.Lock()
ag.cwd = dir
ag.stateMu.Unlock()
}

View File

@ -177,7 +177,7 @@ func (a *Session) handleCommand(ctx context.Context, input string) bool {
d := a.r.newDispatcher()
env := []string{"OLLIE_SESSION_ID=" + a.id, "OLLIE_UNAME=" + a.uname}
env = append(env, a.r.promptEnvExtra...)
rt := BuildRuntime(cfg, d, a.cwd, env, a.r.baseLayers...)
rt := BuildRuntime(cfg, d, a.r.cwd, env, a.r.baseLayers...)
if rt.CfgBackend != "" {
newBe, err := a.r.newBackend(rt.CfgBackend)
if err != nil {

View File

@ -359,7 +359,7 @@ func TestSubmit_PendingInjectAsNextTurn(t *testing.T) {
c := newCore(t, be, nil)
inject := "injected follow-up"
c.pendingInject.Store(&inject)
c.r.pendingInject.Store(&inject)
collectEvents(context.Background(), c, "first")
@ -433,7 +433,7 @@ func TestSubmit_WhileRunning(t *testing.T) {
// Concurrent Submit must not start a new turn — always goes to FIFO.
c.Submit(context.Background(), "concurrent")
if stored := c.pendingInject.Load(); stored != nil {
if stored := c.r.pendingInject.Load(); stored != nil {
t.Errorf("concurrent Submit set pendingInject %q; want FIFO only", *stored)
}
got, inFIFO := c.PopQueue()
@ -683,7 +683,7 @@ func TestCommand_I_SetsInject(t *testing.T) {
}()
waitState(t, c, "thinking")
c.handleCommand(context.Background(), "/i my inject")
p := c.pendingInject.Load()
p := c.r.pendingInject.Load()
if p == nil || *p != "my inject" {
t.Errorf("pendingInject = %v; want %q", p, "my inject")
}
@ -727,9 +727,9 @@ func TestCommand_IRW_Empty(t *testing.T) {
func TestCommand_IRW_OverwritesInject(t *testing.T) {
c := newCore(t, nil, nil)
existing := "old"
c.pendingInject.Store(&existing)
c.r.pendingInject.Store(&existing)
collectEvents(context.Background(), c, "/irw new inject")
p := c.pendingInject.Load()
p := c.r.pendingInject.Load()
if p == nil || *p != "new inject" {
t.Errorf("pendingInject after /irw = %v; want %q", p, "new inject")
}
@ -1593,7 +1593,7 @@ func TestRun_ExecCancelledWithInject(t *testing.T) {
}
c := newCore(t, be, nil)
inject := "user interrupt"
c.pendingInject.Store(&inject)
c.r.pendingInject.Store(&inject)
c.r.runtime.Exec = func(_ context.Context, _ string, _ json.RawMessage) (string, []backend.ContentBlock, error) {
cancel() // ctx cancelled during exec
return "", nil, fmt.Errorf("exec cancelled")

View File

@ -14,7 +14,6 @@ import (
"strconv"
"strings"
"sync"
"sync/atomic"
"syscall"
"time"
@ -312,9 +311,7 @@ type Session struct {
// Session-level state
id string
uname string
cwd string
bus *pubsub.Bus
fifo Fifo
envMu sync.RWMutex
env map[string]string
plan []byte
@ -330,10 +327,7 @@ type Session struct {
turnError func(ctx context.Context, errType, errMsg string) HookResult // overridable for tests
remote string // SSH target for remote execution
startupMessages []string
toolCallCount atomic.Int64
pendingInject atomic.Pointer[string]
mu sync.RWMutex
submitMu sync.Mutex // serializes Submit calls (commands + turns)
auditLog *olog.Logger
// Debounced session persistence
@ -347,7 +341,7 @@ type Session struct {
// session. The counter is monotonically increasing and never resets.
// Blocked calls (pre-tool hook exit 2) are not counted.
func (a *Session) ToolCallCount() int64 {
return a.toolCallCount.Load()
return a.r.toolCallCount.Load()
}
// SetEnv stores a session-scoped variable and propagates it to the execute server.
@ -466,12 +460,12 @@ func New(cfg Config) *Session {
a := &Session{
id: cfg.SessionID,
uname: cfg.Uname,
cwd: paths.ExpandHome(cfg.CWD),
bus: pubsub.NewBus(),
env: make(map[string]string),
r: &Agent{
history: cfg.History,
runtime: rt,
cwd: paths.ExpandHome(cfg.CWD),
agentName: cfg.AgentName,
agentsDir: cfg.AgentsDir,
promptEnvExtra: cfg.PromptEnvExtra,
@ -777,9 +771,10 @@ func (a *Session) Reply() string {
// CWD returns the current working directory for tool execution.
func (a *Session) CWD() string {
if a.cwd != "" {
a.log.Debug("CWD() = %q", a.cwd)
return a.cwd
cwd := a.r.Cwd()
if cwd != "" {
a.log.Debug("CWD() = %q", cwd)
return cwd
}
wd, _ := os.Getwd()
a.log.Debug("CWD() = %q (from getwd)", wd)
@ -796,8 +791,8 @@ func (a *Session) SetCWD(dir string) error {
return fmt.Errorf("cwd: %w", err)
}
}
oldCwd := a.cwd
a.cwd = dir
oldCwd := a.r.Cwd()
a.r.SetCwd(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)
@ -1016,8 +1011,8 @@ func (a *Session) Interrupt(cause error) bool {
func (a *Session) 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.fifo.Push(prompt)
if !a.r.pendingInject.CompareAndSwap(nil, &prompt) {
a.r.fifo.Push(prompt)
return
}
a.emit(Event{Role: "info", Content: "\n"})
@ -1025,18 +1020,18 @@ func (a *Session) Inject(prompt string) {
}
func (a *Session) injectRewrite(prompt string) {
a.pendingInject.Store(&prompt)
a.r.pendingInject.Store(&prompt)
a.emit(Event{Role: "info", Content: "\n"})
a.emit(Event{Role: "user", Content: prompt})
}
func (a *Session) Queue(prompt string) {
a.fifo.Push(prompt)
a.r.fifo.Push(prompt)
a.bus.Publish("queued", prompt)
}
func (a *Session) drainQueue() {
if prompt, ok := a.fifo.Pop(); ok {
if prompt, ok := a.r.fifo.Pop(); ok {
a.Submit(context.Background(), prompt)
}
}
@ -1050,7 +1045,7 @@ func (a *Session) emit(ev Event) {
}
func (a *Session) PopQueue() (string, bool) {
return a.fifo.Pop()
return a.r.fifo.Pop()
}
func (a *Session) IsRunning() bool {
@ -1199,20 +1194,20 @@ func (a *Session) Submit(ctx context.Context, input string) {
if a.handleCommand(ctx, input) {
return
}
a.fifo.Push(input)
a.r.fifo.Push(input)
return
}
// Serialize commands and turns so that e.g. a /compact arriving via
// ctl cannot race with an executeTurn arriving via prompt.
a.submitMu.Lock()
defer a.submitMu.Unlock()
a.r.submitMu.Lock()
defer a.r.submitMu.Unlock()
if a.handleCommand(ctx, input) {
return
}
if a.IsRunning() {
a.fifo.Push(input)
a.r.fifo.Push(input)
return
}
@ -1328,7 +1323,7 @@ func (a *Session) executeTurn(ctx context.Context, input string) string {
a.emit(ev)
}
a.r.cfg.PopInject = func() string {
if p := a.pendingInject.Swap(nil); p != nil {
if p := a.r.pendingInject.Swap(nil); p != nil {
return *p
}
return ""
@ -1351,7 +1346,7 @@ func (a *Session) executeTurn(ctx context.Context, input string) string {
}, a.log)
}
a.r.cfg.IncrToolCallCount = func() int64 {
return a.toolCallCount.Add(1)
return a.r.toolCallCount.Add(1)
}
a.r.cfg.SaveSession = func() { a.saveSession() }
a.r.cfg.ResultCache = &a.r.resultCache
@ -1471,7 +1466,7 @@ func (a *Session) executeTurn(ctx context.Context, input string) string {
}
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.fifo.Pop(); ok {
if next, ok := a.r.fifo.Pop(); ok {
return next
}
return ""
@ -1510,12 +1505,12 @@ func (a *Session) executeTurn(ctx context.Context, input string) string {
// Inject that was pending but never consumed (text-only response with no
// tool calls) — treat it as the next user message.
if p := a.pendingInject.Swap(nil); p != nil {
if p := a.r.pendingInject.Swap(nil); p != nil {
return *p
}
// Drain one item from the FIFO; the outer loop handles the rest.
if next, ok := a.fifo.Pop(); ok {
if next, ok := a.r.fifo.Pop(); ok {
return next
}