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/virtfs" "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() ([]byte, error) { return []byte(strings.Join(backend.Backends(), "\n") + "\n"), nil } var helpFn func() string // helpSpec stores the spec for help generation. Set during tree build. var helpSpec *virtfs.FsNodeDecl func readHelp() ([]byte, error) { if helpFn == nil { return []byte("help not initialized\n"), nil } return []byte(helpFn()), nil } func init() { helpFn = func() string { helpOnce.Do(func() { if helpSpec != nil { helpText = virtfs.GenerateHelp(*helpSpec) } }) return helpText } } var ( helpOnce sync.Once helpText string ) func readModels(mc *ModelCache) ([]byte, error) { if mc != nil { return mc.Get(), nil } return []byte("(no model cache)\n"), nil } func readAgents() ([]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() ([]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 rootCtlHandler(mc *ModelCache, invalidate func(), shutdown func()) func([]byte) ([]byte, error) { handlers := map[string]func([]string) ([]byte, error){ "invalidate": func(_ []string) ([]byte, error) { if mc != nil { mc.Invalidate() } if invalidate != nil { invalidate() } return []byte("ok\n"), nil }, "kill": func(_ []string) ([]byte, error) { if shutdown != nil { shutdown() } return []byte("ok\n"), nil }, } return func(data []byte) ([]byte, error) { return dispatch(handlers, data) } } func readEventwait() ([]byte, error) { return nil, fmt.Errorf("eventwait: use blocking read") } func blockEventwait(cctx context.Context, _ string) ([]byte, string, error) { return session.WaitEvent(cctx) } func requestGenerate(ctx context.Context, 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() ([]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(b *bypass.Broker) ([]byte, error) { if b == nil { return nil, errNoBypass } p := b.GlobalPolicy().Global() return p.Marshal() } func writeBypassPolicy(b *bypass.Broker, data []byte) error { if b == nil { return errNoBypass } var p bypass.Policy if err := bypass.ParsePolicy(data, &p); err != nil { return err } return b.GlobalPolicy().SetGlobal(p) } func readBypassRequest(req *bypass.Request) ([]byte, error) { return []byte(req.Summary() + "\n"), nil } func writeBypassRequest(b *bypass.Broker, req *bypass.Request, data []byte) error { 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 !b.Resolve(req.ID, res) { return fmt.Errorf("request already resolved or timed out") } return nil }