package fs import ( "context" "encoding/json" "fmt" "os" "strings" "sync" "ollie/cmd/olliesrv/internal/agent" "ollie/cmd/olliesrv/internal/backend" "ollie/cmd/olliesrv/internal/bypass" "ollie/fsedsl" "ollie/cmd/olliesrv/internal/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() } } // wireAgentEvents connects an agent's output and state change events. // Called once per agent: at creation (handlers_agent.go) and on restore (newroot.go). // // Contract: // - SetOutput → events flow to the AgentLog (chat stream, tool blocks, etc.) // - SetOnStateChange → state transitions publish to the event bus func wireAgentEvents(sessID string, ag *agent.Agent, al *AgentLog) { ag.SetOutput(NewEventHandler(al)) 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 requestRootCtl(ctx HandlerCtx, data []byte) ([]byte, error) { return rdwrDispatch(rootCtlHandlers)(ctx, data) } var rootCtlHandlers = map[string]rdwrHandler{ "invalidate": func(ctx HandlerCtx, _ []string) ([]byte, error) { if ctx.Models != nil { ctx.Models.Invalidate() } if ctx.Invalidate != nil { ctx.Invalidate() } return []byte("ok\n"), nil }, "kill": func(ctx HandlerCtx, _ []string) ([]byte, error) { if ctx.Shutdown != nil { ctx.Shutdown() } return []byte("ok\n"), nil }, } 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 }