diff --git a/session/agent.go b/session/agent.go index 6e5b01f..02872c0 100644 --- a/session/agent.go +++ b/session/agent.go @@ -30,6 +30,11 @@ type Agent struct { // Execution state — owned by the agent, protected by stateMu. state string // "idle", "thinking", "calling: " 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() +} diff --git a/session/commands.go b/session/commands.go index cd75347..5e8cb1e 100644 --- a/session/commands.go +++ b/session/commands.go @@ -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 { diff --git a/session/core_test.go b/session/core_test.go index 4cd5489..6edea14 100644 --- a/session/core_test.go +++ b/session/core_test.go @@ -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") diff --git a/session/session.go b/session/session.go index ae43969..673cd13 100644 --- a/session/session.go +++ b/session/session.go @@ -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 }