package fs import ( "bytes" "context" "encoding/json" "fmt" "os" "strconv" "strings" "sync" "ollie/agent" "ollie/backend" "ollie/bypass" "ollie/fsedsl" "ollie/paths" "ollie/session" "ollie/toolsrv" ) // mergeCtx returns a context that cancels when either parent or child cancels. // parent is typically the session context; child is the per-read timeout from the server. func mergeCtx(parent, child context.Context) (context.Context, context.CancelFunc) { merged, cancel := context.WithCancel(parent) stop := context.AfterFunc(child, func() { cancel() }) return merged, func() { stop(); cancel() } } // wireAgentStateEvents sets up the agent to publish state change events to the event bus. func wireAgentStateEvents(sessID string, ag *agent.Agent) { ag.SetOnStateChange(func(agentID, state string) { session.PublishEvent("session."+sessID+".agent."+agentID+".state", state) }) } func readBackends(_ HandlerCtx) ([]byte, error) { return []byte(strings.Join(backend.Backends(), "\n") + "\n"), nil } var helpFn func() string func readHelp(_ HandlerCtx) ([]byte, error) { if helpFn == nil { return []byte("help not initialized\n"), nil } return []byte(helpFn()), nil } func init() { helpFn = func() string { helpOnce.Do(func() { helpText = fsedsl.GenerateHelp(treeSpec) }) return helpText } } var ( helpOnce sync.Once helpText string ) func readModels(ctx HandlerCtx) ([]byte, error) { if ctx.Models != nil { return ctx.Models.Get(), nil } return []byte("(no model cache)\n"), nil } func readAgents(_ HandlerCtx) ([]byte, error) { var sb strings.Builder for _, dir := range agent.AgentsDirs() { entries, err := os.ReadDir(dir) if err != nil { continue } for _, e := range entries { if !e.IsDir() && strings.HasSuffix(e.Name(), ".json") { sb.WriteString(strings.TrimSuffix(e.Name(), ".json")) sb.WriteByte('\n') } } } return []byte(sb.String()), nil } func readTools(_ HandlerCtx) ([]byte, error) { tools := toolsrv.DiscoverTools() if len(tools) == 0 { return []byte("(no tools found)\n"), nil } var sb strings.Builder for _, ti := range tools { fmt.Fprintf(&sb, "%-24s %s\n", ti.Name, ti.Description) } return []byte(sb.String()), nil } func writeRootCtl(ctx HandlerCtx, data []byte) error { handlers := map[string]ctlHandler{ "invalidate": func(_ []string, _ HandlerCtx) error { if ctx.Models != nil { ctx.Models.Invalidate() } if ctx.Invalidate != nil { ctx.Invalidate() } return nil }, "kill": func(_ []string, _ HandlerCtx) error { if ctx.Shutdown != nil { ctx.Shutdown() } return nil }, } return ctlDispatch(handlers)(ctx, data) } func readEventwait(_ HandlerCtx) ([]byte, error) { return nil, fmt.Errorf("eventwait: use blocking read") } func blockEventwait(_ HandlerCtx, cctx context.Context, _ string) ([]byte, string, error) { return session.WaitEvent(cctx) } func requestGenerate(ctx HandlerCtx, data []byte) ([]byte, error) { var req backend.GenerateRequest if err := json.Unmarshal(data, &req); err != nil { req.Prompt = strings.TrimSpace(string(data)) } result, err := backend.Generate(ctx, req) if err != nil { return nil, err } return []byte(result + "\n"), nil } func readAliases(_ HandlerCtx) ([]byte, error) { var sb strings.Builder for name, sess := range session.Sessions() { sid := sess.ID fmt.Fprintf(&sb, "%s\tsession/%s\n", sid, name) for _, ag := range sess.Agents() { fmt.Fprintf(&sb, "%s\tsession/%s/agent/%s\n", ag.ID(), name, ag.Name()) } } return []byte(sb.String()), nil } var errNoBypass = fmt.Errorf("bypass broker not available") func readBypassPolicy(ctx HandlerCtx) ([]byte, error) { if ctx.Bypass == nil { return nil, errNoBypass } p := ctx.Bypass.GlobalPolicy().Global() return p.Marshal() } func writeBypassPolicy(ctx HandlerCtx, data []byte) error { if ctx.Bypass == nil { return errNoBypass } var p bypass.Policy if err := bypass.ParsePolicy(data, &p); err != nil { return err } return ctx.Bypass.GlobalPolicy().SetGlobal(p) } // bypassBindings returns bindings for pending bypass requests. func bypassBindings(ctx HandlerCtx) ([]Binding, error) { if ctx.Bypass == nil { return nil, errNoBypass } pending := ctx.Bypass.Pending() out := make([]Binding, len(pending)) for i, r := range pending { req := r // capture out[i] = Binding{ Name: r.ID, Applier: func(c HandlerCtx) HandlerCtx { c.Data = req return c }, } } return out, nil } func readBypassRequest(ctx HandlerCtx) ([]byte, error) { req := ctx.Data.(*bypass.Request) return []byte(req.Summary() + "\n"), nil } func writeBypassRequest(ctx HandlerCtx, data []byte) error { req := ctx.Data.(*bypass.Request) cmd := strings.TrimSpace(string(data)) var res bypass.Resolution switch cmd { case "approve": res = bypass.ResolveApprove case "deny": res = bypass.ResolveDeny case "persist": res = bypass.ResolvePersist default: return fmt.Errorf("unknown resolution: %s (use approve/deny/persist)", cmd) } if !ctx.Bypass.Resolve(req.ID, res) { return fmt.Errorf("request already resolved or timed out") } return nil } func readSessionIdx(_ HandlerCtx) ([]byte, error) { return session.BuildIndex(), nil } func readAgentIdx(ctx HandlerCtx) ([]byte, error) { return session.BuildAgentIndex(ctx.Session.Core), nil } func requestSessionNew(_ HandlerCtx, data []byte) ([]byte, error) { args := strings.Fields(string(data)) name := "" remote := "" for _, arg := range args { if k, v, ok := strings.Cut(arg, "="); ok { switch k { case "name": name = v case "remote": remote = v } } } sess, err := session.CreateEmpty(name, remote) if err != nil { return nil, err } return []byte(sess.Name() + "\n"), nil } // sessionBindings returns bindings for all sessions. // The Blueprint pattern clones the structure and binds ctx.Session. func sessionBindings(ctx HandlerCtx) ([]Binding, error) { rs := ctx.RootState sessions := session.Sessions() out := make([]Binding, 0, len(sessions)) for name, sess := range sessions { // Get or create cached SessionNode wrapper node := rs.GetOrCreateNode(sess) n := name // capture out = append(out, Binding{ Name: sess.Name(), Aliases: []string{sess.ID}, UID: sess.Uname, GID: "agent", Applier: func(c HandlerCtx) HandlerCtx { c.Context = node.Ctx() c.Session = node c.Remove = func() { session.Kill(n) rs.RemoveNode(n) } c.Rename = func(newName string) error { if err := session.Rename(n, newName); err != nil { return err } rs.RenameNode(n, newName) return nil } return c }, Remove: func() { session.Kill(n) rs.RemoveNode(n) }, Rename: func(newName string) error { if err := session.Rename(n, newName); err != nil { return err } rs.RenameNode(n, newName) return nil }, }) } return out, nil } func readSessionEnv(ctx HandlerCtx) ([]byte, error) { s := ctx.Session var sb strings.Builder fmt.Fprintf(&sb, "OLLIE_SESSION_ID=%s\n", s.ID()) for _, e := range os.Environ() { if strings.HasPrefix(e, "OLLIE_") && !strings.HasPrefix(e, "OLLIE_SESSION_ID=") { sb.WriteString(e) sb.WriteByte('\n') } } return []byte(sb.String()), nil } func readSessionPaused(ctx HandlerCtx) ([]byte, error) { if ctx.Session.IsPaused() { return []byte("true\n"), nil } return []byte("false\n"), nil } func readSessionConnected(ctx HandlerCtx) ([]byte, error) { if ctx.Session.IsConnected() { return []byte("true\n"), nil } return []byte("false\n"), nil } func writeSessionCtl(ctx HandlerCtx, data []byte) error { handlers := map[string]ctlHandler{ "kill": func(_ []string, _ HandlerCtx) error { ctx.Remove() return nil }, ".": func(_ []string, _ HandlerCtx) error { ctx.Remove() return nil }, "save": func(_ []string, _ HandlerCtx) error { if ctx.Session.Core != nil { ctx.Session.Core.Save() } return nil }, "invalidate": func(_ []string, _ HandlerCtx) error { ctx.Session.InvalidateModelsCache() return nil }, "pause": func(_ []string, _ HandlerCtx) error { return ctx.Session.Pause() }, "resume": func(_ []string, _ HandlerCtx) error { return ctx.Session.Resume() }, } err := ctlDispatch(handlers)(ctx, data) if err != nil { return fmt.Errorf("session ctl: %w", err) } return nil } func readSessionName(ctx HandlerCtx) ([]byte, error) { return []byte(ctx.Session.Name() + "\n"), nil } func writeSessionName(ctx HandlerCtx, data []byte) error { newName := strings.TrimSpace(string(data)) if newName == "" { return nil } oldName := ctx.Session.Name() if newName != oldName { return ctx.Rename(newName) } return nil } func readSessionID(ctx HandlerCtx) ([]byte, error) { return []byte(ctx.Session.ID() + "\n"), nil } func readSessionBypass(ctx HandlerCtx) ([]byte, error) { if ctx.Bypass == nil { return nil, fmt.Errorf("bypass broker not available") } p := ctx.Bypass.SessionPolicy(ctx.Session.ID()) return p.Marshal() } func writeSessionBypass(ctx HandlerCtx, data []byte) error { if ctx.Bypass == nil { return fmt.Errorf("bypass broker not available") } var p bypass.Policy if err := bypass.ParsePolicy(data, &p); err != nil { return err } ctx.Bypass.SetSessionPolicy(ctx.Session.ID(), p) return nil } func requestAgentNew(ctx HandlerCtx, data []byte) ([]byte, error) { args := strings.Fields(string(data)) ag, err := session.CreateAgent(ctx.Session.Name(), args) if err != nil { return nil, fmt.Errorf("create agent: %w", err) } // Create AgentLog for this agent and wire up the event handler al := NewAgentLog("") ctx.Session.SetAgentLog(ag.ID(), al) ag.SetOutput(NewEventHandler(al)) wireAgentStateEvents(ctx.Session.ID(), ag) return []byte(ag.ID() + "\n"), nil } // agentBindings returns bindings for all agents in a session. // The Blueprint pattern clones the structure and binds ctx.Agent + ctx.AgentLog. func agentBindings(ctx HandlerCtx) ([]Binding, error) { sess := ctx.Session if sess == nil || sess.Core == nil { return nil, nil } var out []Binding for _, ag := range sess.Core.Agents() { a := ag // capture al := sess.AgentLogFor(a.ID()) if al == nil { continue } out = append(out, Binding{ Name: a.Name(), Aliases: []string{a.ID()}, UID: a.ID(), GID: "agent", Applier: func(c HandlerCtx) HandlerCtx { c.Agent = a c.AgentLog = al return c }, Rename: func(newName string) error { a.SetName(newName) session.PersistSession(sess.Name()) return nil }, }) } return out, nil } func writeAgentPrompt(ctx HandlerCtx, data []byte) error { input := strings.TrimSpace(string(data)) if input == "" { return nil } if input == "/invalidate" { ctx.Session.InvalidateModelsCache() return nil } go func() { ctx.Agent.Submit(ctx.Session.Ctx(), input) ctx.AgentLog.EnsureTrailingNewline() }() return nil } func writeAgentFifoIn(ctx HandlerCtx, data []byte) error { input := strings.TrimSpace(string(data)) if input == "" { return nil } ctx.Agent.Queue(input) return nil } func readAgentFifoOut(ctx HandlerCtx) ([]byte, error) { item, ok := ctx.Agent.PopQueue() if !ok { return nil, nil } return []byte(item), nil } func statAgentChat() os.FileInfo { // Note: This needs ctx but Stat doesn't receive it. // We'll need to handle this specially or pass via closure. // For now, return a placeholder - the real stat is done in builder. return &SyntheticFileInfo{Name_: "chat", Mode_: 0444, Size_: 64 * 1024} } func readAgentChat(ctx HandlerCtx) ([]byte, error) { ctx.AgentLog.mu.RLock() defer ctx.AgentLog.mu.RUnlock() data := make([]byte, len(ctx.AgentLog.log)) copy(data, ctx.AgentLog.log) return data, nil } func readAgentChatText(ctx HandlerCtx) ([]byte, error) { ctx.AgentLog.mu.RLock() defer ctx.AgentLog.mu.RUnlock() return stripMarkers(ctx.AgentLog.log), nil } func streamAgentChat(ctx HandlerCtx, cctx context.Context, base string) ([]byte, string, error) { merged, cancel := mergeCtx(ctx, cctx) defer cancel() return streamChat(ctx.AgentLog, merged, base) } func streamAgentChatText(ctx HandlerCtx, cctx context.Context, base string) ([]byte, string, error) { merged, cancel := mergeCtx(ctx, cctx) defer cancel() data, nextBase, err := streamChat(ctx.AgentLog, merged, base) if err != nil || len(data) == 0 { return data, nextBase, err } return stripMarkers(data), nextBase, nil } // stripMarkers removes [[[...]]] marker lines and source fence lines from raw // chat data. Partial lines (no trailing \n) are always passed through — they // are streaming content, never markers. func stripMarkers(data []byte) []byte { var out []byte inFence := false for len(data) > 0 { // Find next newline nl := bytes.IndexByte(data, '\n') if nl < 0 { // Partial line at end — always content, emit it out = append(out, data...) break } line := data[:nl+1] // includes \n data = data[nl+1:] // Drop [[[...]]] marker lines trimmed := bytes.TrimRight(line, "\n") if len(trimmed) >= 7 && bytes.HasPrefix(trimmed, []byte("[[[")) && bytes.HasSuffix(trimmed, []byte("]]]")) { continue } // Drop fence open/close lines (``` with optional language suffix) stripped := bytes.TrimSpace(trimmed) if bytes.HasPrefix(stripped, []byte("```")) { if inFence { inFence = false continue } // Opening fence — only drop if it's a source fence (tool output), // not markdown fences in assistant prose. Tool fences follow // immediately after a marker line, so if we just dropped a marker // and this is the next line, it's a source fence. // Simplification: drop all bare ``` lines (no content on the line). // Keep fences that are part of assistant markdown (they have text before/after). if len(stripped) == 3 || (len(stripped) > 3 && stripped[3] != '`') { inFence = true continue } } if !inFence { out = append(out, line...) } } return out } func blockAgentStateWait(ctx HandlerCtx, cctx context.Context, base string) ([]byte, string, error) { merged, cancel := mergeCtx(ctx, cctx) defer cancel() if base == "" { base = ctx.Agent.State() } v, ok := ctx.Agent.WaitChange(merged, agent.WatchState, base) if !ok { st := ctx.Agent.State() return []byte(st + "\n"), st, nil } return []byte(v + "\n"), v, nil } func readAgentLog(ctx HandlerCtx) ([]byte, error) { const maxWindow = 64 * 1024 ctx.AgentLog.mu.RLock() log := ctx.AgentLog.log start := 0 if len(log) > maxWindow { start = len(log) - maxWindow } data := make([]byte, len(log)-start) copy(data, log[start:]) ctx.AgentLog.mu.RUnlock() return data, nil } func readAgentPlan(ctx HandlerCtx) ([]byte, error) { ctx.AgentLog.mu.RLock() data := make([]byte, len(ctx.AgentLog.plan)) copy(data, ctx.AgentLog.plan) ctx.AgentLog.mu.RUnlock() return data, nil } func writeAgentPlan(ctx HandlerCtx, data []byte) error { ctx.AgentLog.mu.Lock() ctx.AgentLog.plan = make([]byte, len(data)) copy(ctx.AgentLog.plan, data) ctx.AgentLog.mu.Unlock() return nil } func readAgentCfg(ctx HandlerCtx) ([]byte, error) { ctx.AgentLog.mu.RLock() defer ctx.AgentLog.mu.RUnlock() p := ctx.Agent.GenParams() var sb strings.Builder fmt.Fprintf(&sb, "name=%s\n", ctx.Session.ID()) fmt.Fprintf(&sb, "backend=%s\n", ctx.Agent.BackendName()) fmt.Fprintf(&sb, "model=%s\n", ctx.Agent.ModelName()) fmt.Fprintf(&sb, "id=%s\n", ctx.Agent.ID()) fmt.Fprintf(&sb, "profile=%s\n", ctx.Agent.Profile()) fmt.Fprintf(&sb, "displayName=%s\n", ctx.Agent.Name()) fmt.Fprintf(&sb, "cwd=%s\n", ctx.Agent.Cwd()) fmt.Fprintf(&sb, "remote=%s\n", ctx.AgentLog.remote) fmt.Fprintf(&sb, "maxTokens=%d\n", p.MaxTokens) if p.Temperature != nil { fmt.Fprintf(&sb, "temperature=%g\n", *p.Temperature) } if p.TopP != nil { fmt.Fprintf(&sb, "topP=%g\n", *p.TopP) } if p.TopK != nil { fmt.Fprintf(&sb, "topK=%d\n", *p.TopK) } if len(p.Stop) > 0 { fmt.Fprintf(&sb, "stop=%s\n", strings.Join(p.Stop, ",")) } return []byte(sb.String()), nil } func writeAgentCfg(ctx HandlerCtx, data []byte) error { input := strings.TrimSpace(string(data)) if input == "" { return nil } parts := strings.SplitN(input, "=", 2) if len(parts) != 2 { return fmt.Errorf("invalid cfg format (expected key=value)") } switch parts[0] { case "name": if strings.TrimSpace(parts[1]) != "" { ctx.Agent.SetName(strings.TrimSpace(parts[1])) } default: return fmt.Errorf("unknown cfg key: %s", parts[0]) } return nil } func writeAgentCtl(ctx HandlerCtx, data []byte) error { handlers := map[string]ctlHandler{ "kill": func(_ []string, _ HandlerCtx) error { if ctx.Session.Core != nil { ctx.Session.Core.RemoveAgent(ctx.Agent.ID()) session.PublishEvent("session."+ctx.Session.ID()+".agent."+ctx.Agent.ID()+".kill", "") } return nil }, "stop": func(_ []string, _ HandlerCtx) error { ctx.Agent.Interrupt(agent.ErrInterrupted) return nil }, "compact": func(_ []string, _ HandlerCtx) error { ctx.Agent.HandleCommand(ctx.Session.Ctx(), "/compact") return nil }, "clear": func(_ []string, _ HandlerCtx) error { ctx.Agent.HandleCommand(ctx.Session.Ctx(), "/clear") return nil }, "model": func(args []string, _ HandlerCtx) error { if len(args) > 0 { if be := ctx.Agent.Backend(); be != nil { be.SetModel(strings.Join(args, " ")) } } return nil }, "name": func(args []string, _ HandlerCtx) error { if len(args) > 0 { ctx.Agent.SetName(strings.Join(args, " ")) } return nil }, "cwd": func(args []string, _ HandlerCtx) error { if len(args) > 0 { ctx.Agent.SetCWD(paths.ExpandHome(strings.Join(args, " "))) } return nil }, } err := ctlDispatch(handlers)(ctx, data) if err != nil { return fmt.Errorf("agent ctl: %w", err) } return nil } func readAgentCwd(ctx HandlerCtx) ([]byte, error) { return []byte(ctx.Agent.Cwd() + "\n"), nil } func writeAgentCwd(ctx HandlerCtx, data []byte) error { input := strings.TrimSpace(string(data)) if input == "" { return nil } input = paths.ExpandHome(input) if _, err := os.Stat(input); err != nil { return fmt.Errorf("cwd: %w", err) } ctx.Agent.SetCWD(input) return nil } func readAgentStats(ctx HandlerCtx) ([]byte, error) { s := "usage=" + ctx.Agent.UsageStr() + "\n" + "cost=" + ctx.Agent.CostStr() + "\n" + "ctxsz=" + ctx.Agent.CtxSz() + "\n" return []byte(s), nil } func readAgentModels(ctx HandlerCtx) ([]byte, error) { return []byte(ctx.Session.CachedListModels() + "\n"), nil } func readAgentSystemprompt(ctx HandlerCtx) ([]byte, error) { return []byte(ctx.Agent.SystemPrompt() + "\n"), nil } func readAgentTools(ctx HandlerCtx) ([]byte, error) { if ctx.Session.ToolsConn() == nil { return []byte("(no tool server)\n"), nil } loaded, err := ctx.Session.ToolsConn().ListTools() if err != nil { return []byte(fmt.Sprintf("(error: %v)\n", err)), nil } var sb strings.Builder for _, ti := range loaded { if ti.Description != "" { fmt.Fprintf(&sb, "%-20s %s\n", ti.Name, ti.Description) } else { sb.WriteString(ti.Name + "\n") } } return []byte(sb.String()), nil } func writeAgentTools(ctx HandlerCtx, data []byte) error { return ctx.Session.LoadTool(string(data), ctx.Agent) } func readAgentName(ctx HandlerCtx) ([]byte, error) { return []byte(ctx.Agent.Name() + "\n"), nil } func writeAgentName(ctx HandlerCtx, data []byte) error { newName := strings.TrimSpace(string(data)) if newName == "" { return nil } ctx.Agent.SetName(newName) session.PersistSession(ctx.Session.Name()) return nil } func readAgentID(ctx HandlerCtx) ([]byte, error) { return []byte(ctx.Agent.ID() + "\n"), nil } func streamChat(al *AgentLog, cctx context.Context, base string) ([]byte, string, error) { ctx, cancel := context.WithCancel(cctx) defer cancel() context.AfterFunc(ctx, func() { al.chatCond.Broadcast() }) var offset int if base != "" { fmt.Sscanf(base, "%d", &offset) } else { al.mu.RLock() offset = len(al.log) al.mu.RUnlock() } al.mu.RLock() for len(al.log) <= offset { if ctx.Err() != nil { al.mu.RUnlock() return nil, "", nil } al.chatCond.Wait() } data := make([]byte, len(al.log)-offset) copy(data, al.log[offset:]) newOffset := len(al.log) al.mu.RUnlock() return data, fmt.Sprintf("%d", newOffset), nil } // processBindings returns bindings for detached processes. func processBindings(ctx HandlerCtx) ([]Binding, error) { if ctx.Agent == nil { return nil, nil } procs := ctx.Agent.ListDetached() out := make([]Binding, len(procs)) for i, p := range procs { pid := p.PID // capture out[i] = Binding{ Name: strconv.Itoa(p.PID), Applier: func(c HandlerCtx) HandlerCtx { c.Data = pid return c }, } } return out, nil } func readProcess(ctx HandlerCtx) ([]byte, error) { pid := ctx.Data.(int) output, err := ctx.Agent.GetDetachedOutput(pid) if err != nil { return nil, err } return []byte(output), nil } func removeProcess(ctx HandlerCtx) error { pid := ctx.Data.(int) if !ctx.Agent.DismissDetached(pid) { return fmt.Errorf("process %d not found or still running", pid) } return nil } func nopRead(_ HandlerCtx) ([]byte, error) { return nil, nil }